共计 2480 个字符，预计需要花费 7 分钟才能阅读完成。
为什么需要对话归档系统
最近在项目里接入了 ChatGPT API，发现对话数据如果不加管理就会变成一团乱麻。典型的场景是：
- 用户连续提问 5 次后想回溯第 2 个问题的答案
- 三个月后需要审计某次敏感对话的完整上下文
- 不同设备间需要同步历史对话记录
这些需求暴露出原始日志文件的致命缺陷：
- 上下文断裂：对话被分割存储在不同文件
- 检索低效：grep 无法理解语义关联
- 合规风险：用户删除请求后仍残留敏感数据
存储方案选型实战
关系型数据库方案
用 PostgreSQL 实现的结构设计：
CREATE TABLE conversations (session_id VARCHAR(36) PRIMARY KEY,
user_id INT NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW());
CREATE TABLE messages (
id SERIAL PRIMARY KEY,
session_id VARCHAR(36) REFERENCES conversations,
role VARCHAR(10) CHECK (role IN ('user','assistant')),
content TEXT,
tokens INT,
seq_num INT -- 维护对话顺序的关键字段
);优势：
- 事务保证数据一致性
- 外键维护对话完整性
- 成熟的备份方案
劣势：
- JSON 格式的 prompt 需要额外处理
- 长对话性能下降明显
NoSQL 方案
MongoDB 的文档结构示例：
{
"_id": "session_123",
"metadata": {
"user": "u789",
"start_time": "2023-07-15T08:00:00Z"
},
"threads": [
{
"role": "user",
"text": "如何用 Python 发送邮件?",
"timestamp": "2023-07-15T08:00:01Z"
},
{
"role": "assistant",
"text": "可以使用 smtplib 模块...",
"tokens": 42,
"timestamp": "2023-07-15T08:00:03Z"
}
]
}优势：
- 天然适合非结构化数据
- 水平扩展容易
- 内置全文检索
劣势：
- 没有原生的事务支持
- 存储成本较高
混合架构实现
我们的生产环境最终采用分层方案：
flowchart LR
A[API 网关] --> B[Redis 热数据]
B --> C{7 天未访问?}
C -->|Yes| D[MongoDB 冷存储]
C -->|No| B
D --> E[S3 冰川备份]Python 异步归档核心代码
import aiohttp
from motor.motor_asyncio import AsyncIOMotorClient
class ChatArchiver:
def __init__(self):
self.db = AsyncIOMotorClient("mongodb://localhost:27017").chat_db
async def save_streaming_response(self, session_id, response):
buffer = ""
async for chunk in response.content:
buffer += chunk.decode()
# 遇到完整消息边界
if "\n\n" in buffer:
msg_part, buffer = buffer.split("\n\n", 1)
await self.db.sessions.update_one({"_id": session_id},
{"$push": {"messages": {"$each": [msg_part]}}},
upsert=True
)关键点说明：
- 使用 motor 实现异步 MongoDB 操作
- 流式处理避免内存溢出
- 原子操作保证消息顺序
生产环境特别注意事项
时间戳冲突解决方案
我们遇到过时钟回拨导致消息乱序的问题，最终采用三种防御措施：
- 逻辑时钟：Lamport 时间戳算法
- 混合序号：
timestamp_epoch + atomic_counter - 服务端时间：直接使用 API 返回的 created 字段
限流处理模板
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type
)
@retry(stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10),
retry=retry_if_exception_type(aiohttp.ClientError)
)
async def call_with_retry(prompt):
async with aiohttp.ClientSession() as session:
async with session.post(API_URL, json={"prompt": prompt}) as resp:
if resp.status == 429:
raise aiohttp.ClientError("Rate limited")
return await resp.json()进阶思考
如何实现这样的搜索：” 找出所有讨论过 Python 装饰器但未提及 @staticmethod 的对话 ”？
这需要：
- 构建对话向量索引（可用 sentence-transformers）
- 实现负向关键词过滤
- 跨消息的语义关联分析
推荐用 promptfoo 这个工具评估归档后的对话质量，它支持自动化测试：
npm install -g promptfoo
promptfoo eval -c config.yaml踩坑心得
最大的教训是：不要假设对话是线性发生的！实际会遇到：
- 用户同时从手机和 PC 端发送消息
- API 响应因网络抖动乱序到达
- 客户端重试导致重复消息
最终我们通过 (session_id, sequence_num) 的联合索引解决问题，其中 sequence_num 由服务端严格单调递增生成。
这套系统上线后，客服工单处理速度提升了 60%，因为现在能快速定位历史对话上下文了。如果你有更好的实现方案，欢迎交流讨论！
正文完
