ChatGPT导出功能深度解析：从API调用到数据持久化实战

1次阅读

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

在实际使用 ChatGPT API 进行对话导出时，开发者经常会遇到一些棘手的问题。这些问题不仅影响导出效率，还可能导致数据丢失或格式混乱。

API 限制问题
每个 API 请求都有 token 限制，长对话需要分多次请求
严格的速率限制（RPM 和 TPM）容易触发 429 错误
流式响应 (streaming response) 需要特殊处理
数据结构挑战
原始响应是深度嵌套的 JSON 结构
多轮对话的上下文关联难以维护
角色切换（user/assistant）需要额外标记

针对上述问题，我们设计了一套完整的解决方案：

异步请求处理
使用 aiohttp 替代 requests 库实现并发请求
通过 semaphore 控制并发度避免触发速率限制
数据分块策略
根据 token 计数自动拆分长对话
实现 continuation 标记维持对话连贯性
数据结构设计
使用 message_id 构建对话树
添加 parent_id 字段维护对话线索
采用 thread 模型组织复杂会话

import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def fetch_chat_completion(session, payload):
    headers = {"Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    async with session.post(
        "https://api.openai.com/v1/chat/completions",
        json=payload,
        headers=headers
    ) as response:
        if response.status != 200:
            raise Exception(f"API 请求失败: {response.status}")
        return await response.json()

import pandas as pd

def normalize_messages(conversation):
    records = []
    for msg in conversation['messages']:
        records.append({'message_id': msg['id'],
            'parent_id': msg.get('parent_id'),
            'role': msg['role'],
            'content': msg['content'],
            'timestamp': msg['created_at']
        })
    return pd.DataFrame(records)

def to_markdown(df):
    output = []
    for _, row in df.iterrows():
        output.append(f"## {row['role'].upper()} ({row['timestamp']})")
        output.append(row['content'] + "\n")
    return '\n'.join(output)

监控指标
API 响应时间百分位监控(P99/P95)
请求成功率看板
数据完整性校验（对话轮次连续性检查）
安全防护

使用正则表达式自动过滤敏感信息：

import re
def sanitize_text(text):
    return re.sub(r'\b\d{4}[-]?\d{4}[-]?\d{4}\b', '[REDACTED]', text)

缓存策略
本地 SQLite 缓存已导出对话
使用 message_id 作主键避免重复
设置 TTL 自动清理旧缓存

flowchart TD
    A[开始导出] --> B{对话长度?}
    B -- 短对话 --> C[直接请求 API]
    B -- 长对话 --> D[拆分对话块]
    D --> E[异步并发请求]
    C --> F[接收流式响应]
    E --> F
    F --> G[构建对话树]
    G --> H[持久化存储]
    H --> I[格式转换]