共计 3962 个字符，预计需要花费 10 分钟才能阅读完成。
背景痛点
在使用 ChatGPT 这类对话模型时，经常会遇到历史对话丢失的问题。未存档的对话会导致几个明显的痛点：
- 数据丢失：当会话超时或页面刷新后，之前的有价值对话内容无法找回
- 上下文断裂：在多轮对话中，模型无法参考之前的对话历史，影响回答质量
- 检索困难：当需要查找特定信息时，没有有效的搜索和分类机制
技术选型
对话数据有其独特的特点：非结构化、增长快速、需要灵活查询。我们对比几种常见数据库：
- MongoDB：文档型数据库，适合存储 JSON 格式的对话记录，支持灵活查询和水平扩展
- MySQL：关系型数据库，对事务支持好但表结构固定，不适合频繁变化的对话结构
- Redis：内存数据库，速度快但持久化能力有限，适合缓存而非长期存储
综合考虑，MongoDB 是最佳选择，因为它：
- 支持嵌套文档结构，完美匹配对话的树形结构
- 提供丰富的查询操作符
- 易于水平扩展应对数据增长
核心实现
Flask API 搭建
首先安装必要依赖：
pip install flask pymongo python-dotenv创建基础 Flask 应用：
from flask import Flask, request, jsonify
from pymongo import MongoClient
from datetime import datetime
import logging
app = Flask(__name__)
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# MongoDB 连接
client = MongoClient('mongodb://localhost:27017/')
db = client['chatgpt_archive']
conversations = db['conversations']
@app.route('/api/conversations', methods=['POST'])
def save_conversation():
try:
data = request.json
# 基础验证
if not data or 'messages' not in data:
return jsonify({'error': 'Invalid data'}), 400
# 构建文档
doc = {'user_id': data.get('user_id', 'anonymous'),
'timestamp': datetime.utcnow(),
'messages': data['messages'],
'metadata': data.get('metadata', {})
}
# 插入数据库
result = conversations.insert_one(doc)
return jsonify({'id': str(result.inserted_id)}), 201
except Exception as e:
logger.error(f"Error saving conversation: {str(e)}")
return jsonify({'error': 'Internal server error'}), 500MongoDB 文档设计
我们的对话文档设计包含以下关键字段：
_id: 自动生成的唯一标识user_id: 用户标识timestamp: 对话时间messages: 消息数组，每个消息包含:role: user/assistantcontent: 消息内容timestamp: 消息时间metadata: 额外元数据，如:tags: 自定义标签topic: 对话主题
这种设计支持：
- 按用户查询所有对话
- 按时间范围检索
- 对消息内容进行全文搜索
高级功能
用户对话检索
添加按用户 ID 查询的端点：
@app.route('/api/conversations/<user_id>', methods=['GET'])
def get_conversations(user_id: str):
try:
page = int(request.args.get('page', 1))
per_page = int(request.args.get('per_page', 10))
skip = (page - 1) * per_page
cursor = conversations.find({'user_id': user_id},
sort=[('timestamp', -1)],
skip=skip,
limit=per_page
)
results = []
for doc in cursor:
doc['_id'] = str(doc['_id']) # 转换 ObjectId 为字符串
results.append(doc)
return jsonify(results)
except Exception as e:
logger.error(f"Error retrieving conversations: {str(e)}")
return jsonify({'error': 'Internal server error'}), 500语义搜索实现
使用 scikit-learn 实现 TF-IDF 搜索：
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
# 假设我们已经从 MongoDB 加载了所有对话
all_conversations = [...]
def search_conversations(query: str, top_n: int = 5):
# 将所有对话内容合并为文本
texts = [''.join([msg['content'] for msg in conv['messages']])
for conv in all_conversations]
# 创建 TF-IDF 向量器
vectorizer = TfidfVectorizer()
tfidf_matrix = vectorizer.fit_transform(texts)
# 处理查询
query_vec = vectorizer.transform([query])
# 计算相似度
similarities = cosine_similarity(query_vec, tfidf_matrix).flatten()
# 获取最相似的对话
top_indices = similarities.argsort()[-top_n:][::-1]
return [all_conversations[i] for i in top_indices]性能优化
批量插入策略
当需要存档大量对话时，使用批量插入：
def bulk_insert_conversations(conversation_list: list):
try:
# 准备文档
docs = [{'user_id': conv['user_id'],
'timestamp': datetime.utcnow(),
'messages': conv['messages'],
'metadata': conv.get('metadata', {})
} for conv in conversation_list]
# 批量插入
result = conversations.insert_many(docs)
return result.inserted_ids
except Exception as e:
logger.error(f"Bulk insert failed: {str(e)}")
raise索引优化
为常用查询字段创建索引：
# 在应用启动时创建索引
conversations.create_index([('user_id', 1)])
conversations.create_index([('timestamp', -1)])
conversations.create_index([('metadata.tags', 1)])避坑指南
敏感信息过滤
在保存对话前过滤敏感信息：
SENSITIVE_KEYWORDS = ['password', 'credit card', 'ssn']
def sanitize_content(content: str) -> str:
for keyword in SENSITIVE_KEYWORDS:
if keyword in content.lower():
content = content.replace(keyword, '[REDACTED]')
return content分块存储策略
对于长对话，实施分块存储：
- 设定最大消息数（如 50 条）
- 当达到阈值时，自动创建新对话文档
- 使用
prev_conversation_id字段维护对话链
延伸思考：转为知识库
将此系统扩展为知识库的几种方式：
- 添加问答对提取功能，从对话中识别有价值的 Q &A
- 实现自动标记和分类系统
- 构建知识图谱，发现对话间的关联
- 添加审核流程，确保知识库质量
本地测试
使用 cURL 测试 API：
# 保存对话
curl -X POST http://localhost:5000/api/conversations \
-H "Content-Type: application/json" \
-d '{"user_id":"user123","messages": [{"role": "user", "content": "如何学习 Python?"},
{"role": "assistant", "content": "建议从官方教程开始..."}
]
}'
# 查询对话
curl http://localhost:5000/api/conversations/user123通过以上步骤，我们构建了一个完整的 ChatGPT 对话存档系统。这个系统不仅解决了基础的数据持久化问题，还提供了高级检索和分析能力，为进一步的知识管理奠定了基础。
正文完
