共计 2245 个字符，预计需要花费 6 分钟才能阅读完成。
背景痛点分析
最近在项目里接入了 ChatGPT API，踩了不少坑。这里总结下开发者最常遇到的几个问题：
- 响应延迟不稳定 ：高峰期 API 响应时间波动从 1 秒到 15 秒不等，直接影响用户体验
- Token 消耗黑洞 ：发现用户一个简单问题可能消耗 200-300 tokens，复杂对话轻松突破 1000 tokens（GPT- 4 每 1k tokens 费用 $0.03/$0.06）
- 状态维护困难 ：多轮对话时，客户端维护历史消息的 token 数很快突破模型限制（如 gpt-3.5-turbo 的 4096 限制）
实测数据：当并发请求达到 50QPS 时，每月 API 成本会突然从 $200 飙升至 $5000+，主要来自：
- 无效的重复请求
- 未优化的 prompt 结构
- 缺少 streaming 处理
核心技术方案
1. 调用模式选择
# 非 streaming 模式（适合简单请求）response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "你好"}]
)
# streaming 模式（推荐生产环境使用）response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[...],
stream=True # 关键参数
)差异对比：
| 特性 | Streaming 模式 | 普通模式 |
|---|---|---|
| 首字节时间 (TTFB) | 0.5-2s | 2-15s |
| 内存占用 | 低 | 高 |
| 适用场景 | 长文本生成 | 简单应答 |
2. System Message 设计模板
优化前后的 prompt 对比：
| 版本 | 示例 | Token 消耗 |
|---|---|---|
| 原始版 | “ 请回答用户问题 ” | 8 |
| 优化版 | “ 你是一个专业客服，用中文回答，保持简洁 ” | 15 |
虽然 token 变多，但实际测试发现优化版能减少 20% 的后续交互次数。关键技巧：
- 明确角色定位（如客服 / 导师 / 助手）
- 指定响应语言
- 设置 temperature=0.7 平衡创意与稳定
3. 自动重试机制
import asyncio
from aiohttp import ClientSession
async def chat_completion_with_retry(messages, max_retries=3):
backoff_factor = 1 # 初始等待 1 秒
for attempt in range(max_retries):
try:
async with ClientSession() as session:
response = await session.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={"model": "gpt-3.5-turbo", "messages": messages},
timeout=10
)
if response.status == 429:
wait_time = backoff_factor * (2 ** attempt)
await asyncio.sleep(wait_time)
continue
return await response.json()
except Exception as e:
print(f"Attempt {attempt + 1} failed: {str(e)}")
raise Exception("Max retries exceeded")生产环境实战
压力测试方案
使用 Locust 模拟高并发：
from locust import HttpUser, task, between
class ChatGPTUser(HttpUser):
wait_time = between(1, 3)
@task
def ask_question(self):
self.client.post("/chat", json={"messages": [{"role": "user", "content": "如何学习 Python?"}],
"max_tokens": 100
})敏感信息过滤
import re
def sanitize_input(text):
patterns = [r"\b\d{4}[-]?\d{4}[-]?\d{4}\b", # 信用卡号
r"\b\d{3}-?\d{2}-?\d{4}\b" # SSN
]
for pattern in patterns:
text = re.sub(pattern, "[REDACTED]", text)
return text避坑指南
- Prompt 注入防护 ：
- 永远不要直接拼接用户输入到 prompt
-
使用如下转义方法：
user_input.replace("{", "{{}").replace("}", "{}}") -
对话深度控制 ：
-
实现简单状态机：
graph LR A[开始] --> B{轮次 <5?} B -->| 是 | C[继续对话] B -->| 否 | D[结束会话] -
GDPR 合规要点 ：
- 日志保留不超过 30 天
- 提供用户数据删除接口
思考题
- 分级缓存策略可以这样设计：
- Level1：内存缓存高频问题（5 分钟 TTL）
- Level2：Redis 缓存常见问题（1 小时 TTL）
-
Level3：数据库存储标准问答（永久）
-
PDF 处理 token 分段建议：
- 按章节拆分
- 每段保留上下文摘要
- 使用文本向量检索关键段落
经过这些优化，我们的生产系统实现了：
– API 延迟降低 40%
– 月度成本下降 35%
– 错误率从 8% 降至 0.5%
关键心得：与其盲目升级模型版本，不如先优化现有调用方式。下期会分享如何结合 LangChain 实现更复杂的对话管理。
正文完
