ChatGPT聊天记录归档实战：从数据收集到长期存储的完整解决方案

1次阅读

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

作为开发者，我们经常需要与 ChatGPT API 进行交互，生成大量的聊天记录。这些记录可能包含重要的业务信息、客户交流内容或者有价值的知识积累。然而，随着时间推移，这些数据可能会丢失或变得难以管理。今天，我将分享一套完整的解决方案，帮助你实现 ChatGPT 聊天记录的自动化归档。

在开始技术实现之前，让我们先了解为什么这是个重要的问题：

数据安全：API 返回的聊天记录如果不及时保存，可能会因为各种原因丢失
合规要求：某些行业需要对客户交互进行存档以满足监管要求
历史查询：未来可能需要回溯过去的对话内容
知识积累：有价值的对话可以成为组织的知识资产

我们有几个主要的存储选择，各有优缺点：

关系型数据库(如 MySQL)：
优点：结构化查询能力强
缺点：对话数据的半结构化特性不适合严格的表结构
文件存储(如 JSON 文件)：
优点：简单易用
缺点：难以查询和管理大量数据
文档数据库(如 MongoDB)：
优点：天然适合存储 JSON 格式的对话数据
缺点：需要额外的备份方案

基于这些考虑，我们选择 MongoDB 作为主要存储，并定期备份到 AWS S3 实现长期保存。

我们使用 Python 的 openai 库来获取聊天记录。首先需要安装必要的包：

pip install openai pymongo boto3

对话数据的结构设计很关键。我们采用这样的文档结构：

{
  "conversation_id": "唯一对话 ID",
  "user_id": "用户标识",
  "messages": [
    {
      "role": "user|assistant",
      "content": "消息内容",
      "timestamp": "时间戳"
    }
  ],
  "metadata": {"tags": ["标签 1", "标签 2"],
    "project": "关联项目"
  }
}

我们使用 AWS 的 boto3 库实现定期备份。关键是要处理大文件分块上传和断点续传。

下面是核心功能的 Python 实现：

import openai
from pymongo import MongoClient
import boto3
from datetime import datetime
import json

# 初始化连接
openai.api_key = 'your-api-key'
mongo_client = MongoClient('mongodb://localhost:27017/')
db = mongo_client['chatgpt_archive']
s3 = boto3.client('s3', 
                 aws_access_key_id='your-access-key',
                 aws_secret_access_key='your-secret-key')

# 获取聊天记录并存储
def archive_conversations(batch_size=100):
    conversations = []
    last_id = None

    # 实现分页获取
    while True:
        params = {"limit": batch_size}
        if last_id:
            params["after"] = last_id

        response = openai.ChatCompletion.list(**params)
        if not response.data:
            break

        for conv in response.data:
            # 构造 MongoDB 文档
            doc = {
                "conversation_id": conv.id,
                "messages": [
                    {
                        "role": m.role,
                        "content": m.content,
                        "timestamp": m.created_at
                    } for m in conv.messages
                ],
                "created_at": datetime.now()}
            conversations.append(doc)
            last_id = conv.id

        # 批量插入
        if len(conversations) >= batch_size:
            db.conversations.insert_many(conversations)
            conversations = []

    # 插入剩余记录
    if conversations:
        db.conversations.insert_many(conversations)

# 备份到 S3
def backup_to_s3(bucket_name):
    # 获取所有对话的 JSON 格式
    cursor = db.conversations.find({})
    data = [doc for doc in cursor]

    # 生成时间戳文件名
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"chatgpt_backup_{timestamp}.json"

    # 上传到 S3
    s3.put_object(
        Bucket=bucket_name,
        Key=filename,
        Body=json.dumps(data, default=str)
    )

在实际部署时，我们需要考虑几个关键点：