ChatGPT聊天归档实战：从零搭建高效对话存储系统

1次阅读

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

在开发基于 ChatGPT 的应用时，对话历史的管理常常被忽视，但实际业务中却至关重要。它不仅有助于审计和合规需求，还能实现跨会话的上下文延续，提升用户体验。本文将带你从零构建一个轻量级但高效的聊天归档系统。

SQLite：
优点：零配置、单文件部署，适合嵌入式场景；支持 ACID 事务；WAL 模式优化并发写入
缺点：无网络访问能力，适合本地存储
MySQL：
优点：成熟的关系型数据库，支持复杂查询
缺点：需要单独部署，对简单归档场景略显笨重
MongoDB：
优点：文档模型天然适合 JSON 消息存储
缺点：内存占用高，对简单查询性能不如关系型数据库

综合来看，SQLite 以其轻量级和足够的功能成为本场景的最佳选择。

{
    "timestamp": "2023-05-20T14:30:00Z",  # ISO8601 格式
    "session_id": "abcd1234",           # 会话唯一标识
    "role": "user|assistant",           # 消息角色
    "content": "Hello, world!",         # 消息内容
    "meta": {                            # 扩展元数据
        "model": "gpt-4",
        "tokens": 15
    }
}

使用 SHA256 对消息内容生成指纹
查询数据库是否存在相同指纹
仅存储新消息

import sqlite3

conn = sqlite3.connect('chat_archive.db')
conn.execute('PRAGMA journal_mode=WAL')  # 启用 WAL 模式

import sqlite3
import json
from datetime import datetime
import hashlib

class ChatArchiver:
    def __init__(self, db_path='chat_archive.db'):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute('''CREATE TABLE IF NOT EXISTS messages
                            (id INTEGER PRIMARY KEY AUTOINCREMENT,
                             session_id TEXT NOT NULL,
                             timestamp TEXT NOT NULL,
                             role TEXT NOT NULL,
                             content TEXT NOT NULL,
                             meta TEXT,
                             content_hash TEXT UNIQUE)''')

        # 创建索引
        self.conn.execute('CREATE INDEX IF NOT EXISTS idx_session ON messages(session_id)')
        self.conn.execute('CREATE INDEX IF NOT EXISTS idx_timestamp ON messages(timestamp)')

        # 启用 WAL 模式
        self.conn.execute('PRAGMA journal_mode=WAL')

    def _generate_hash(self, content):
        return hashlib.sha256(content.encode('utf-8')).hexdigest()

    def add_message(self, session_id, role, content, meta=None):
        timestamp = datetime.utcnow().isoformat() + 'Z'
        content_hash = self._generate_hash(content)
        meta_json = json.dumps(meta) if meta else None

        try:
            self.conn.execute(
                '''INSERT INTO messages 
                (session_id, timestamp, role, content, meta, content_hash)
                VALUES (?, ?, ?, ?, ?, ?)''',
                (session_id, timestamp, role, content, meta_json, content_hash)
            )
            self.conn.commit()
            return True
        except sqlite3.IntegrityError:  # 处理重复消息
            return False

    def get_messages(self, session_id, limit=100):
        cursor = self.conn.execute(
            '''SELECT timestamp, role, content, meta 
               FROM messages 
               WHERE session_id = ? 
               ORDER BY timestamp ASC
               LIMIT ?''',
            (session_id, limit)
        )
        return [{'timestamp': row[0],
            'role': row[1],
            'content': row[2],
            'meta': json.loads(row[3]) if row[3] else None
        } for row in cursor]