作为一家 AI 应用公司的技术负责人，我最近被老板问到一个灵魂拷问：上个月我们到底在各个模型上花了多少钱？每个部门用了多少？有没有超出预算？当我打开账单看到一堆美元数字时，突然意识到——如果按照官方汇率（人民币兑美元约 7.3:1）计算，光是 GPT-4.1 的输出费用就要 $8/MTok × 100万token = $800 ≈ ¥5,840，Claude Sonnet 4.5 更贵——$15/MTok × 100万token = $1,500 ≈ ¥10,950。
但当我把目光转向 HolySheep 中转站时，发现他们按 ¥1=$1 无损结算（官方汇率 ¥7.3=$1），这意味着我可以直接省下 85%+ 的成本。Gemini 2.5 Flash 更是低至 $2.50/MTok，DeepSeek V3.2 只有 $0.42/MTok，简直是成本优化的神器。
为什么你的 AI 账单总是超支？
我见过太多团队出现这种情况：每个开发者都在调用 API，但没有人知道钱花到哪里去了。直到月底账单来了才发现——测试环境跑了 thousands of tokens，生产环境的 prompt 太长，或者某个项目偷偷用了最贵的模型。
本文将手把手教你：
- 用 Python 实现 Token 用量追踪系统
- 按部门和项目维度拆分成本
- 设置实时预算告警
- 用 HolySheep API 统一管理所有模型调用
Token 成本对比：官方 vs HolySheep
| 模型 | 官方价格 (output/MTok) | 官方折算 (¥/MTok) | HolySheep 价格 (¥/MTok) | 节省比例 |
|---|---|---|---|---|
| GPT-4.1 | $8.00 | ¥58.40 | ¥8.00 | 86.3% |
| Claude Sonnet 4.5 | $15.00 | ¥109.50 | ¥15.00 | 86.3% |
| Gemini 2.5 Flash | $2.50 | ¥18.25 | ¥2.50 | 86.3% |
| DeepSeek V3.2 | $0.42 | ¥3.07 | ¥0.42 | 86.3% |
以每月 100 万 output tokens 计算：
- GPT-4.1：官方 ¥5,840 → HolySheep ¥800，节省 ¥5,040
- Claude Sonnet 4.5：官方 ¥10,950 → HolySheep ¥1,500，节省 ¥9,450
- DeepSeek V3.2：官方 ¥307 → HolySheep ¥42，节省 ¥265
如果你每月在 AI 模型上花费超过 ¥5,000，使用 HolySheep 一年就能省下一台 MacBook Pro。
为什么选 HolySheep
我在实际项目中使用 HolySheep 三个月后，发现它不只是便宜：
- 国内直连延迟 <50ms：之前用官方 API，从北京到美国东海岸要 200-300ms，现在走 HolySheep 国内节点，延迟稳定在 30-50ms
- 微信/支付宝充值：再也不用折腾信用卡和外币账户
- 汇率无损：¥1=$1，账单清晰，不用担心汇率波动
- 统一接口：OpenAI、Claude、Gemini、DeepSeek 一个平台全搞定
- 免费额度：注册即送免费 token，新手友好
适合谁与不适合谁
| 场景 | 推荐指数 | 原因 |
|---|---|---|
| 企业 AI 应用研发团队 | ⭐⭐⭐⭐⭐ | 成本敏感，需按部门核算 |
| 独立开发者/SaaS 产品 | ⭐⭐⭐⭐⭐ | 成本控制优先，预算有限 |
| AI 培训机构/教育场景 | ⭐⭐⭐⭐ | 用量大，调用频繁 |
| 大型企业（已有专属协议） | ⭐⭐ | 官方可能有定制折扣 |
| 偶尔测试/学习用途 | ⭐⭐ | 免费额度足够 |
价格与回本测算
假设你的团队构成如下：
- 研发部门：每月 500 万 tokens（混合模型）
- 产品部门：每月 300 万 tokens（主要是 Claude）
- 运营部门：每月 200 万 tokens（主要是 Gemini Flash）
月度总成本对比（按模型配比估算）：
| 部门 | 官方月度成本 | HolySheep 月度成本 | 月节省 | 年节省 |
|---|---|---|---|---|
| 研发部（混合） | ¥29,200 | ¥4,000 | ¥25,200 | ¥302,400 |
| 产品部（Claude） | ¥32,850 | ¥4,500 | ¥28,350 | ¥340,200 |
| 运营部（Gemini） | ¥3,650 | ¥500 | ¥3,150 | ¥37,800 |
| 总计 | ¥65,700 | ¥9,000 | ¥56,700 | ¥680,400 |
使用 HolySheep 后，年节省约 68 万元，这还没算上国内直连带来的开发效率提升和 API 稳定性溢价。
Token 用量追踪系统实现
1. 统一 API 封装层
首先，我封装了一个统一的 API 调用类，支持 HolySheep 的所有主流模型：
import requests
import time
import json
from datetime import datetime
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class TokenUsage:
"""Token 使用量记录"""
timestamp: str
model: str
department: str
project: str
input_tokens: int
output_tokens: int
cost_usd: float
cost_cny: float
latency_ms: int
request_id: Optional[str] = None
@dataclass
class BudgetAlert:
"""预算告警配置"""
department: str
project: str
monthly_budget_usd: float
warning_threshold: float = 0.8 # 80% 告警
critical_threshold: float = 0.95 # 95% 紧急
class HolySheepAPIClient:
"""
HolySheep AI API 统一封装
支持 OpenAI/Claude/Gemini/DeepSeek 格式
自动追踪 Token 用量和成本
"""
# HolySheep 官方价格表（output tokens, $/MTok）
MODEL_PRICES = {
"gpt-4.1": 8.00,
"claude-sonnet-4-5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
}
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url.rstrip("/")
self.usage_records: List[TokenUsage] = []
self.department_costs: Dict[str, float] = defaultdict(float)
self.project_costs: Dict[str, float] = defaultdict(float)
def chat_completion(
self,
model: str,
messages: List[Dict],
department: str = "default",
project: str = "default",
temperature: float = 0.7,
max_tokens: Optional[int] = None,
**kwargs
) -> Dict[str, Any]:
"""
统一的 Chat Completion 接口
Args:
model: 模型名称 (gpt-4.1, claude-sonnet-4-5, etc.)
messages: 消息列表
department: 部门名称（用于成本拆分）
project: 项目名称（用于成本拆分）
temperature: 温度参数
max_tokens: 最大输出 tokens
Returns:
API 响应结果
"""
start_time = time.time()
# 构建请求
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
}
if max_tokens:
payload["max_tokens"] = max_tokens
payload.update(kwargs)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# 发送请求
url = f"{self.base_url}/chat/completions"
response = requests.post(url, json=payload, headers=headers, timeout=60)
latency_ms = int((time.time() - start_time) * 1000)
if response.status_code != 200:
raise APIError(
f"API 请求失败: {response.status_code} - {response.text}",
status_code=response.status_code,
response=response.text
)
result = response.json()
# 提取 usage 信息
usage = result.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
# 计算成本（USD）- HolySheep 汇率 ¥1=$1
price_per_mtok = self.MODEL_PRICES.get(model, 0)
cost_usd = (input_tokens + output_tokens) / 1_000_000 * price_per_mtok
# 记录用量
usage_record = TokenUsage(
timestamp=datetime.now().isoformat(),
model=model,
department=department,
project=project,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=cost_usd,
cost_cny=cost_usd, # HolySheep: ¥1=$1
latency_ms=latency_ms,
request_id=result.get("id")
)
self.usage_records.append(usage_record)
# 累计部门/项目成本
self.department_costs[department] += cost_usd
self.project_costs[f"{department}:{project}"] += cost_usd
return result
def get_department_report(self, department: str) -> Dict[str, Any]:
"""生成部门月度报告"""
dept_records = [r for r in self.usage_records if r.department == department]
total_input = sum(r.input_tokens for r in dept_records)
total_output = sum(r.output_tokens for r in dept_records)
total_cost = sum(r.cost_usd for r in dept_records)
return {
"department": department,
"period": "monthly",
"total_requests": len(dept_records),
"input_tokens": total_input,
"output_tokens": total_output,
"total_tokens": total_input + total_output,
"total_cost_usd": round(total_cost, 2),
"total_cost_cny": round(total_cost, 2), # HolySheep 汇率
}
def get_project_breakdown(self, department: str) -> Dict[str, Any]:
"""生成项目维度拆分"""
projects = {}
for key, cost in self.project_costs.items():
if key.startswith(f"{department}:"):
project_name = key.split(":")[1]
projects[project_name] = round(cost, 2)
return projects
使用示例
if __name__ == "__main__":
client = HolySheepAPIClient(
api_key="YOUR_HOLYSHEEP_API_KEY" # 替换为你的 HolySheep API Key
)
# 研发部门 - A项目
response1 = client.chat_completion(
model="gpt-4.1",
messages=[
{"role": "system", "content": "你是一个专业的 Python 开发者"},
{"role": "user", "content": "写一个快速排序算法"}
],
department="engineering",
project="backend-service",
max_tokens=2000
)
# 产品部门 - B项目
response2 = client.chat_completion(
model="claude-sonnet-4-5",
messages=[
{"role": "user", "content": "帮我写一份产品需求文档模板"}
],
department="product",
project="mobile-app",
max_tokens=3000
)
print("=== 部门成本报告 ===")
for dept in ["engineering", "product"]:
report = client.get_department_report(dept)
print(f"\n{dept.upper()} 部门:")
print(f" 总请求数: {report['total_requests']}")
print(f" Input Tokens: {report['input_tokens']:,}")
print(f" Output Tokens: {report['output_tokens']:,}")
print(f" 总成本: ${report['total_cost_usd']:.2f} (约 ¥{report['total_cost_cny']:.2f})")
print("\n=== 项目拆分 ===")
print(f"工程部项目: {client.get_project_breakdown('engineering')}")
print(f"产品部项目: {client.get_project_breakdown('product')}")2. 预算告警系统
我实现了一个独立的预算监控模块，支持多维度告警：
import smtplib
import asyncio
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import requests
@dataclass
class AlertRecord:
"""告警记录"""
timestamp: str
level: str # "warning", "critical", "resolved"
department: str
project: str
current_cost: float
budget: float
usage_percent: float
message: str
class BudgetAlertSystem:
"""
预算告警系统
支持邮件、Webhook、企业微信多渠道告警
"""
def __init__(self, holy_sheep_client: HolySheepAPIClient):
self.client = holy_sheep_client
self.alerts: List[AlertRecord] = []
self.alert_configs: Dict[str, BudgetAlert] = {}
def add_budget_alert(self, alert: BudgetAlert):
"""添加预算告警配置"""
key = f"{alert.department}:{alert.project}"
self.alert_configs[key] = alert
def check_budget(self) -> List[AlertRecord]:
"""
检查所有配置的预算状态
返回需要告警的记录列表
"""
new_alerts = []
for key, config in self.alert_configs.items():
department, project = key.split(":")
# 获取当前成本
report = self.client.get_department_report(department)
project_costs = self.client.get_project_breakdown(department)
current_cost = project_costs.get(project, 0)
usage_percent = current_cost / config.monthly_budget_usd if config.monthly_budget_usd > 0 else 0
# 检查是否触发告警
alert_level = None
message = None
if usage_percent >= config.critical_threshold:
alert_level = "critical"
message = f"🚨 【紧急】{department}/{project} 预算已使用 {usage_percent*100:.1f}%，当前成本 ${current_cost:.2f}，月度预算 ${config.monthly_budget_usd:.2f}"
elif usage_percent >= config.warning_threshold:
alert_level = "warning"
message = f"⚠️ 【警告】{department}/{project} 预算已使用 {usage_percent*100:.1f}%，当前成本 ${current_cost:.2f}，月度预算 ${config.monthly_budget_usd:.2f}"
# 检查是否恢复到正常（从告警状态恢复）
prev_alerts = [a for a in self.alerts if a.department == department and a.project == project]
if prev_alerts and usage_percent < config.warning_threshold:
# 之前有告警，现在恢复正常
new_alerts.append(AlertRecord(
timestamp=datetime.now().isoformat(),
level="resolved",
department=department,
project=project,
current_cost=current_cost,
budget=config.monthly_budget_usd,
usage_percent=usage_percent,
message=f"✅ 【已恢复】{department}/{project} 预算使用率降至 {usage_percent*100:.1f}%"
))
if alert_level:
new_alerts.append(AlertRecord(
timestamp=datetime.now().isoformat(),
level=alert_level,
department=department,
project=project,
current_cost=current_cost,
budget=config.monthly_budget_usd,
usage_percent=usage_percent,
message=message
))
self.alerts.extend(new_alerts)
return new_alerts
def send_email_alert(
self,
alert: AlertRecord,
smtp_server: str,
smtp_port: int,
sender_email: str,
sender_password: str,
receiver_emails: List[str]
):
"""发送邮件告警"""
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[{alert.level.upper()}] AI 预算告警 - {alert.department}/{alert.project}"
msg["From"] = sender_email
msg["To"] = ", ".join(receiver_emails)
html_content = f"""
<html>
<body>
<h2>{alert.message}</h2>
<table border="1" cellpadding="5">
<tr><td>部门</td><td>{alert.department}</td></tr>
<tr><td>项目</td><td>{alert.project}</td></tr>
<tr><td>当前成本</td><td>${alert.current_cost:.2f}</td></tr>
<tr><td>月度预算</td><td>${alert.budget:.2f}</td></tr>
<tr><td>使用率</td><td>{alert.usage_percent*100:.1f}%</td></tr>
<tr><td>时间</td><td>{alert.timestamp}</td></tr>
</table>
<p>请及时处理，避免超支。</p>
</body>
</html>
"""
msg.attach(MIMEText(html_content, "html"))
try:
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(sender_email, sender_password)
server.sendmail(sender_email, receiver_emails, msg.as_string())
print(f"邮件告警已发送: {alert.message}")
except Exception as e:
print(f"邮件发送失败: {e}")
def send_webhook_alert(self, alert: AlertRecord, webhook_url: str):
"""发送 Webhook 告警（支持钉钉/飞书/企业微信）"""
payload = {
"msgtype": "markdown",
"markdown": {
"title": f"[{alert.level.upper()}] AI 预算告警",
"text": f"### {alert.message}\n\n"
f"| 项目 | 值 |\n"
f"| --- | --- |\n"
f"| 部门 | {alert.department} |\n"
f"| 项目 | {alert.project} |\n"
f"| 当前成本 | ${alert.current_cost:.2f} |\n"
f"| 月度预算 | ${alert.budget:.2f} |\n"
f"| 使用率 | {alert.usage_percent*100:.1f}% |\n"
f"| 时间 | {alert.timestamp} |"
}
}
try:
response = requests.post(webhook_url, json=payload, timeout=10)
if response.status_code == 200:
print(f"Webhook 告警已发送: {alert.message}")
else:
print(f"Webhook 发送失败: {response.status_code}")
except Exception as e:
print(f"Webhook 请求异常: {e}")
使用示例
if __name__ == "__main__":
# 初始化客户端
client = HolySheepAPIClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# 初始化告警系统
alert_system = BudgetAlertSystem(client)
# 配置告警规则
alert_system.add_budget_alert(BudgetAlert(
department="engineering",
project="backend-service",
monthly_budget_usd=500.0, # $500/月
warning_threshold=0.8,
critical_threshold=0.95
))
alert_system.add_budget_alert(BudgetAlert(
department="product",
project="mobile-app",
monthly_budget_usd=300.0, # $300/月
warning_threshold=0.8,
critical_threshold=0.95
))
# 模拟一些请求
for i in range(10):
client.chat_completion(
model="gpt-4.1",
messages=[{"role": "user", "content": "测试请求"}],
department="engineering",
project="backend-service"
)
# 检查预算并发送告警
new_alerts = alert_system.check_budget()
for alert in new_alerts:
print(f"\n[{alert.level.upper()}] {alert.message}")
# 根据告警级别发送通知
if alert.level == "critical":
# 发送邮件
# alert_system.send_email_alert(
# alert,
# smtp_server="smtp.gmail.com",
# smtp_port=587,
# sender_email="[email protected]",
# sender_password="your_password",
# receiver_emails=["[email protected]"]
# )
# 发送 Webhook
alert_system.send_webhook_alert(
alert,
webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_WEBHOOK_KEY"
)
elif alert.level == "warning":
# 仅发送 Webhook
alert_system.send_webhook_alert(
alert,
webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_WEBHOOK_KEY"
)3. 定时任务与数据持久化
为了实现真正的自动化审计，我编写了一个定时任务脚本，配合数据库实现长期数据追踪：
import sqlite3
from datetime import datetime, timedelta
from typing import List
import pandas as pd
class TokenAuditDatabase:
"""
Token 审计数据库
使用 SQLite 存储历史记录，支持按月/季度/年生成报表
"""
def __init__(self, db_path: str = "token_audit.db"):
self.db_path = db_path
self._init_database()
def _init_database(self):
"""初始化数据库表结构"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 用量记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS token_usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
model TEXT NOT NULL,
department TEXT NOT NULL,
project TEXT NOT NULL,
input_tokens INTEGER NOT NULL,
output_tokens INTEGER NOT NULL,
cost_usd REAL NOT NULL,
latency_ms INTEGER NOT NULL,
request_id TEXT,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# 预算配置表
cursor.execute("""
CREATE TABLE IF NOT EXISTS budget_config (
id INTEGER PRIMARY KEY AUTOINCREMENT,
department TEXT NOT NULL,
project TEXT NOT NULL,
monthly_budget_usd REAL NOT NULL,
start_date TEXT NOT NULL,
end_date TEXT,
is_active INTEGER DEFAULT 1,
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
UNIQUE(department, project, start_date)
)
""")
# 告警记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS alert_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
level TEXT NOT NULL,
department TEXT NOT NULL,
project TEXT NOT NULL,
current_cost_usd REAL NOT NULL,
budget_usd REAL NOT NULL,
usage_percent REAL NOT NULL,
message TEXT,
acknowledged INTEGER DEFAULT 0,
acknowledged_by TEXT,
acknowledged_at TEXT
)
""")
# 创建索引
cursor.execute("CREATE INDEX IF NOT EXISTS idx_usage_timestamp ON token_usage(timestamp)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_usage_dept ON token_usage(department)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_usage_project ON token_usage(department, project)")
conn.commit()
conn.close()
def save_usage(self, usage: TokenUsage):
"""保存单条用量记录"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO token_usage
(timestamp, model, department, project, input_tokens, output_tokens, cost_usd, latency_ms, request_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
usage.timestamp, usage.model, usage.department, usage.project,
usage.input_tokens, usage.output_tokens, usage.cost_usd,
usage.latency_ms, usage.request_id
))
conn.commit()
conn.close()
def batch_save_usage(self, usage_list: List[TokenUsage]):
"""批量保存用量记录"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.executemany("""
INSERT INTO token_usage
(timestamp, model, department, project, input_tokens, output_tokens, cost_usd, latency_ms, request_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", [(u.timestamp, u.model, u.department, u.project,
u.input_tokens, u.output_tokens, u.cost_usd,
u.latency_ms, u.request_id) for u in usage_list])
conn.commit()
conn.close()
print(f"已批量保存 {len(usage_list)} 条记录")
def get_monthly_report(self, year: int, month: int, department: str = None) -> pd.DataFrame:
"""生成月度报告"""
conn = sqlite3.connect(self.db_path)
start_date = f"{year}-{month:02d}-01"
if month == 12:
end_date = f"{year+1}-01-01"
else:
end_date = f"{year}-{month+1:02d}-01"
query = """
SELECT
department,
project,
model,
COUNT(*) as request_count,
SUM(input_tokens) as total_input,
SUM(output_tokens) as total_output,
SUM(input_tokens + output_tokens) as total_tokens,
SUM(cost_usd) as total_cost_usd,
AVG(latency_ms) as avg_latency_ms,
MAX(latency_ms) as max_latency_ms
FROM token_usage
WHERE timestamp >= ? AND timestamp < ?
"""
params = [start_date, end_date]
if department:
query += " AND department = ?"
params.append(department)
query += " GROUP BY department, project, model ORDER BY total_cost_usd DESC"
df = pd.read_sql_query(query, conn, params=params)
conn.close()
return df
def get_cost_trend(self, days: int = 30) -> pd.DataFrame:
"""获取成本趋势（按天）"""
conn = sqlite3.connect(self.db_path)
query = """
SELECT
DATE(timestamp) as date,
department,
SUM(cost_usd) as daily_cost
FROM token_usage
WHERE timestamp >= DATE('now', ?)
GROUP BY DATE(timestamp), department
ORDER BY date
"""
df = pd.read_sql_query(query, conn, params=[f"-{days} days"])
conn.close()
return df
def export_to_csv(self, year: int, month: int, output_path: str):
"""导出月度 CSV 报表"""
df = self.get_monthly_report(year, month)
df.to_csv(output_path, index=False, encoding="utf-8-sig")
print(f"报表已导出至: {output_path}")
定时任务执行脚本（可配合 cron 或 systemd timer）
if __name__ == "__main__":
import sys
db = TokenAuditDatabase("token_audit.db")
client = HolySheepAPIClient("YOUR_HOLYSHEEP_API_KEY")
# 从 HolySheep API 拉取历史数据（示例）
# 实际使用时需要实现数据同步逻辑
# 生成月度报告
now = datetime.now()
report = db.get_monthly_report(now.year, now.month)
if not report.empty:
print(f"\n=== {now.year}年{now.month}月 AI 成本报告 ===")
print(report.to_string(index=False))
# 导出 CSV
csv_path = f"ai_cost_report_{now.year}_{now.month}.csv"
db.export_to_csv(now.year, now.month, csv_path)
else:
print("当月暂无数据")常见报错排查
错误 1：Authentication Error（认证失败）
错误信息：AuthenticationError: Invalid API key provided
常见原因：
- API Key 拼写错误或缺少前后空格
- 使用了官方 API Key 而非 HolySheep Key
- Key 已过期或被撤销
解决方案：
# 错误示例
api_key = "sk-xxxx" # 官方 Key，HolySheep 不认
正确示例
api_key = "YOUR_HOLYSHEEP_API_KEY" # HolySheep Key
base_url = "https://api.holysheep.ai/v1"
client = HolySheepAPIClient(
api_key=api_key,
base_url=base_url # 必须指定 HolySheep 端点
)错误 2：Rate Limit Exceeded（速率限制）
错误信息：RateLimitError: Rate limit exceeded for model gpt-4.1
常见原因：
- 并发请求过多，触发速率限制
- 账户余额不足（部分套餐有限流）
- 短时间内大量请求
解决方案：
import time
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def create_session_with_retry(max_retries=3):
"""创建带重试机制的 Session"""
session = requests.Session()
retry_strategy = Retry(
total=max_retries,
backoff_factor=1, # 重试间隔：1s, 2s, 4s
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
使用
session = create_session_with_retry()
def call_with_retry(client, model, messages, max_attempts=3):
"""带重试的 API 调用"""
for attempt in range(max_attempts):
try:
response = client.chat_completion(model, messages)
return response
except RateLimitError as e:
if attempt < max_attempts - 1:
wait_time = 2 ** attempt
print(f"触发限流，等待 {wait_time}s 后重试...")
time.sleep(wait_time)
else:
raise e
调用
result = call_with_retry(client, "gpt-4.1", messages)