ChatGPT聊天归档实战指南：从数据管理到价值挖掘

1次阅读

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

模型优化 ：历史对话是训练 AI 的黄金数据，通过归档可持续改进模型响应质量
用户分析 ：对话模式分析能揭示用户真实需求，指导产品设计迭代
合规审计 ：满足数据留存要求，为争议场景提供完整沟通凭证

自研方案
优势：完全可控，可深度定制存储结构和检索逻辑
挑战：需自行处理分页、限流、错误重试等边缘 case
LangChain 框架
优势：内置对话管理模块，快速实现基础功能
限制：灵活性较差，高级功能仍需二次开发

建议：中小项目用 LangChain 快速验证，日对话量超 1 万条时考虑自研

import openai
from datetime import datetime
import backoff  # 指数退避重试库

@backoff.on_exception(backoff.expo, openai.error.RateLimitError)
def archive_conversations(api_key, user_id):
    """
    自动归档用户历史对话
    :param api_key: OpenAI API 密钥
    :param user_id: 用户唯一标识
    """
    openai.api_key = api_key
    conversations = []
    after_id = None  # 分页游标

    while True:
        # 获取对话列表（含分页处理）resp = openai.ChatCompletion.list(
            user=user_id,
            limit=100,
            after=after_id
        )

        if not resp.data:
            break

        # 结构化存储关键字段
        for chat in resp.data:
            conversations.append({
                "chat_id": chat.id,
                "created": datetime.fromtimestamp(chat.created),
                "tokens": chat.usage.total_tokens,
                "messages": [{"role": m.role, "content": m.content} 
                    for m in chat.messages
                ]
            })

        after_id = resp.data[-1].id  # 更新分页游标

    return conversations

from flask import Flask, request, jsonify
from functools import wraps
import jwt

app = Flask(__name__)
SECRET_KEY = "your_jwt_secret"

# 模拟数据库
conversation_db = {}

def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('Authorization')
        if not token:
            return jsonify({"error": "Token 缺失"}), 403
        try:
            data = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        except:
            return jsonify({"error": "Token 无效"}), 403
        return f(*args, **kwargs)
    return decorated

@app.route('/chats/<user_id>', methods=['GET'])
@token_required
def get_chats(user_id):
    """
    分页查询用户对话记录
    参数说明:
    - page: 页码（默认 1）- size: 每页条数（默认 20）"""page = int(request.args.get('page', 1))
    size = int(request.args.get('size', 20))

    chats = conversation_db.get(user_id, [])
    total = len(chats)

    # 内存分页（生产环境用数据库分页）result = chats[(page-1)*size : page*size]

    return jsonify({
        "data": result,
        "pagination": {
            "total": total,
            "page": page,
            "page_size": size
        }
    })

{"_id": ObjectId("..."),
  "user_id": "user123",
  "session_id": "session456",
  "created_at": ISODate("..."),
  "messages": [
    {
      "role": "user",
      "content": "如何学习 Python？",
      "timestamp": ISODate("...")
    },
    {
      "role": "assistant",
      "content": "建议从基础语法开始...",
      "timestamp": ISODate("...")
    }
  ],
  "metadata": {
    "token_count": 42,
    "topics": ["教育", "编程"]
  }
}

# elasticsearch.yml 关键配置
index:
  analysis:
    analyzer:
      chat_analyzer:
        type: custom
        tokenizer: ik_smart  # 中文智能分词
        filter: [lowercase, synonym_filter]
    filter:
      synonym_filter:
        type: synonym
        synonyms_path: analysis/synonyms.txt