共计 1924 个字符，预计需要花费 5 分钟才能阅读完成。
背景与痛点
ChatGPT 的 API 默认采用无状态设计，这意味着每次 API 调用都是独立的，服务器不会自动保存之前的对话历史。这种设计带来了几个关键问题：
- 无状态特性：每个 API 请求都是独立的，服务器不保留任何上下文信息
- Token 限制：即使传递历史消息，也要受模型最大 token 数限制（如 gpt-3.5-turbo 通常为 4096 tokens）
- 会话标识缺失：默认不提供会话 ID 机制，开发者需要自行管理对话连续性
技术方案对比
方案 1：外部存储维护完整会话历史
适用场景：需要长期保存完整对话记录的复杂应用
优点：
– 完整保存对话上下文
– 支持对话历史检索和分析
缺点：
– 存储成本较高
– 需要处理数据序列化和反序列化
方案 2：利用 messages 参数传递上下文
适用场景：短期对话或 token 消耗可控的场景
优点：
– 实现简单，无需额外存储
– API 原生支持
缺点：
– 受 token 限制影响大
– 长对话时效率下降
方案 3：自定义会话 ID+ 数据库混合方案
适用场景：需要平衡性能和持久化需求的中大型应用
优点：
– 灵活控制存储粒度
– 可实现分级存储
缺点：
– 实现复杂度较高
– 需要维护会话状态
核心实现
Redis 存储示例
import json
from datetime import timedelta
import redis
from pydantic import BaseModel
class ChatMessage(BaseModel):
role: str # 'system', 'user' or 'assistant'
content: str
class ChatSession:
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def save_session(self, session_id: str, messages: list[ChatMessage], ttl_hours: int = 24):
"""保存会话到 Redis，设置 TTL"""
serialized = json.dumps([msg.dict() for msg in messages])
self.redis.setex(f"chat:{session_id}", timedelta(hours=ttl_hours), serialized)
def load_session(self, session_id: str) -> list[ChatMessage] | None:
"""从 Redis 加载会话"""
if data := self.redis.get(f"chat:{session_id}"):
return [ChatMessage(**msg) for msg in json.loads(data)]
return None构造 messages 数组
def prepare_messages(session_messages: list[ChatMessage],
new_message: str,
max_tokens: int = 3500
) -> list[dict]:
"""构造符合 API 要求的 messages 数组，自动截断过长的历史"""
messages = [msg.dict() for msg in session_messages]
messages.append({"role": "user", "content": new_message})
# 简单的 token 计数和截断策略
while len(json.dumps(messages)) > max_tokens and len(messages) > 1:
messages.pop(1) # 保留系统提示，移除最早的用户 / 助手消息
return messages避坑指南
- Token 计算陷阱
- 实际 token 数与 JSON 字符串长度并非 1:1 关系
-
建议使用 tiktoken 库精确计算
-
对话隔离
- 多用户场景要确保会话 ID 不冲突
-
考虑添加用户 ID 前缀
-
敏感信息加密
- 对医疗 / 金融等敏感对话建议加密存储
- 可使用 AWS KMS 或类似服务管理加密密钥
性能考量
| 方案 | 延迟 | 存储开销 | API 成本 |
|---|---|---|---|
| 外部存储 | 中 | 高 | 低 |
| messages 参数 | 低 | 无 | 高 |
| 混合方案 | 中 | 中 | 中 |
架构流程图（方案 3）
flowchart TD
A[用户请求] --> B{会话 ID 存在?}
B -->| 是 | C[从数据库加载历史]
B -->| 否 | D[创建新会话]
C --> E[构造 messages 数组]
D --> E
E --> F[调用 ChatGPT API]
F --> G[保存更新后的会话]
G --> H[返回响应]开放性问题
当对话历史超过 10 万条时，如何设计分级存储方案？可以考虑：
1. 热数据（最近 3 天）保持 Redis 缓存
2. 温数据（1 月内）存入关系型数据库
3. 冷数据归档到对象存储（如 S3）
4. 实现基于时间 / 访问频率的自动迁移策略
正文完
