ChatGPT多轮对话开发实战：从会话管理到上下文保持

1次阅读

没有评论

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

最近在对接 OpenAI 的 ChatGPT API 时，发现要实现流畅的多轮对话，开发者常遇到几个头疼的问题：

上下文丢失：用户问了三个问题后，AI 突然『失忆』，完全不记得之前的对话
Token 超限：当对话历史超过 4096 个 token 时，API 直接返回错误
状态管理混乱：异步请求下多个客户端可能互相覆盖对话记录

这些问题不解决，做出来的聊天机器人就像金鱼一样只有 7 秒记忆。下面分享我们在实际项目中总结的解决方案。

很多新手会困惑 OpenAI 为什么提供两种 API：

传统 Completion
输入：单条文本提示
输出：续写内容
缺点：需要手动拼接历史对话
ChatCompletion

输入：messages 数组（含角色标记）

[{"role": "system", "content": "你是个助手"},
    {"role": "user", "content": "你好"},
    {"role": "assistant", "content": "你好！"}
]

优势：自动维护对话上下文
关键参数：temperature（创意程度）、max_tokens（响应长度限制）

下面这个 DialogueManager 类解决了 80% 的问题：

from typing import List, Dict, Optional
import tiktoken  # 用于 token 计数

class DialogueManager:
    def __init__(self, max_history_tokens: int = 3000):
        self.encoder = tiktoken.get_encoding("cl100k_base")
        self.max_history = max_history_tokens
        self.sessions: Dict[str, List[Dict]] = {}  # {session_id: messages}

    def add_message(self, session_id: str, role: str, content: str) -> None:
        """添加消息并自动修剪超限历史"""
        if session_id not in self.sessions:
            self.sessions[session_id] = []

        self.sessions[session_id].append({"role": role, "content": content})
        self._trim_history(session_id)

    def _trim_history(self, session_id: str) -> None:
        """优先保留最近对话，移除最早消息直到 token 数达标"""
        while True:
            total = sum(len(self.encoder.encode(msg["content"])) 
                       for msg in self.sessions[session_id])

            if total <= self.max_history or len(self.sessions[session_id]) <= 1:
                break

            # 保留 system 提示词，优先移除最早的用户 / 助手对话
            if self.sessions[session_id][0]["role"] == "system":
                removed = self.sessions[session_id].pop(1)
            else:
                removed = self.sessions[session_id].pop(0)

            print(f"移除消息: {removed['content'][:50]}...")

    async def get_response(self, session_id: str) -> Optional[str]:
        """调用 OpenAI API 并处理异常"""
        try:
            import openai
            resp = await openai.ChatCompletion.acreate(
                model="gpt-3.5-turbo",
                messages=self.sessions[session_id],
                temperature=0.7,
                max_tokens=500
            )
            return resp.choices[0].message.content

        except openai.InvalidRequestError as e:
            if "maximum context length" in str(e):
                self._handle_token_overflow(session_id)
            return "对话过长，已自动清理历史，请重新输入"
        except Exception as e:
            print(f"API 错误: {type(e).__name__}: {e}")
            return None

关键设计点：

Token 计数 ：使用官方tiktoken 库精确计算
修剪策略：永远保留 system 指令，按 FIFO 移除旧对话
错误处理：特别处理 4096 token 超限错误

预加载：对高频问题预生成回答模板
流式响应 ：使用stream=True 参数逐步显示结果
本地缓存：对相同问题直接返回缓存答案

def contains_sensitive(content: str) -> bool:
    blacklist = ["银行卡", "密码", "身份证号"]
    return any(word in content for word in blacklist)

# 在 add_message 中调用
if contains_sensitive(user_input):
    return "抱歉，此问题涉及敏感信息无法回答"

当不得不清空历史时：

保留最后 1 轮问答
插入系统提示：” 由于对话过长，已清理早期记录 ”
建议用户用更简洁的方式提问

import asyncio
from contextlib import asynccontextmanager

class DialogueManager:
    # ... 其他代码 ...

    @asynccontextmanager
    async def session_lock(self, session_id: str):
        """防止多个请求同时修改同一会话"""
        while hasattr(self, '_locks') and session_id in self._locks:
            await asyncio.sleep(0.1)

        if not hasattr(self, '_locks'):
            self._locks = {}

        self._locks[session_id] = True
        try:
            yield
        finally:
            del self._locks[session_id]

# 使用方式
async with manager.session_lock(user_id):
    await manager.get_response(user_id)

当用户量达到百万级时，内存中的 sessions 字典会成为瓶颈。你会如何设计分布式会话存储？考虑：