ChatGPT项目实战：从零搭建智能对话系统的避坑指南

1次阅读

共计 2160 个字符，预计需要花费 6 分钟才能阅读完成。

第一次用 ChatGPT API 做项目时，我踩过不少坑。最头疼的就是这三个问题：API 调用限制、长对话记忆丢失和 token 成本飙升。

API 速率限制：免费账户每分钟只有 3 次调用机会，付费账户也有 TPM（每分钟 token 数）限制，高峰期容易被限制
对话记忆丢失：默认 API 是无状态的，如果不处理上下文，机器人就像得了健忘症
token 成本失控：长对话的 token 消耗是指数级增长的，特别是用 gpt- 4 模型时

直接裸调 API 和使用 LangChain 框架各有优劣：

对比维度	直接调用 API	使用 LangChain
开发速度	快	中等（需学习框架）
延迟	低（直接 HTTP 请求）	略高（框架抽象层）
扩展性	需自行实现	内置记忆、工具等模块
适合场景	简单需求 / 原型开发	复杂 Agent 系统

对于大多数应用，我建议先用裸 API 快速验证，再逐步引入框架。

import openai
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def get_chat_response(messages, model="gpt-3.5-turbo"):
    try:
        response = openai.ChatCompletion.create(
            model=model,
            messages=messages,
            stream=True  # 启用流式响应
        )

        full_content = ""
        for chunk in response:
            content = chunk["choices"][0].get("delta", {}).get("content", "")
            if content:
                full_content += content
                yield content  # 流式返回

        return full_content
    except Exception as e:
        print(f"API 调用失败: {str(e)}")
        raise

关键点说明：
1. 使用 tenacity 实现指数退避重试
2. stream=True开启流式传输，改善用户体验
3. 通过 yield 逐步返回内容

import redis
import pickle
import time

class DialogueManager:
    def __init__(self, redis_host='localhost', ttl=3600):
        self.redis = redis.Redis(host=redis_host)
        self.ttl = ttl  # 上下文过期时间

    def save_context(self, session_id, messages):
        # 序列化并存储
        serialized = pickle.dumps(messages)
        self.redis.setex(f"chat:{session_id}", self.ttl, serialized)

    def load_context(self, session_id):
        # 反序列化读取
        serialized = self.redis.get(f"chat:{session_id}")
        return pickle.loads(serialized) if serialized else []

优化技巧：
– 设置合理的 TTL（如 1 小时），避免内存浪费
– 使用 pickle 序列化，比 JSON 更节省空间
– key 设计加入前缀（如chat:）方便管理

计算公式：

总 token = 提示 token + 完成 token
预估费用 = (总 token / 1000) * 模型单价

压缩策略：
1. 定期清理历史消息（保留最近 3 轮）
2. 对长文本先做摘要再传入
3. 设置 max_tokens 参数限制回复长度

import asyncio

async def process_concurrent_requests(queries):
    semaphore = asyncio.Semaphore(5)  # 控制并发数

    async def single_request(query):
        async with semaphore:
            return await get_async_response(query)

    return await asyncio.gather(*[single_request(q) for q in queries])

def contains_sensitive_content(text):
    blacklist = [...]  # 自定义敏感词库
    return any(word in text.lower() for word in blacklist)

# 在返回响应前检查
if contains_sensitive_content(response):
    return "抱歉，我无法回答这个问题"