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

1次阅读

没有评论

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

在企业和开发者使用 ChatGPT 的过程中，对话记录导出是一个常见需求。主要应用场景包括：

客服审计 ：分析客服对话质量，优化服务流程
知识库构建 ：将高质量对话整理为结构化知识
用户行为分析 ：研究用户提问模式和偏好
合规存档 ：满足数据留存的法律要求

OpenAI 提供了两种获取对话记录的方式：

官方 Dashboard 导出
优点：无需编程，一键导出
缺点：仅限最近数据，无法批量筛选
限制：最多导出 1000 条记录
API 调用导出
优点：可获取完整历史记录
优点：支持条件筛选和自定义字段
缺点：需要开发实现

首先安装必要的 Python 包：

pip install openai pandas sqlalchemy

使用 OpenAI Python SDK 初始化客户端并获取对话列表：

import openai
from datetime import datetime

# 初始化客户端
openai.api_key = 'your-api-key'

# 获取对话列表
def get_conversations(limit=100, before=None):
    params = {'limit': limit,}
    if before:
        params['before'] = before

    return openai.ChatCompletion.list(**params)

由于 API 有单次请求条数限制，需要实现分页获取：

def export_all_conversations(output_file='conversations.csv'):
    all_conversations = []
    last_id = None

    while True:
        # 获取当前页数据
        response = get_conversations(before=last_id)
        conversations = response['data']

        if not conversations:
            break

        # 记录最后一条的 ID 用于下一页
        last_id = conversations[-1]['id']
        all_conversations.extend(conversations)

        print(f'已获取 {len(all_conversations)} 条记录')

    # 保存到 CSV
    import pandas as pd
    df = pd.DataFrame(all_conversations)
    df.to_csv(output_file, index=False)

将数据保存到 SQLite 数据库：

from sqlalchemy import create_engine

def save_to_sqlite(conversations, db_file='conversations.db'):
    engine = create_engine(f'sqlite:///{db_file}')

    # 转换数据结构
    records = []
    for conv in conversations:
        records.append({'id': conv['id'],
            'created': datetime.fromtimestamp(conv['created']),
            'model': conv['model'],
            'content': conv['choices'][0]['message']['content']
        })

    # 保存到数据库
    import pandas as pd
    df = pd.DataFrame(records)
    df.to_sql('conversations', engine, if_exists='append', index=False)

避免将 API 密钥硬编码在代码中，推荐做法：

使用环境变量
密钥管理服务（如 AWS KMS）
配置文件（.env）+ gitignore

import os
from dotenv import load_dotenv

load_dotenv()  # 加载.env 文件
openai.api_key = os.getenv('OPENAI_API_KEY')

对含敏感信息的对话内容进行加密：

from cryptography.fernet import Fernet

# 生成密钥
key = Fernet.generate_key()
cipher = Fernet(key)

# 加密函数
def encrypt_content(content):
    return cipher.encrypt(content.encode()).decode()

# 解密函数
def decrypt_content(encrypted_content):
    return cipher.decrypt(encrypted_content.encode()).decode()

使用异步请求提高获取速度：

import aiohttp
import asyncio

async def fetch_conversations(session, before=None):
    params = {'limit': 100}
    if before:
        params['before'] = before

    async with session.get(
        'https://api.openai.com/v1/conversations',
        headers={'Authorization': f'Bearer {openai.api_key}'},
        params=params
    ) as response:
        return await response.json()

实现简单的本地缓存机制：

import json
import os

CACHE_FILE = 'conversations_cache.json'

# 读取缓存
def load_cache():
    if os.path.exists(CACHE_FILE):
        with open(CACHE_FILE, 'r') as f:
            return json.load(f)
    return []

# 保存缓存
def save_cache(data):
    with open(CACHE_FILE, 'w') as f:
        json.dump(data, f)