共计 1393 个字符,预计需要花费 4 分钟才能阅读完成。
在业务中集成 ChatGPT API 时,开发者常遇到三个典型问题:突发流量导致 429 限频错误、长对话场景下的上下文丢失,以及流式响应时的延迟抖动。本文将用可落地的代码和配置解决这些问题。

环境配置与协议选型
OpenCode 核心配置
# .opencode/project.yaml
chatgpt:
endpoint: azure # 可选 openai/azure
api_version: 2023-05-15
throttle_retry: 3 # 429 错误重试次数
timeout:
connect: 10.0
read: 30.0
# 敏感信息通过环境变量注入
# VAULT_PATH: /secrets/chatgpt-api
端点选择决策树
- 是否需要企业级 SLA?→ 选 Azure
- 是否需要最新模型(如 GPT-4-turbo)?→ 选 OpenAI
- 是否需私有化部署?→ 选 Azure
异步流式实现
import aiohttp
from backoff import on_exception, expo
class ChatGPTStreamer:
def __init__(self, api_key):
self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)
)
@on_exception(expo, aiohttp.ClientError, max_tries=3)
async def stream_response(self, messages):
async with self.session.post(f"{self.base_url}/chat/completions",
json={
"messages": messages,
"stream": True,
"max_tokens": 500 # 实测最佳平衡点
},
headers={"Authorization": f"Bearer {self.api_key}"}
) as resp:
async for chunk in resp.content:
yield chunk.decode()
性能优化实战
Token 与延迟测试数据
| max_tokens | 平均延迟 (ms) | 费用 / 千次调用 |
|---|---|---|
| 200 | 420 | $1.20 |
| 500 | 680 | $2.85 |
| 1000 | 1200 | $5.60 |
Prompt 压缩技术
def compress_prompt(text):
# 移除重复短语
words = text.split()
return ' '.join(sorted(set(words), key=words.index))
安全防护体系
Vault 密钥注入
# 原危险写法
- API_KEY = "sk-xxxx"
# 正确写法
+ API_KEY = os.environ['CHATGPT_KEY']
输入过滤器
BANNED_WORDS = [...] # 从合规部门获取
def sanitize_input(text):
return not any(word in text.lower() for word in BANNED_WORDS)
生产检查清单
- 监控 API 每分钟调用量 ≤ 300 次 / 分钟
- 平均响应时间 ≤ 800ms (P99 < 1.5s)
- 错误率(4xx/5xx)< 0.5%
- Token 消耗速率 ≤ 50k/ 分钟
- 上下文缓存命中率 > 70%
这套方案在我们客服系统中稳定运行了 6 个月,日均处理 20 万请求。关键是把重试机制和流式响应结合起来,既保证可靠性又不牺牲用户体验。
正文完
