ChatGPT对话记录导出实战指南:从API调用到数据持久化

1次阅读
没有评论

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

image.webp

为什么我们需要导出对话记录?

在工作中,我发现很多团队使用 ChatGPT 进行技术讨论、会议记录或创意生成,但这些有价值的对话往往散落在不同成员的聊天历史中。常见需求场景包括:

ChatGPT 对话记录导出实战指南:从 API 调用到数据持久化

  • 知识管理:将技术问答整理成内部知识库
  • 审计追踪:满足合规性要求的对话记录存档
  • 数据分析:统计高频问题优化客服流程

但手动复制粘贴不仅效率低下,而且无法获取完整的对话上下文。这就是为什么我们需要通过 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]'

进阶思考

增量导出方案

  1. 记录最后导出对话的 created_at 时间戳
  2. 下次请求时添加时间过滤条件:
    params['created_after'] = last_export_time

版本控制设计

建议采用类似 git 的机制:

  • 每个对话生成 SHA256 哈希作为版本 ID
  • 使用轻量级数据库(如 SQLite)存储版本变更
  • 实现差异对比功能显示内容变化

最终建议

对于企业级应用,建议:

  1. 将导出服务封装为定时任务
  2. 添加邮件通知功能
  3. 存储到云存储服务(如 S3)实现自动归档

通过本文介绍的方法,我们团队成功实现了每周自动备份所有技术讨论记录,极大提高了知识管理效率。希望这个方案对您也有所帮助!

正文完
 0
评论(没有评论)