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

问题背景与需求分析

在实际开发中，很多团队需要将 ChatGPT 的对话历史保存下来用于后续分析、训练或审计。常见痛点包括：

• 官方 API 默认不提供完整的对话导出功能
• 手动复制粘贴在批量处理时效率极低
• 网络不稳定可能导致数据丢失
• 缺乏结构化存储格式

架构设计

系统采用分层设计：

1. API 交互层：处理与 ChatGPT 的通信
2. 数据转换层：将 API 响应标准化
3. 存储层：支持多种存储后端
4. 调度层：管理导出任务队列

flowchart TD
    A[用户请求] --> B[任务队列]
    B --> C{异步处理}
    C -->| 成功 | D[存储系统]
    C -->| 失败 | E[重试机制]

核心代码实现

基础导出功能

import openai
import json
from datetime import datetime

class ChatGPTExporter:
    def __init__(self, api_key):
        openai.api_key = api_key
        self.session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"

    def export_conversation(self, prompt, max_retries=3):
        """
        导出单次对话记录
        :param prompt: 用户输入的提示词
        :param max_retries: 最大重试次数
        :return: 标准化格式的对话记录
        """
        retry_count = 0
        while retry_count < max_retries:
            try:
                response = openai.ChatCompletion.create(
                    model="gpt-3.5-turbo",
                    messages=[{"role": "user", "content": prompt}]
                )
                return self._format_record(prompt, response)
            except Exception as e:
                retry_count += 1
                print(f"Attempt {retry_count} failed: {str(e)}")
                if retry_count == max_retries:
                    raise

    def _format_record(self, prompt, response):
        return {
            "session_id": self.session_id,
            "timestamp": datetime.now().isoformat(),
            "prompt": prompt,
            "response": response.choices[0].message.content,
            "metadata": {
                "model": response.model,
                "usage": dict(response.usage)
            }
        }

增量导出实现

class IncrementalExporter(ChatGPTExporter):
    def __init__(self, api_key, storage_backend):
        super().__init__(api_key)
        self.storage = storage_backend

    def export_incrementally(self, prompts, batch_size=10):
        """
        批量增量导出
        :param prompts: 提示词列表
        :param batch_size: 每批处理数量
        """
        results = []
        for i in range(0, len(prompts), batch_size):
            batch = prompts[i:i + batch_size]
            for prompt in batch:
                try:
                    record = self.export_conversation(prompt)
                    results.append(record)
                    self.storage.save(record)  # 立即保存
                except Exception as e:
                    print(f"Failed to export prompt'{prompt[:20]}...': {str(e)}")

        return results

性能优化

并发控制策略

1. 令牌桶算法：控制 API 调用频率
2. 指数退避：失败请求的重试间隔
3. 连接池复用：减少 TCP 握手开销

from ratelimit import limits, sleep_and_retry

class OptimizedExporter(ChatGPTExporter):
    # 限制 30 次 / 分钟
    @sleep_and_retry
    @limits(calls=30, period=60)
    def export_conversation(self, prompt):
        return super().export_conversation(prompt)

生产环境常见问题

1. 速率限制：
2. 解决方案：实现请求队列 + 定时器

3. 监控指标：429 错误率

4. 长对话截断：

5. 解决方案：自动拆分大文本

6. 使用 tiktoken 计算 token 数

7. 数据一致性：

8. 实现要点：
   • 每个记录添加唯一 ID
   • 写入前校验
   • 实现幂等操作

安全实践

1. API 密钥管理：
2. 使用环境变量
3. 定期轮换

4. 最小权限原则

5. 数据加密：

6. 传输层：强制 TLS
7. 存储加密：AES-256
8. 敏感信息脱敏

扩展思考

未来可扩展方向：

1. 多 LLM 支持架构
2. 抽象 Provider 接口

3. 统一数据格式

4. 实时导出模式

5. WebSocket 长连接

6. 流式处理

7. 可视化分析

8. 集成 Superset
9. 对话情感分析

完整项目示例可参考 GitHub 仓库：

git clone https://github.com/example/chatgpt-exporter.git