ChatGPT对话导出实战：从API调用到数据持久化的完整解决方案

1次阅读

没有评论

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

在开发基于 ChatGPT 的应用时，我们常常遇到几个关键问题：

临时性数据缺陷：默认情况下，ChatGPT 的对话记录仅保存在内存中或短期缓存里，一旦服务重启或会话超时，历史对话就会消失。这对于需要长期保存对话记录的业务场景（如客服系统、咨询记录）来说是个致命缺陷。
合规性需求：在企业级应用中，尤其是金融、医疗等行业，通常有严格的审计合规要求。必须能够追溯完整的对话历史，并且保证记录不可篡改。
上下文丢失风险：ChatGPT 的多轮对话依赖上下文连贯性。如果中间某次对话丢失，后续的交互质量会显著下降。

直接存储 JSON：
优点：实现简单，无需数据库依赖
缺点：难以查询和统计分析，数据量大时性能差
数据库存储：
优点：支持复杂查询，易于扩展
缺点：需要额外的数据库服务

同步写入：
适合对数据一致性要求极高的场景
但会显著增加 API 响应时间
异步队列：
使用消息队列（如 RabbitMQ/Kafka）缓冲写入请求
提高系统吞吐量，但存在短暂的数据不一致窗口

增量导出：仅同步新增的对话内容
适合高频对话场景
需要完善的 ID 跟踪机制
全量导出：定期完整备份所有对话
实现简单
资源消耗大，适合低频场景

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

import aiohttp
import asyncio

async def fetch_conversation(conversation_id):
    async with aiohttp.ClientSession() as session:
        headers = {"Authorization": f"Bearer {API_KEY}"}
        async with session.get(f"https://api.openai.com/v1/conversations/{conversation_id}",
            headers=headers
        ) as response:
            return await response.json()

使用正则提取关键信息：

import re

def parse_message(raw_text):
    # 匹配用户和 AI 的对话段落
    pattern = r"(User|AI): (.+?)(?=\n\w+:|\Z)"
    return re.findall(pattern, raw_text, re.DOTALL)

使用 SQLAlchemy 定义数据模型：

from sqlalchemy import Column, Integer, String, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Conversation(Base):
    __tablename__ = 'conversations'

    id = Column(Integer, primary_key=True)
    conversation_id = Column(String(64), unique=True)  # ChatGPT 对话 ID
    user_id = Column(String(36))  # 业务系统用户 ID
    created_at = Column(DateTime)
    content = Column(Text)  # 结构化后的对话内容
    checksum = Column(String(32))  # 数据校验 MD5

import os
import hashlib
from datetime import datetime

# 环境变量加载 API Key
API_KEY = os.getenv("OPENAI_API_KEY")

async def export_conversations():
    """导出所有对话记录并存入数据库"""
    try:
        # 获取对话列表
        conversations = await list_conversations()

        # 批量处理
        for conv in conversations:
            # 获取完整对话内容
            details = await fetch_conversation(conv['id'])

            # 计算校验和
            raw_content = str(details)
            checksum = hashlib.md5(raw_content.encode()).hexdigest()

            # 构建 ORM 对象
            record = Conversation(conversation_id=conv['id'],
                user_id=conv['user'],
                created_at=datetime.fromtimestamp(conv['created']),
                content=raw_content,
                checksum=checksum
            )

            # 保存到数据库
            await save_to_db(record)

    except Exception as e:
        # 异常处理和重试逻辑
        logger.error(f"Export failed: {str(e)}")
        raise

使用复合主键：(conversation_id, turn_number)
添加数据库唯一约束

def sanitize_content(text):
    """脱敏处理"""
    # 替换信用卡号
    text = re.sub(r"\b\d{4}[-]?\d{4}[-]?\d{4}[-]?\d{4}\b", "[CREDIT_CARD]", text)
    # 替换手机号
    text = re.sub(r"\b1[3-9]\d{9}\b", "[PHONE]", text)
    return text

推荐使用 AES-256 加密备份文件：

from cryptography.fernet import Fernet

def encrypt_backup(data):
    key = Fernet.generate_key()
    cipher = Fernet(key)
    return cipher.encrypt(data.encode()), key

import redis

# 对话元数据缓存
r = redis.Redis()

def cache_conversation_meta(conv_id, meta):
    r.hset(f"conv:{conv_id}", mapping=meta)
    r.expire(f"conv:{conv_id}", 3600)  # 1 小时过期

建议每 100-500 条记录批量提交一次事务