ChatGPT聊天记录导出技术解析：从API调用到数据持久化

1次阅读

没有评论

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

手动导出 ChatGPT 对话记录存在几个明显问题：

每次只能单条导出，无法批量操作
历史记录超过一定时间后难以追溯
缺乏结构化存储，不利于后续分析

对于开发者或企业用户，需要将对话记录用于审计、训练或分析时，自动化导出成为刚需。

优点：

合法合规，稳定性高
直接获取结构化数据
支持完整的对话元数据

缺点：

有 API 调用限制
需要处理分页逻辑

优点：

不依赖官方接口

缺点：

违反服务条款
解析 HTML 结构脆弱
无法获取完整对话上下文

结论：生产环境强烈建议使用官方 API。

import openai
import os
from dotenv import load_dotenv
import json
from datetime import datetime
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 加载环境变量
load_dotenv()
openai.api_key = os.getenv('OPENAI_API_KEY')

def get_conversations(limit=100):
    try:
        conversations = []
        has_more = True
        after = None

        while has_more and len(conversations) < limit:
            params = {"limit": min(20, limit - len(conversations)),
            }
            if after:
                params["after"] = after

            response = openai.ChatCompletion.list(**params)
            conversations.extend(response["data"])
            has_more = response["has_more"]
            after = response["data"][-1]["id"] if has_more else None

            # 遵守速率限制
            time.sleep(0.5)

        return conversations
    except Exception as e:
        logger.error(f"获取对话列表失败: {str(e)}")
        raise

def export_conversation(conversation_id, output_dir='exports'):
    try:
        os.makedirs(output_dir, exist_ok=True)
        messages = []

        # 获取完整对话历史
        response = openai.ChatCompletion.retrieve(conversation_id)
        messages = response["messages"]

        # 处理可能的消息截断
        while len(messages) < response["message_count"]:
            last_id = messages[-1]["id"]
            more_messages = openai.ChatCompletion.retrieve(
                conversation_id,
                before_message=last_id
            )
            messages.extend(more_messages["messages"])

        # 结构化存储
        filename = f"{output_dir}/{conversation_id}.json"
        with open(filename, 'w') as f:
            json.dump({
                "metadata": {"created_at": response["created_at"],
                    "updated_at": response["updated_at"],
                    "title": response["title"]
                },
                "messages": messages
            }, f, indent=2)

        logger.info(f"成功导出对话: {conversation_id}")
        return filename
    except Exception as e:
        logger.error(f"导出对话 {conversation_id} 失败: {str(e)}")
        raise

对于大规模导出（>1000 条对话）：

采用分批处理，每 100 条保存一次
使用生成器而非列表保存中间结果
考虑使用 SQLite 临时存储

import aiohttp
import asyncio

async def async_export(conversation_ids, output_dir):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for cid in conversation_ids:
            task = asyncio.create_task(fetch_conversation(session, cid, output_dir)
            )
            tasks.append(task)
        await asyncio.gather(*tasks)

永远不要硬编码 API 密钥
使用环境变量或密钥管理服务
设置密钥的细粒度权限

检查 message_count 与实际获取数量的差异
使用 before_message 参数获取历史消息
注意 OpenAI API 对单次返回消息数的限制

导出前获取用户同意
匿名化处理敏感信息
遵守 GDPR 等数据保护法规

导出的数据可用于：

微调自定义模型
构建对话知识库
用户行为分析
客服质量评估

# 示例：准备微调数据
def prepare_fine_tuning_data(export_dir):
    training_data = []
    for filename in os.listdir(export_dir):
        with open(f"{export_dir}/{filename}") as f:
            conv = json.load(f)
            for i in range(len(conv["messages"])-1):
                if conv["messages"][i]["role"] == "user":
                    training_data.append({"prompt": conv["messages"][i]["content"],
                        "completion": conv["messages"][i+1]["content"]
                    })
    return training_data

通过 OpenAI 官方 API 导出对话记录是最可靠的方式。关键点在于：