共计 4059 个字符，预计需要花费 11 分钟才能阅读完成。
作为开发者，当我们尝试将 ChatGPT 集成到自动化流程时，往往会遇到三个棘手的核心问题：API 的速率限制让我们难以高效处理大量请求，长对话中的上下文容易丢失导致对话不连贯，以及异步响应处理不当会造成资源浪费和性能瓶颈。下面我将分享一套经过实战检验的解决方案。
技术方案选型
OpenAI 官方 SDK vs 自定义封装
- 官方 SDK 优势：开箱即用，内置类型提示和参数验证，适合快速原型开发
- 自定义封装优势：可以灵活添加重试机制、监控指标和自定义日志，更适合生产环境
推荐生产环境采用混合方案：基于官方 SDK 进行二次封装。这样既能利用其健壮性，又能扩展企业级功能。
异步批量请求实现
使用 aiohttp 配合指数退避策略的典型实现：
import aiohttp
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
async def async_chat_completion(messages, session):
payload = {
"model": "gpt-3.5-turbo",
"messages": messages,
"temperature": 0.7
}
async with session.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload
) as response:
if response.status == 429:
raise Exception("Rate limit hit")
return await response.json()
async def batch_process(messages_list):
connector = aiohttp.TCPConnector(limit_per_host=5) # 控制并发连接数
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [async_chat_completion(msg, session) for msg in messages_list]
return await asyncio.gather(*tasks, return_exceptions=True)关键点说明：
- 通过
@retry装饰器实现指数退避重试 - TCPConnector 限制每主机连接数避免过量并发
- return_exceptions=True 确保单个失败不影响整个批处理
对话状态管理方案
方案一：Redis 存储（分布式场景）
import redis
import pickle
r = redis.Redis(host='localhost', port=6379, db=0)
def save_conversation(user_id, messages):
# 使用 pickle 序列化消息列表
serialized = pickle.dumps(messages)
r.setex(f"conv:{user_id}", 3600, serialized) # 1 小时过期
def load_conversation(user_id):
data = r.get(f"conv:{user_id}")
return pickle.loads(data) if data else []方案二：本地缓存（单机场景）
from cachetools import TTLCache
# 最多缓存 1000 个对话，每个保留 30 分钟
conversation_cache = TTLCache(maxsize=1000, ttl=1800)
def handle_message(user_id, new_msg):
history = conversation_cache.get(user_id, [])
history.append({"role": "user", "content": new_msg})
# 调用 API 并保存新上下文
response = get_chat_response(history)
history.append({"role": "assistant", "content": response})
conversation_cache[user_id] = history
return response流式响应处理
当需要处理大篇幅生成内容时，流式响应可以显著改善用户体验：
async def stream_response(messages):
async with aiohttp.ClientSession() as session:
payload = {
"model": "gpt-4",
"messages": messages,
"stream": True
}
async with session.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload
) as resp:
buffer = ""
async for chunk in resp.content:
if chunk:
decoded = chunk.decode('utf-8')
for line in decoded.split('\n'):
if line.startswith('data:'):
json_str = line[6:]
if json_str != "[DONE]":
data = json.loads(json_str)
delta = data["choices"][0]["delta"]
if "content" in delta:
buffer += delta["content"]
# 这里可以逐段处理内容
yield delta["content"]
return buffer性能优化实战
令牌消耗监控
def count_tokens(messages):
"""粗略估算 token 使用量"""
token_count = 0
for msg in messages:
token_count += len(msg["content"].split()) * 1.3 # 近似估值
return int(token_count)
class TokenTracker:
def __init__(self):
self.total_tokens = 0
def add_request(self, messages):
tokens = count_tokens(messages)
self.total_tokens += tokens
return tokens
# 使用示例
tracker = TokenTracker()
tokens_used = tracker.add_request(conversation_history)
print(f"本月已用 token: {tracker.total_tokens}")请求合并策略
对于批量处理场景，可以将相似请求合并：
def batch_similar_queries(queries, similarity_threshold=0.8):
"""使用文本相似度算法合并相似查询"""
grouped = {}
for query in queries:
matched = False
for template in grouped.keys():
if text_similarity(query, template) > similarity_threshold:
grouped[template].append(query)
matched = True
break
if not matched:
grouped[query] = [query]
return grouped
def process_batch(grouped_queries):
results = {}
for template, queries in grouped_queries.items():
# 只对模板执行一次 API 调用
response = get_chat_response([{"role":"user","content":template}])
for query in queries:
results[query] = customize_response(response, query)
return results避坑指南
敏感数据过滤
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内容审核集成
强烈建议在调用 ChatGPT 前先通过 Moderation API 检查内容：
def check_moderation(text):
response = openai.Moderation.create(input=text)
if response["results"][0]["flagged"]:
raise ContentPolicyViolation("输入内容违反使用政策")开放性问题
随着自动化规模扩大，我们需要考虑：
- 如何设计分布式任务队列来管理数百万计的 ChatGPT 请求？Celery 还是 Kafka 更合适？
- 对于需要外部知识库增强的场景，推荐使用 OpenAI 的 text-embedding-ada-002 还是开源的 Sentence-Transformers 模型？
这些问题的答案可能取决于你的具体用例和技术栈，但提前考虑这些扩展性问题能避免后期的架构返工。
正文完
