共计 2365 个字符，预计需要花费 6 分钟才能阅读完成。
背景痛点
直接调用 ChatGPT Plus API 时，开发者常遇到三个典型问题：
- 速率限制（429 错误）：OpenAI 对免费账号和 Plus 账号都有严格的 RPM（每分钟请求数）和 TPM（每分钟 token 数）限制。在突发流量下，直接调用会导致大量请求被拒绝。
- 响应延迟波动 ：API 的响应时间受服务器负载影响显著，尤其在多轮对话场景下，延迟可能从 200ms 陡增到 2s 以上。
- Token 浪费 ：未优化的单次请求往往包含冗余的 system prompt 或重复上下文，导致 token 消耗超出实际需求。
技术方案对比
轮询模式
- 优点：实现简单，适合低频场景
- 缺点：高延迟（需等待轮询间隔），高资源消耗
Webhook 模式
- 优点：实时性好，服务器压力小
- 缺点：需要公网回调地址，调试复杂
消息队列模式（推荐）
- 优点：削峰填谷，支持背压控制
- 缺点：需要额外维护队列服务
核心方案实现
Redis 请求缓冲队列
使用 Redis Sorted Set 实现优先级队列，其中 score 为请求时间戳，member 为序列化的请求体。关键操作：
import redis
from datetime import datetime
r = redis.Redis()
def enqueue_request(request_id: str, payload: dict, priority: int = 0):
timestamp = datetime.now().timestamp()
# 优先级越高 score 越小
score = timestamp - priority * 1000
r.zadd('chatgpt:queue', {json.dumps(payload): score})动态批处理算法
通过合并相似请求的 system prompt 减少重复 token 消耗：
- 从队列批量获取 10-20 个请求
- 提取共用的 system prompt 作为基础模板
- 为每个请求生成差异化的 user prompt
- 合并发送批量请求
def batch_requests(requests: List[dict]) -> dict:
base_prompt = find_common_prompt(requests) # 实现相似度检测算法
messages = [{"role": "system", "content": base_prompt},
*[{"role": "user", "content": r["prompt"]} for r in requests]
]
return {"messages": messages}指数退避重试
对于失败请求采用阶梯式重试间隔：
import time
def retry_with_backoff(fn, max_retries=3):
for attempt in range(max_retries):
try:
return fn()
except Exception as e:
wait = min(2 ** attempt, 60) # 上限 60 秒
time.sleep(wait)
raise Exception(f"Failed after {max_retries} retries")性能测试数据
在 4C8G 的 AWS c5.xlarge 实例上测试：
| 方案 | 成功率 (1000QPS) | P99 延迟 | Token/ 请求 |
|---|---|---|---|
| 直接调用 | 72% | 4.2s | 1200 |
| 本方案 | 99.6% | 1.8s | 680 |
避坑指南
Streaming 内存泄漏
处理流式响应时必须及时消费数据：
# 错误示例：会累积所有 chunks 在内存
response = client.chat.completions.create(stream=True)
full_content = "".join([chunk.choices[0].delta.content for chunk in response]) # 内存爆炸！# 正确做法
with open('output.txt', 'w') as f:
for chunk in response:
if chunk_content := chunk.choices[0].delta.content:
f.write(chunk_content) # 即时写入 成本监控
通过 Prometheus 暴露关键指标：
from prometheus_client import Counter, Gauge
REQUEST_COST = Counter('chatgpt_token_consumed', 'Total tokens used')
API_LATENCY = Gauge('chatgpt_response_ms', 'API latency in milliseconds')
def track_usage(response):
REQUEST_COST.inc(response.usage.total_tokens)
API_LATENCY.set(response.response_time_ms)延伸思考
本文方案可以迁移到其他 LLM 服务，但需注意：
- Claude 等模型有更严格的上下文窗口限制
- 自托管模型需要考虑 GPU 显存与批处理大小的平衡
- 部分商业 API 会限制单个请求的最大 token 数
建议通过抽象 Provider 接口实现多引擎支持：
class LLMProvider(ABC):
@abstractmethod
def chat_completion(self, messages: List[dict]) -> dict: ...
class OpenAIImpl(LLMProvider): ...
class AnthropicImpl(LLMProvider): ...结语
通过队列缓冲 + 动态批处理的组合方案，我们成功将生产环境的 API 稳定性提升到 SLA 99.9% 的水平。实际部署时建议配合 HPA 自动扩展处理 worker，并设置每日 token 预算告警。这套方案已在电商客服场景下验证，日均处理 200 万条消息，相比原始实现节省约 $15,000/ 月的 API 成本。
正文完
