ChatGPT聊天记录归档与检索实战指南：从存储到高效查询

1次阅读

没有评论

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

在日常使用 ChatGPT API 进行开发时，聊天记录的管理往往成为开发者头疼的问题。原始日志存储方式通常存在以下痛点：

性能瓶颈：随着聊天记录量的增长，线性查找效率急剧下降
数据管理困难：缺乏结构化存储，难以实现高效检索
存储成本高：未压缩的原始日志占用大量空间

这些痛点在实际开发中会导致查询响应慢、存储成本高、运维复杂度增加等问题。

针对聊天记录的特点（半结构化、高频写入、按时间查询为主），我们对几种常见存储方案进行了对比：

关系型数据库(MySQL)
优点：事务支持完善，查询功能强大
缺点：Schema 固定，扩展性较差
文档数据库(MongoDB)
优点：灵活的数据模型，适合存储 JSON 格式的聊天记录
缺点：分布式部署复杂度高
对象存储(S3)
优点：存储成本低，扩展性强
缺点：查询能力有限

综合考虑后，我们采用 MongoDB 作为主存储，S3 用于长期归档的混合方案。

使用 Python 的 logging 模块实现结构化日志记录，关键配置如下：

import logging
import json
from datetime import datetime

class StructuredLogger:
    def __init__(self, name):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)

        handler = logging.StreamHandler()
        formatter = JSONFormatter()
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)

    def log_chat(self, conversation_id, user_input, ai_response):
        log_entry = {'timestamp': datetime.utcnow().isoformat(),
            'conversation_id': conversation_id,
            'user_input': user_input,
            'ai_response': ai_response
        }
        self.logger.info(log_entry)

class JSONFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps(record.msg)

在 MongoDB 中创建复合索引，优化查询性能：

# 创建索引
from pymongo import MongoClient, ASCENDING

db = MongoClient('mongodb://localhost:27017/').chat_logs

db.conversations.create_index([('conversation_id', ASCENDING),
    ('timestamp', ASCENDING)
])

使用 Python 的 croniter 库实现定时归档任务：

import os
from croniter import croniter
from datetime import datetime
import boto3

class ArchiveManager:
    def __init__(self, cron_expression):
        self.cron = croniter(cron_expression)
        self.s3 = boto3.client('s3')

    def should_archive(self):
        return datetime.now() >= self.cron.get_next(datetime)

    def archive_conversations(self, db, bucket_name):
        cutoff_date = datetime.now() - timedelta(days=30)
        old_conversations = db.conversations.find({'timestamp': {'$lt': cutoff_date}
        })

        # 压缩并上传到 S3
        for conv in old_conversations:
            filename = f"{conv['conversation_id']}.json.gz"
            with gzip.open(filename, 'wt') as f:
                json.dump(conv, f)

            self.s3.upload_file(filename, bucket_name, filename)
            db.conversations.delete_one({'_id': conv['_id']})