共计 2604 个字符,预计需要花费 7 分钟才能阅读完成。
为什么我们需要导出对话记录?
在工作中,我发现很多团队使用 ChatGPT 进行技术讨论、会议记录或创意生成,但这些有价值的对话往往散落在不同成员的聊天历史中。常见需求场景包括:

- 知识管理:将技术问答整理成内部知识库
- 审计追踪:满足合规性要求的对话记录存档
- 数据分析:统计高频问题优化客服流程
但手动复制粘贴不仅效率低下,而且无法获取完整的对话上下文。这就是为什么我们需要通过 API 实现自动化导出。
技术方案选型
方案对比表
| 维度 | 官方 API 方案 | 第三方工具库 |
|---|---|---|
| 数据完整性 | 100% 官方支持 | 可能缺失最新字段 |
| 速率限制 | 每分钟 3 -60 次(取决于套餐) | 受限于工具封装逻辑 |
| 历史记录追溯 | 支持游标分页 | 通常只获取最新记录 |
| 维护成本 | 需自行处理错误重试 | 内置重试机制 |
经过实践验证,我推荐直接使用官方 API,虽然需要自己处理分页逻辑,但能获得最可靠的数据源。
核心代码实现
环境准备
首先安装必要的库:
pip install openai pandas
基础数据获取
import openai
from typing import List, Dict, Optional
import pandas as pd
# 建议将 API_KEY 放在环境变量中
openai.api_key = os.getenv('OPENAI_API_KEY')
def list_conversations(limit: int = 100) -> List[Dict]:
"""
获取对话列表(自动处理分页):param limit: 最大获取数量
:return: 对话记录列表
"""
all_conversations = []
cursor = None
while len(all_conversations) < limit:
params = {'limit': min(50, limit - len(all_conversations)),
'order': 'desc' # 从最新开始获取
}
if cursor:
params['after'] = cursor
try:
response = openai.ChatCompletion.list(**params)
batch = response.get('data', [])
all_conversations.extend(batch)
if not response.get('has_more'):
break
cursor = response['last_id']
except Exception as e:
print(f"获取失败: {str(e)}")
break
return all_conversations[:limit]
数据导出实战
def export_to_csv(conversations: List[Dict], filename: str):
"""
导出对话记录到 CSV
:param conversations: 对话数据
:param filename: 输出文件名
"""
# 数据扁平化处理
rows = []
for conv in conversations:
for msg in conv.get('messages', []):
rows.append({'conversation_id': conv['id'],
'create_time': conv['created_at'],
'role': msg['role'],
'content': msg['content'][:500] # 防止超长文本
})
df = pd.DataFrame(rows)
df.to_csv(filename, index=False)
print(f"成功导出 {len(df)} 条记录到 {filename}")
# 使用示例
if __name__ == "__main__":
conversations = list_conversations(200)
export_to_csv(conversations, "chatgpt_dialogs.csv")
生产环境注意事项
1. API 配额优化
- 使用
time.sleep()控制请求频率(建议每分钟不超过 20 次) - 优先获取最近对话(设置
order='desc') - 合理设置
limit参数减少请求次数
2. 异常处理增强版
from tenacity import retry, stop_after_attempt, wait_exponential
import time
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def safe_api_call(params):
"""带自动重试的 API 调用"""
try:
response = openai.ChatCompletion.list(**params)
if 'error' in response:
raise ValueError(response['error'])
return response
except Exception as e:
print(f"API 调用失败: {e}")
time.sleep(5) # 冷却时间
raise
3. 敏感信息过滤
建议在导出前进行内容扫描:
def contains_sensitive(text: str) -> bool:
"""简单敏感词检测示例"""
sensitive_terms = ['password', '密钥', 'token']
return any(term.lower() in text.lower() for term in sensitive_terms)
# 在导出循环中添加检查
for msg in conv.get('messages', []):
if contains_sensitive(msg['content']):
msg['content'] = '[REDACTED]'
进阶思考
增量导出方案
- 记录最后导出对话的
created_at时间戳 - 下次请求时添加时间过滤条件:
params['created_after'] = last_export_time
版本控制设计
建议采用类似 git 的机制:
- 每个对话生成 SHA256 哈希作为版本 ID
- 使用轻量级数据库(如 SQLite)存储版本变更
- 实现差异对比功能显示内容变化
最终建议
对于企业级应用,建议:
- 将导出服务封装为定时任务
- 添加邮件通知功能
- 存储到云存储服务(如 S3)实现自动归档
通过本文介绍的方法,我们团队成功实现了每周自动备份所有技术讨论记录,极大提高了知识管理效率。希望这个方案对您也有所帮助!
正文完
