共计 2605 个字符,预计需要花费 7 分钟才能阅读完成。
背景痛点
最近在整理 ChatGPT 的对话记录时,发现网页端的管理功能非常有限。每次想回顾之前的对话,要么得手动一页页翻找,要么只能复制粘贴到本地文档。这种操作方式存在几个明显问题:

- 跨会话追踪困难:ChatGPT 的对话记录是按时间倒序排列的,当对话数量多时,很难找到特定的某次对话
- 格式混乱:直接复制的对话内容会丢失原有的 Markdown 格式,代码块和表格尤其明显
- 效率低下:对于长期使用 ChatGPT 的用户,手动导出上百条对话记录几乎是不可能完成的任务
技术方案对比
在寻找解决方案时,我主要考察了三种方法:
- 官方导出工具:ChatGPT 目前没有提供完整的对话导出功能,只能单次复制
- 第三方爬虫:虽然能实现批量导出,但存在账号封禁风险,且违反服务条款
- OpenAI API 方案:完全合规,通过官方提供的接口获取数据,稳定性有保障
最终选择了 API 方案,不仅因为它的合法性,还因为它提供了更灵活的后续处理能力。根据 OpenAI 的使用政策,通过 API 获取自己的对话记录是完全允许的。
核心实现
1. 会话链获取
OpenAI API 提供了 conversation_id 来追踪完整的对话链。以下是核心代码片段:
import openai
# 初始化客户端
openai.api_key = 'your-api-key'
# 获取会话列表
def get_conversations(limit=1000):
# limit=1000 是 API 允许的单次请求最大值
return openai.ChatCompletion.list(
limit=limit,
order='desc' # 按时间倒序
)
2. 异步请求处理
为了提高效率,使用 aiohttp 实现异步请求:
import aiohttp
import asyncio
async def fetch_conversation(session, conversation_id):
url = f'https://api.openai.com/v1/conversations/{conversation_id}'
headers = {'Authorization': f'Bearer {API_KEY}'}
try:
async with session.get(url, headers=headers) as response:
if response.status == 200:
return await response.json()
# 处理 429/503 错误
elif response.status in [429, 503]:
await asyncio.sleep(2)
return await fetch_conversation(session, conversation_id)
else:
response.raise_for_status()
except Exception as e:
print(f"Error fetching {conversation_id}: {e}")
return None
3. API 密钥轮换
对于高频请求,建议实现密钥轮换机制:
API_KEYS = ['key1', 'key2', 'key3'] # 多个 API 密钥
current_key_index = 0
def get_current_key():
global current_key_index
key = API_KEYS[current_key_index]
current_key_index = (current_key_index + 1) % len(API_KEYS)
return key
进阶优化
1. 增量导出
使用 Message Delta 可以只获取新增的对话内容:
def get_updates(last_message_id):
params = {'after': last_message_id}
return openai.ChatCompletion.list(**params)
2. Markdown 特殊处理
对话中的代码块和表格需要特殊转义:
import re
def escape_markdown(text):
# 处理代码块
text = re.sub(r'```(.*?)```', r'\`\`\`\1\`\`\`', text, flags=re.DOTALL)
# 处理表格
text = re.sub(r'\|', r'\\|', text)
return text
3. 性能调优
经过测试,建议将请求频率控制在每秒 3 - 5 次。OpenAI API 的速率限制较为严格,过度请求会导致 429 错误。
避坑指南
- 自动重试策略:对于 429/503 错误,采用指数退避算法
import random
def exponential_backoff(retries):
delay = min(2 ** retries + random.uniform(0, 1), 60)
await asyncio.sleep(delay)
- 敏感信息过滤:使用正则表达式过滤 API 密钥等敏感信息
SENSITIVE_PATTERNS = [r'sk-[a-zA-Z0-9]{24}', # API 密钥
r'\b\d{4}[-]?\d{4}[-]?\d{4}[-]?\d{4}\b' # 信用卡
]
def filter_sensitive(text):
for pattern in SENSITIVE_PATTERNS:
text = re.sub(pattern, '[REDACTED]', text)
return text
- 本地缓存:避免重复请求相同对话
import pickle
import os
CACHE_FILE = 'conversations.cache'
def load_cache():
if os.path.exists(CACHE_FILE):
with open(CACHE_FILE, 'rb') as f:
return pickle.load(f)
return set()
def save_cache(cache):
with open(CACHE_FILE, 'wb') as f:
pickle.dump(cache, f)
延伸思考
如何实现跨设备会话同步?可以考虑以下思路:
- 使用云存储(如 Dropbox)自动同步导出的对话文件
- 开发一个简单的 Web 服务,集中管理所有设备的对话记录
- 利用 OpenAI 的 fine-tuning 功能,将重要对话提炼成知识库
这个方案在我实际使用中效果很好,从原来手动处理每小时只能导出 10-20 条对话,到现在可以轻松管理上千条记录。希望对有同样需求的朋友有所帮助。
正文完
