ChatGPT聊天归档后找不到了？手把手教你搭建本地对话存储系统

1次阅读

没有评论

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

许多开发者在使用 ChatGPT API 时都会遇到一个共同的烦恼：对话一旦结束，就像被扔进了黑洞，再也找不回来。这其实是因为 ChatGPT API 默认采用无状态设计——每次请求都是独立的，服务器不会保存你的聊天记录。

这种机制带来了几个实际问题：

技术讨论中断 ：当你在调试代码时与 ChatGPT 进行了多轮对话，第二天想继续讨论却发现上下文全无，不得不重新描述问题
知识管理困难 ：ChatGPT 给出的有价值的技术方案或学习资源，因为没有保存而无法后续查阅
效率降低 ：重复解释相同问题浪费大量时间，尤其在进行长期项目开发时

要解决这个问题，我们需要一个本地存储方案。常见的选择有：

内存缓存 ：速度快但程序重启就丢失，不适合长期保存
文件存储 ：简单但难以实现复杂查询
数据库 ：功能全面但多数需要独立服务

SQLite 脱颖而出成为最佳选择，因为：

零配置，单个文件即可运行
支持标准 SQL 语法
内置在 Python 标准库中
能处理 GB 级数据，完全满足个人使用

首先创建存储对话的表结构：

import sqlite3

def init_db(db_path='chat_history.db'):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS conversations (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
        role TEXT NOT NULL,  -- 'user' or 'assistant'
        content TEXT NOT NULL,
        session_id TEXT NOT NULL,
        tags TEXT  -- 逗号分隔的标签
    )
    ''')
    conn.commit()
    conn.close()

创建数据库操作类，注意使用参数化查询防止 SQL 注入：

class ChatDB:
    def __init__(self, db_path='chat_history.db'):
        self.db_path = db_path

    def _get_conn(self):
        return sqlite3.connect(self.db_path)

    def save_message(self, role, content, session_id, tags=None):
        with self._get_conn() as conn:
            conn.execute("INSERT INTO conversations(role, content, session_id, tags) VALUES(?,?,?,?)",
                (role, content, session_id, tags)
            )

    def get_session_messages(self, session_id, limit=100):
        with self._get_conn() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT role, content FROM conversations WHERE session_id=? ORDER BY timestamp DESC LIMIT ?",
                (session_id, limit)
            )
            return cursor.fetchall()

在调用 API 时自动保存对话：

import openai

class ChatSession:
    def __init__(self, api_key, session_id):
        openai.api_key = api_key
        self.session_id = session_id
        self.db = ChatDB()

    def chat(self, prompt):
        # 保存用户消息
        self.db.save_message('user', prompt, self.session_id)

        try:
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}]
            )
            reply = response['choices'][0]['message']['content']

            # 保存 AI 回复
            self.db.save_message('assistant', reply, self.session_id)
            return reply

        except Exception as e:
            print(f"API 调用失败: {str(e)}")
            raise

为常用查询字段添加索引：

# 在 init_db 函数中追加
cursor.execute("CREATE INDEX IF NOT EXISTS idx_session ON conversations(session_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON conversations(timestamp)")

定期将旧对话移动到归档表：

def archive_old_messages(months=3):
    with sqlite3.connect('chat_history.db') as conn:
        conn.execute("""
        INSERT INTO archived_conversations
        SELECT * FROM conversations 
        WHERE timestamp < datetime('now', ?)
        """, (f'-{months} months',))

        conn.execute("""
        DELETE FROM conversations 
        WHERE timestamp < datetime('now', ?)
        """, (f'-{months} months',))
        conn.commit()

使用 SQLite 的加密扩展或 Python 加密库：

from cryptography.fernet import Fernet

# 生成密钥（首次运行时执行一次并妥善保存）# key = Fernet.generate_key() 

class EncryptedChatDB(ChatDB):
    def __init__(self, db_path, encryption_key):
        super().__init__(db_path)
        self.cipher = Fernet(encryption_key)

    def save_message(self, role, content, session_id, tags=None):
        encrypted_content = self.cipher.encrypt(content.encode())
        super().save_message(role, encrypted_content, session_id, tags)

    def get_session_messages(self, session_id, limit=100):
        messages = super().get_session_messages(session_id, limit)
        return [(role, self.cipher.decrypt(content).decode())
            for role, content in messages
        ]