ChatGPT对话导出技术解析：从API调用到数据持久化实战

1次阅读

没有评论

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

在实际开发中，导出 ChatGPT 对话数据的需求逐渐增多。常见的场景包括：

审计需求 ：企业需要定期审查员工与 AI 的交互记录，确保合规性
训练数据收集 ：开发者希望用历史对话数据微调自己的模型
用户数据备份 ：为满足 GDPR 等数据法规要求，需提供数据导出功能

但直接通过循环调用 OpenAI API 会遇到几个关键问题：

速率限制 (Rate Limiting)：官方 API 有严格的请求频率控制（如 3,500 次 / 分钟）
数据完整性 ：简单分页可能导致重复或遗漏记录
稳定性 ：网络波动或 API 异常会导致导出任务中断

同步请求 ：
实现简单，适合小批量数据
吞吐量低，受限于网络延迟

代码示例：

import requests

def sync_fetch(url):
    response = requests.get(url)
    return response.json()

异步处理 ：
使用 asyncio/aiohttp 实现并发
吞吐量可提升 5 -10 倍
需要处理更复杂的错误恢复

代码框架：

import aiohttp

async def async_fetch(session, url):
    async with session.get(url) as response:
        return await response.json()

维度	官方 SDK	裸 REST API
开发效率	✅ 高阶封装	❌ 需要自己处理细节
灵活性	❌ 功能受限	✅ 完全控制请求流程
更新时效	❌ 可能滞后新特性	✅ 即时使用最新 API

from tenacity import retry, stop_after_attempt, wait_exponential
import openai

@retry(stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30)
)
def get_chat_history(session_id: str) -> list[dict]:
    try:
        return openai.ChatCompletion.list(
            session_id=session_id,
            limit=100
        ).data
    except openai.error.APIError as e:
        print(f"API Error: {e}")
        raise

from typing import Iterator
from datetime import datetime

def export_conversations(
    start_date: datetime,
    last_id: str = None
) -> Iterator[dict]:
    params = {
        'limit': 100,
        'created_after': int(start_date.timestamp())
    }
    if last_id:
        params['after'] = last_id

    while True:
        batch = openai.ChatCompletion.list(**params).data
        if not batch:
            break

        yield from batch
        params['after'] = batch[-1].id

使用令牌桶算法控制请求速率
分布式环境需用 Redis 维护全局计数器

动态调整策略示例：

from redis import Redis

def check_rate_limit(api_key: str) -> bool:
    r = Redis()
    key = f"rate_limit:{api_key}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, 60)
    return current <= 60  # 60 次 / 分钟

删除个人身份信息 (PII)
使用正则替换敏感词

示例实现：

import re

def anonymize_text(text: str) -> str:
    # 替换邮箱
    text = re.sub(r'[\w\.-]+@[\w\.-]+\.\w+', '[EMAIL]', text)
    # 替换电话号码
    text = re.sub(r'\+?[\d\s-]{10,}', '[PHONE]', text)
    return text

设置合理的 User-Agent
包含请求来源标识

示例：

headers = {'User-Agent': 'DataExporter/1.0 (company.com)',
    'X-Request-Source': 'batch-export'
}

使用生成器替代列表存储
分批写入磁盘

采用流式 JSON 处理：

import json
import gzip

def stream_save(iter_data, filename: str):
    with gzip.open(filename, 'wt') as f:
        for item in iter_data:
            f.write(json.dumps(item) + '\n')

通过本文介绍的方法，可以实现安全高效的 ChatGPT 对话导出。以下是三个值得深入探讨的方向：