ChatGPT聊天记录导出实战：从API调用到本地存储的完整指南

1次阅读

没有评论

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

在日常开发中，我们经常需要将 ChatGPT 的聊天记录导出保存，比如客服对话归档、AI 训练数据收集等。但直接使用网页版的 ChatGPT，只能手动复制粘贴，效率低下且容易出错。现有的解决方案要么功能单一，要么缺乏灵活性。

比如，有些工具只能导出纯文本，丢失了对话的结构信息；有些则无法处理大量数据的导出，遇到 API 限流就直接崩溃。这些问题让开发者头疼不已。

目前主要有三种方式可以实现聊天记录导出：

直接调用 API：最灵活，适合定制化需求，但需要处理分页、限流等问题
使用 SDK：开发简单，但功能可能受限，且更新可能滞后于 API
Webhook 推送：实时性好，但需要搭建接收服务，复杂度较高

对于大多数 Python 开发者来说，直接调用 API+ 本地处理是最平衡的选择。

首先，我们需要通过 OpenAI 的 ChatCompletion 接口获取完整的对话记录。这里有个关键点：要正确处理分页返回的结果。

import openai
from typing import List, Dict

def get_chat_history(conversation_id: str, max_results: int = 100) -> List[Dict]:
    """
    获取指定会话的完整聊天记录
    :param conversation_id: 会话 ID
    :param max_results: 最大返回数量
    :return: 消息列表
    """
    messages = []
    last_message_id = None

    while len(messages) < max_results:
        params = {
            "conversation_id": conversation_id,
            "limit": min(50, max_results - len(messages))
        }

        if last_message_id:
            params["after"] = last_message_id

        try:
            response = openai.ChatCompletion.list_messages(**params)
            new_messages = response["data"]
            if not new_messages:
                break

            messages.extend(new_messages)
            last_message_id = new_messages[-1]["id"]

        except Exception as e:
            print(f"获取消息失败: {e}")
            break

    return messages[:max_results]

获取到原始 JSON 数据后，我们通常需要转换为更易处理的格式。以下是转换为 CSV 和 Markdown 的示例：

import csv
from datetime import datetime

def to_csv(messages: List[Dict], output_path: str):
    """将消息列表导出为 CSV 文件"""
    with open(output_path, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['时间', '角色', '内容'])

        for msg in messages:
            timestamp = datetime.fromtimestamp(msg['created_at'])
            writer.writerow([timestamp.isoformat(),
                msg['role'],
                msg['content']
            ])

def to_markdown(messages: List[Dict], output_path: str):
    """将消息列表导出为 Markdown 文件"""
    with open(output_path, 'w', encoding='utf-8') as f:
        for msg in messages:
            timestamp = datetime.fromtimestamp(msg['created_at'])
            f.write(f"## {timestamp.isoformat()} {msg['role']}\n\n")
            f.write(f"{msg['content']}\n\n")

建议按日期组织目录结构：

chat_exports/
├── 2023-08-01/
│   ├── conversation_abc123.csv
│   └── conversation_def456.md
├── 2023-08-02/
│   └── ...
└── archive/  # 超过 30 天的存档

对于企业级应用，建议使用数据库存储。以下是 PostgreSQL 的 Schema 设计示例：

CREATE TABLE chat_messages (
    id SERIAL PRIMARY KEY,
    conversation_id VARCHAR(64) NOT NULL,
    message_id VARCHAR(64) NOT NULL UNIQUE,
    role VARCHAR(16) NOT NULL,
    content TEXT NOT NULL,
    created_at TIMESTAMP NOT NULL,
    metadata JSONB
);

CREATE INDEX idx_conversation ON chat_messages(conversation_id);
CREATE INDEX idx_created ON chat_messages(created_at);

OpenAI API 有严格的速率限制，建议：

实现指数退避重试机制
监控剩余配额
对大任务进行分批处理

导出前应该对敏感信息进行处理：

def sanitize_content(text: str) -> str:
    """脱敏处理"""
    # 替换邮箱
    text = re.sub(r'[\w.+-]+@[\w-]+\.[\w.-]+', '[EMAIL]', text)
    # 替换电话号码
    text = re.sub(r'\b\d{3}[.-]?\d{3}[.-]?\d{4}\b', '[PHONE]', text)
    return text

对于大量数据导出，建议使用异步任务队列：

from celery import Celery

app = Celery('chat_export')

@app.task(bind=True)
def export_conversation(self, conversation_id):
    try:
        messages = get_chat_history(conversation_id)
        # 处理导出逻辑...
    except Exception as e:
        self.retry(exc=e, countdown=60)