ChatGPT聊天记录归档实战：从数据收集到存储优化的完整方案

1次阅读

没有评论

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

在开发基于 ChatGPT 的应用时，聊天记录的归档管理往往容易被忽视。但随着对话量的增长，未归档的聊天记录会导致数据丢失、检索困难等问题。本文将分享一套完整的解决方案，帮助开发者高效管理聊天记录。

数据丢失风险：临时存储的聊天记录容易因服务器重启或故障而丢失
检索困难：海量对话内容难以快速定位特定会话
性能瓶颈：随着数据量增长，直接查询 API 会变得缓慢
缺乏分析能力：无法对历史对话进行统计和分析

优点：简单易用，无需额外数据库
缺点：不适合大规模数据，检索效率低

优点：事务支持完善，查询功能强大
缺点：固定 schema 不适合动态对话结构

优点：灵活文档模型，天然适合 JSON 数据
优点：水平扩展能力强，支持丰富查询
最终选择：MongoDB 作为存储方案

使用 aiohttp 实现高效异步请求：

import aiohttp
import asyncio

async def fetch_chat_history(api_key, session_id):
    headers = {'Authorization': f'Bearer {api_key}'}
    async with aiohttp.ClientSession() as session:
        async with session.get(f'https://api.openai.com/v1/sessions/{session_id}/messages',
            headers=headers
        ) as response:
            return await response.json()

优化的文档结构示例：

{"_id": ObjectId(...),
    "session_id": "sess_abc123",
    "user_id": "user_123",
    "messages": [
        {
            "role": "user|assistant",
            "content": "...",
            "timestamp": ISODate(...),
            "tokens": 42
        }
    ],
    "created_at": ISODate(...),
    "updated_at": ISODate(...),
    "metadata": {"tags": ["support", "billing"],
        "language": "zh-CN"
    }
}

使用 Python 的 html 模块进行转义：

import html

def sanitize_content(content):
    return html.escape(content)

from pymongo import MongoClient
from datetime import datetime, timedelta

class ChatArchiver:
    def __init__(self, mongo_uri):
        self.client = MongoClient(mongo_uri)
        self.db = self.client.chat_archive

    async def sync_new_messages(self, api_key, session_id):
        # 获取最后同步时间
        last_sync = self.db.sync_timestamps.find_one({"session_id": session_id})
        since = last_sync["timestamp"] if last_sync else None

        # 获取新消息
        new_messages = await fetch_messages_since(api_key, session_id, since)

        # 批量插入
        if new_messages:
            self.db.messages.insert_many([{**msg, "session_id": session_id}
                for msg in new_messages
            ])

            # 更新同步时间
            new_timestamp = max(msg["timestamp"] for msg in new_messages)
            self.db.sync_timestamps.update_one({"session_id": session_id},
                {"$set": {"timestamp": new_timestamp}},
                upsert=True
            )

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def fetch_with_retry(url, headers):
    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as response:
            response.raise_for_status()
            return await response.json()

def chunk_content(content, max_size=102400):
    chunks = []
    for i in range(0, len(content), max_size):
        chunks.append({"chunk_index": len(chunks),
            "content": content[i:i+max_size]
        })
    return chunks

会话查询索引：db.messages.create_index([("session_id", 1)])
时间范围索引：db.messages.create_index([("timestamp", -1)])
复合索引：db.messages.create_index([("user_id", 1), ("timestamp", -1)])

client = MongoClient(
    "mongodb://primary:27017,secondary1:27017,secondary2:27017/?replicaSet=rs0",
    read_preference=ReadPreference.SECONDARY_PREFERRED
)

启用 MongoDB 的 Snappy 压缩：storage: wiredTiger: engineConfig: cacheSizeGB: 2 journalCompressor: snappy collectionConfig: blockCompressor: snappy
应用层压缩大文本字段

from cryptography.fernet import Fernet

key = Fernet.generate_key()
cipher = Fernet(key)

encrypted_content = cipher.encrypt(b"Sensitive info")
decrypted_content = cipher.decrypt(encrypted_content)

# MongoDB 角色定义
use admin
db.createRole({
    role: "chat_reader",
    privileges: [{ resource: { db: "chat_archive", collection: "messages"}, actions: ["find"] }
    ],
    roles: []})