最近和几个做 AI 应用的朋友聊天,大家不约而同都在抱怨一件事:和不同 AI 助手的对话记录散落得到处都是,想回头翻一个几周前让 AI 帮忙设计的接口方案,却怎么都找不到。换一个工具,历史对话就归零;换一台电脑,本地记录就丢失。更别提那些有价值的 prompt 调试过程、代码生成思路、问题排查问答,全都躺在各自平台的对话框里,变成一座座无法检索的数据孤岛。
所以就有了这篇文章的标题:重铸 AI 聊天荣光。并不是说 AI 聊天本身出了问题,而是想让那些被“聊”出来的内容真正沉淀下来,变成可管理、可检索、可复用的资产。我决定自己动手写一个轻量级的聊天记录归档工具,命名为ChatArchive。
这篇文章会完整记录 ChatArchive 的设计思路和实现过程。文章会从需求分析讲起,再逐步拆解数据模型、存储方案、导入导出逻辑,最后给出一个可运行的完整项目。如果你正在做 AI 应用开发、Prompt 工程,或者单纯想管理自己的 AI 对话历史,这篇文章应该能给你不少启发。
1. 为什么需要一个 ChatArchive
1.1 分散的 AI 对话记录:这个痛点真实存在
现在的 AI 聊天场景已经非常丰富。工作上有 AI 编程助手,学习上有大模型问答平台,生活里还有各种垂直领域的 ChatBot。你可能会在同一个上午,先和某个模型讨论接口设计,再去另一个工具里让它生成 SQL,晚上又换了一个助手做文本润色。
问题也随之而来:
- 对话记录跨平台分散,没有统一入口。
- 平台一旦调整策略,历史记录可能无法导出。
- 本地缓存清理后,聊天上下文直接丢失。
- 想做 Prompt 复盘或数据沉淀,却连原始数据都找不到。
这些问题在个人使用场景下只是“有点麻烦”,但在 AI 应用开发、数据标注、Prompt 评测等工程场景里,就是实打实的效率瓶颈。试想一下,你需要准备一批多轮对话样本来评估模型效果,结果数据散落在 5 个平台上,格式各不相同,字段口径也不一样,那光是数据清洗就足以让人崩溃。
1.2 ChatArchive 是什么
ChatArchive 是一个面向 AI 聊天场景的记录归档系统。它的核心目标不是替代现有聊天工具,而是做“聊天记录的中转站和仓库”:
- 统一存储:把不同 AI 平台的对话历史,统一转换成一套通用的数据模型。
- 本地优先:数据默认存储在你的本机 SQLite 数据库中,不强制上传云端。
- 快速检索:通过关键词、会话 ID、角色类型等维度,快速找到某一条历史消息。
- 灵活导出:支持将查询结果导出为 Markdown、CSV 等通用格式,方便分享、备份和二次加工。
从工程角度看,ChatArchive 本质上是一个“数据管道”:入口是各种异构的聊天记录,出口是结构化、可查询、可导出的统一数据文件。中间的核心是数据模型设计和存储实现。
1.3 适合谁来用
- AI 应用开发者:需要积累真实对话数据,用来评估 Prompt 效果或构建评测集。
- Prompt 工程师:需要保留每一次调试过程,对比不同写法的输出差异。
- 知识管理爱好者:把 AI 对话产生的知识片段归档为个人知识库素材。
- 后端开发者:想了解 SQLite 建模、命令行工具开发、数据导入导出等基础工程实践。
读完本文,你可以掌握一套从零搭建 ChatArchive 的完整思路,并直接运行一个具备“导入、查询、导出”能力的命令行工具。
2. 环境准备与项目结构
2.1 运行环境与依赖
ChatArchive 是一个 Python 项目,核心依赖非常少,适合用来学习,也适合快速改造成自己的工具。
| 项目 | 建议版本 | 说明 |
|---|---|---|
| Python | 3.9 及以上 | 使用标准库 sqlite3、argparse、json、csv |
| SQLite | 3.24 及以上 | 使用 UPSERT 语法,低版本需要调整 |
| PyYAML | 6.0 及以上 | 用于读取 YAML 格式配置文件 |
Python 3.9 以上通常内置了满足要求的 SQLite 版本。PyYAML 只用于解析config.yaml,如果不想引入这个依赖,也可以把配置文件改成 JSON,文章后面会给出对应写法。
建议在项目目录下创建虚拟环境,避免污染全局 Python 环境:
python -m venv venvWindows 下激活:
venv\Scripts\activatemacOS / Linux 下激活:
source venv/bin/activate然后安装依赖:
pip install -r requirements.txtrequirements.txt内容如下:
PyYAML>=6.02.2 项目结构规划
为了让代码清晰分层,我把项目按功能拆成几个模块:
chatarchive-project/ ├── chatarchive/ │ ├── __init__.py │ ├── config.py # 配置加载 │ ├── models.py # 数据模型定义 │ ├── storage.py # SQLite 存储与查询 │ ├── importer.py # JSON 数据导入 │ ├── exporter.py # Markdown / CSV 导出 │ └── cli.py # 命令行入口 ├── data/ │ └── example_export.json # 示例导入数据 ├── exports/ # 导出结果目录 ├── config.yaml # 配置文件 └── requirements.txt把包名和项目根目录区分开,是为了避免出现chatarchive目录嵌套时,模块解析混乱的问题。实际运行命令时,需要在chatarchive-project根目录下执行。
2.3 配置文件设计
配置文件的作用是让数据库路径、导出目录等参数可以在不改代码的前提下调整。
# config.yaml database: data/chatarchive.db export_dir: exports如果配置加载逻辑中增加更多参数,比如默认模型名称、时区、日志级别等,只需要在配置文件中追加字段,并在config.py中做默认值合并即可。
对应的配置加载代码:
# 文件路径:chatarchive/config.py import yaml from pathlib import Path DEFAULT_CONFIG = { "database": "data/chatarchive.db", "export_dir": "exports", } def load_config(path: str = "config.yaml") -> dict: config_path = Path(path) if not config_path.exists(): return DEFAULT_CONFIG with open(config_path, "r", encoding="utf-8") as f: user_config = yaml.safe_load(f) or {} # 用默认配置兜底,用户配置只覆盖对应字段 return {**DEFAULT_CONFIG, **user_config}这种“默认值 + 用户覆盖”的方式,在配置项逐渐增多时非常实用,可以避免每次读取配置都要处理缺失字段。
3. 核心数据模型与存储设计
3.1 数据模型抽象
不同 AI 平台的聊天记录格式千差万别,但抽象到最后,都能归纳为两层结构:
- 会话(Session):一次完整的对话过程,包含会话 ID、标题、使用的模型名称、创建时间。
- 消息(Message):会话中的单条记录,包含角色、内容、时间、附加元数据。
用户和 AI 之间的多轮对话,本质上就是“一个会话下面挂多条消息”的树形结构,展开后是线性的消息流。这个模型虽然简单,但足以覆盖绝大多数聊天场景。
在 Python 中,我用dataclass来定义这两个数据结构:
# 文件路径:chatarchive/models.py from dataclasses import dataclass, field from datetime import datetime from typing import Optional @dataclass class ChatSession: session_id: str title: str = "未命名会话" model: str = "unknown" created_at: str = field( default_factory=lambda: datetime.now().isoformat(timespec="seconds") ) @dataclass class ChatMessage: session_id: str role: str # user / assistant / system content: str created_at: str = field( default_factory=lambda: datetime.now().isoformat(timespec="seconds") ) metadata: Optional[dict] = field(default_factory=dict)metadata字段用于保存暂时无法结构化、但未来可能需要的信息,比如 token 用量、模型返回的思考过程、平台原始 ID 等。这个字段可以保持 JSON 格式存储,需要时再解析。
3.2 数据库表结构
数据库使用 SQLite,这非常适合本地优先的工具类应用。SQLite 无需独立服务进程,单文件存储,备份和迁移都很方便。
建表语句如下:
CREATE TABLE IF NOT EXISTS chat_sessions ( id TEXT PRIMARY KEY, title TEXT NOT NULL, model TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS chat_messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL, metadata_json TEXT NOT NULL DEFAULT '{}', FOREIGN KEY(session_id) REFERENCES chat_sessions(id) ); CREATE INDEX IF NOT EXISTS idx_messages_session_time ON chat_messages(session_id, created_at);这里有几个设计细节值得说明:
chat_sessions.id使用业务主键,也就是导入数据中的原始会话 ID。这样做的好处是重复导入时可以通过主键判断是否已存在。chat_messages.id使用自增主键,消息本身在多次导入时可能重复,需要业务层去重。metadata_json字段用TEXT类型保存 JSON 字符串,因为 SQLite 没有原生的 JSON 字段类型。- 索引
(session_id, created_at)覆盖了“按会话查询消息时间线”的典型场景。
3.3 为什么选 SQLite
在做本地归档工具时,SQLite 几乎是默认选择:
- 零配置,Python 标准库直接支持。
- 单文件存储,复制文件就能完成备份。
- 支持事务、索引、UPSERT,功能足够强大。
- 避免引入 MySQL/PostgreSQL 等重型组件,降低部署成本。
当然,如果后续需要多人协作、并发写入量很大、或者要做全文检索和向量检索,SQLite 很快会成为瓶颈,那时可以平滑迁移到 PostgreSQL,或者引入专门的检索引擎。当前阶段,SQLite 的“够用 + 简单”是最重要的。
3.4 存储层封装思路
我不希望在 CLI 代码里直接写 SQL,而是把数据库操作封装到一个完整的存储类中。这样可以做到:
- 上层业务不关心 SQLite 连接细节。
- 写操作统一走事务,避免部分成功部分失败。
- 后续换数据库时,只需要替换存储层实现。
存储类的核心代码如下:
# 文件路径:chatarchive/storage.py import json import sqlite3 from contextlib import contextmanager from pathlib import Path from chatarchive.models import ChatMessage, ChatSession class ChatArchiveStorage: def __init__(self, db_path: str): self.db_path = Path(db_path) self.db_path.parent.mkdir(parents=True, exist_ok=True) self.conn = sqlite3.connect(self.db_path) self.conn.row_factory = sqlite3.Row self.conn.execute("PRAGMA journal_mode=WAL;") self._init_schema() def _init_schema(self): with self._transaction() as cur: cur.execute( """ CREATE TABLE IF NOT EXISTS chat_sessions ( id TEXT PRIMARY KEY, title TEXT NOT NULL, model TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL ); """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS chat_messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL, metadata_json TEXT NOT NULL DEFAULT '{}', FOREIGN KEY(session_id) REFERENCES chat_sessions(id) ); """ ) cur.execute( """ CREATE INDEX IF NOT EXISTS idx_messages_session_time ON chat_messages(session_id, created_at); """ ) @contextmanager def _transaction(self): try: cur = self.conn.cursor() yield cur self.conn.commit() except Exception: self.conn.rollback() raise def upsert_session(self, session: ChatSession): with self._transaction() as cur: cur.execute( """ INSERT INTO chat_sessions(id, title, model, created_at) VALUES (?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET title = excluded.title, model = excluded.model """, (session.session_id, session.title, session.model, session.created_at), ) def insert_message(self, message: ChatMessage): with self._transaction() as cur: cur.execute( """ INSERT INTO chat_messages(session_id, role, content, created_at, metadata_json) VALUES (?, ?, ?, ?, ?) """, ( message.session_id, message.role, message.content, message.created_at, json.dumps(message.metadata, ensure_ascii=False), ), ) def list_sessions(self): rows = self.conn.execute( """ SELECT s.*, COUNT(m.id) AS message_count FROM chat_sessions s LEFT JOIN chat_messages m ON s.id = m.session_id GROUP BY s.id ORDER BY s.created_at DESC """ ).fetchall() return [dict(row) for row in rows] def query_messages( self, session_id: str = None, keyword: str = None, limit: int = 100, ): sql = "SELECT * FROM chat_messages WHERE 1=1" params = [] if session_id: sql += " AND session_id = ?" params.append(session_id) if keyword: sql += " AND content LIKE ?" params.append(f"%{keyword}%") sql += " ORDER BY created_at ASC LIMIT ?" params.append(limit) rows = self.conn.execute(sql, params).fetchall() return [dict(row) for row in rows] def close(self): self.conn.close()这里有一个关键操作:PRAGMA journal_mode=WAL;。WAL 模式可以在读操作不阻塞写操作的同时,减少“数据库文件被锁定”的概率,对本地归档应用来说体验更好。
upsert_session使用了 SQLite 的 UPSERT 语法,在会话已存在时会更新标题和模型名称,不会删除原有消息。
4. 完整实战:从零实现 ChatArchive
这一节我们把整个工具串起来,让它可以真正跑起来。
4.1 初始化数据库
运行 ChatArchive 时,存储类会在首次初始化时自动创建数据库文件和数据表。不需要单独执行初始化脚本,也不需要手动建表。这也是“本地优先”工具最舒服的地方:拿到即用。
如果要验证数据库是否创建成功,可以在项目根目录执行:
python -c "from chatarchive.storage import ChatArchiveStorage; storage = ChatArchiveStorage('data/chatarchive.db'); print('初始化成功'); storage.close()"执行后,data/目录下会出现chatarchive.db文件。
4.2 写入聊天记录
写入记录的业务逻辑由importer.py负责。为了让导入逻辑清晰,我约定一个统一的 JSON 格式:
{ "sessions": [ { "id": "sess_001", "title": "Python 学习助手", "model": "gpt-4o-mini", "created_at": "2024-06-01T10:00:00", "messages": [ { "role": "user", "content": "Python 的 with 语句怎么用?", "created_at": "2024-06-01T10:00:10" }, { "role": "assistant", "content": "with 语句用于管理上下文资源,例如文件读写。它的核心是上下文管理器协议,包含 __enter__ 和 __exit__ 方法。", "created_at": "2024-06-01T10:00:12" } ] } ] }导入器读取这个文件,把数据写入 SQLite:
# 文件路径:chatarchive/importer.py import json from chatarchive.models import ChatMessage, ChatSession from chatarchive.storage import ChatArchiveStorage def import_from_json(storage: ChatArchiveStorage, json_path: str): with open(json_path, "r", encoding="utf-8") as f: data = json.load(f) imported_sessions = 0 imported_messages = 0 for session_data in data.get("sessions", []): session = ChatSession( session_id=session_data["id"], title=session_data.get("title", "未命名会话"), model=session_data.get("model", "unknown"), created_at=session_data.get("created_at", ""), ) storage.upsert_session(session) imported_sessions += 1 for message_data in session_data.get("messages", []): message = ChatMessage( session_id=session_data["id"], role=message_data.get("role", "user"), content=message_data.get("content", ""), created_at=message_data.get("created_at", ""), metadata=message_data.get("metadata", {}), ) storage.insert_message(message) imported_messages += 1 return imported_sessions, imported_messages注意,代码里对缺失字段做了默认值处理,比如role默认是user,metadata默认是空字典。这样即使原始数据不规范,导入过程也不会直接崩溃。
4.3 查询会话与消息
查询是 ChatArchive 最常用的能力。我提供了两个查询操作:
- 列出所有会话,并统计每个会话的消息数量。
- 查询消息,支持按会话过滤和关键词模糊匹配。
这些方法已经在storage.py中实现。实际使用时,命令行封装如下:
# 文件路径:chatarchive/cli.py import argparse from chatarchive.config import load_config from chatarchive.exporter import export_messages_to_csv, export_messages_to_markdown from chatarchive.importer import import_from_json from chatarchive.storage import ChatArchiveStorage def main(): parser = argparse.ArgumentParser(description="ChatArchive - AI 聊天记录归档工具") subparsers = parser.add_subparsers(dest="command") import_parser = subparsers.add_parser("import", help="从 JSON 文件导入聊天记录") import_parser.add_argument("--file", required=True, help="JSON 文件路径") import_parser.add_argument("--config", default="config.yaml", help="配置文件路径") query_parser = subparsers.add_parser("query", help="查询聊天记录") query_parser.add_argument("--session", help="会话 ID") query_parser.add_argument("--keyword", help="关键词") query_parser.add_argument("--limit", type=int, default=100) query_parser.add_argument("--export-md", help="导出到 Markdown 文件") query_parser.add_argument("--export-csv", help="导出到 CSV 文件") query_parser.add_argument("--config", default="config.yaml", help="配置文件路径") list_parser = subparsers.add_parser("sessions", help="列出全部会话") list_parser.add_argument("--config", default="config.yaml", help="配置文件路径") args = parser.parse_args() if not args.command: parser.print_help() return storage = ChatArchiveStorage(load_config(args.config)["database"]) try: if args.command == "import": sessions, messages = import_from_json(storage, args.file) print(f"导入完成:{sessions} 个会话,{messages} 条消息") elif args.command == "query": messages = storage.query_messages( session_id=args.session, keyword=args.keyword, limit=args.limit, ) print(f"查询到 {len(messages)} 条消息") if args.export_md: path = export_messages_to_markdown(messages, args.export_md) print(f"已导出 Markdown:{path}") if args.export_csv: path = export_messages_to_csv(messages, args.export_csv) print(f"已导出 CSV:{path}") elif args.command == "sessions": for session in storage.list_sessions(): print( f"{session['id']} | {session['title']} | " f"{session['model']} | {session['message_count']} 条消息 | " f"{session['created_at']}" ) finally: storage.close() if __name__ == "__main__": main()如果你只想把 ChatArchive 当作 Python 库来调用,也可以绕过 CLI,直接操作ChatArchiveStorage和import_from_json,这样能更方便地集成到你自己的 AI 应用项目中。
4.4 导出 Markdown 与 CSV
导出功能是归档工具的另一半能力。查询出来的数据如果不方便带走,价值就会大打折扣。
Markdown 导出很适合生成可读的对话记录文档:
# 文件路径:chatarchive/exporter.py import csv from pathlib import Path def export_messages_to_markdown(messages: list[dict], output_path: str) -> str: output_file = Path(output_path) output_file.parent.mkdir(parents=True, exist_ok=True) with open(output_file, "w", encoding="utf-8") as f: for message in messages: role = message["role"] created_at = message["created_at"] content = message["content"] f.write(f"### {role} · {created_at}\n\n") f.write(f"{content}\n\n") return str(output_file) def export_messages_to_csv(messages: list[dict], output_path: str) -> str: output_file = Path(output_path) output_file.parent.mkdir(parents=True, exist_ok=True) fieldnames = ["id", "session_id", "role", "created_at", "content"] with open(output_file, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for message in messages: writer.writerow({field: message.get(field, "") for field in fieldnames}) return str(output_file)CSV 导出特意使用了utf-8-sig编码。这个细节是为了避免用 Excel 打开 CSV 文件时中文出现乱码。utf-8-sig会在文件开头写入 BOM 标记,Excel 能正确识别编码。
4.5 命令行运行演示
现在我们把整个流程跑通。先创建一个示例数据文件data/example_export.json,内容如下:
{ "sessions": [ { "id": "sess_001", "title": "Python 学习助手", "model": "gpt-4o-mini", "created_at": "2024-06-01T10:00:00", "messages": [ { "role": "user", "content": "Python 的 with 语句怎么用?", "created_at": "2024-06-01T10:00:10" }, { "role": "assistant", "content": "with 语句用于管理上下文资源,比如文件读写。它要求对象实现上下文管理器协议,也就是 __enter__ 和 __exit__ 方法。", "created_at": "2024-06-01T10:00:12" }, { "role": "user", "content": "它和 try-finally 有什么区别?", "created_at": "2024-06-01T10:00:30" } ] } ] }在chatarchive-project根目录下执行:
python -m chatarchive.cli import --file data/example_export.json预期输出:
导入完成:1 个会话,3 条消息列出会话:
python -m chatarchive.cli sessions预期输出:
sess_001 | Python 学习助手 | gpt-4o-mini | 3 条消息 | 2024-06-01T10:00:00查询关键词并把结果导出到 Markdown:
python -m chatarchive.cli query --keyword "with" --export-md exports/with_result.md预期输出:
查询到 2 条消息 已导出 Markdown:exports/with_result.md打开exports/with_result.md,可以看到对话内容已经被格式化保存。
5. 进阶:向真实 AI 平台数据对接
5.1 统一导入格式的重要性
ChatArchive 接收的是统一 JSON 格式,但真实平台的导出格式五花八门。有的是 JSONL,有的是 HTML,有的是 CSV。所以真实落地时,需要为每个平台写一个“适配器”,把平台格式转换成 ChatArchive 的统一格式。
适配器通常只做三件事:
- 读入平台原始导出文件。
- 映射字段,提取会话 ID、角色、内容、时间。
- 输出 ChatArchive 标准 JSON。
这样,无论上游平台怎么变化,下游存储逻辑都不用改。这种“面向统一模型编程”的思路,在 AI 应用开发中同样重要。因为 AI 平台迭代速度很快,今天导出的字段明天可能就变了,保持一个稳定的中间层能减少后续维护成本。
5.2 增量导入与去重
重复导入是很容易踩的坑。如果同一个平台的导出文件被导入了两次,数据库里会出现大量重复消息。
目前的实现里,会话使用了 UPSERT,所以会话不会重复;但消息没有唯一键,重复导入会产生重复记录。解决思路有两种:
- 给消息增加
source_id字段,保存平台原始消息 ID,然后建唯一索引。 - 导入前先按
(session_id, created_at, role, content)去重。
第一种方案更可靠,但要求平台导出的数据里包含稳定的消息 ID。第二种方案不依赖平台 ID,但极端情况下内容完全相同的两条消息可能被误判为重复。
生产环境更推荐第一种方案。你可以在chat_messages表中增加一列source_id TEXT,并建立唯一索引:
CREATE UNIQUE INDEX idx_messages_source_id ON chat_messages(source_id);导入时使用INSERT OR IGNORE,遇到重复 ID 就跳过,实现真正稳定的增量导入。
5.3 把聊天语料用于大模型场景
ChatArchive 不只是“备份聊天记录”,它也可以成为 AI 应用的数据基础设施。
举个例子:如果你想微调一个模型,让它学会某种回答风格,需要准备多轮对话语料。ChatArchive 的统一 JSON 格式可以直接转换为训练集格式。如果你想做 RAG 应用,可以把历史问答中的用户问题作为检索入口,把 AI 回答作为知识片段拆分后入库。
这意味着,ChatArchive 的价值会随着你积累的对话数据量增长而增长。使用越久,数据资产越丰富。
6. 常见问题与排查思路
在开发和运行 ChatArchive 的过程中,可能会遇到下面这些问题。
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| ModuleNotFoundError: No module named 'yaml' | 没有安装 PyYAML | 执行pip install -r requirements.txt |
| sqlite3.OperationalError: no such table: chat_messages | 数据库文件损坏或被手动删除,或 db_path 配置错误 | 检查 config.yaml 的 database 路径;正常情况下重启应用会自动建表 |
| 中文关键词查询不到预期结果 | LIKE 对中文和分词支持有限,关键词包含换行或特殊字符时容易漏匹配 | 先简化关键词重试;明确大小写和分词限制;后续引入 FTS5 全文检索 |
| 批量导入几万条数据时速度慢 | 每条消息都单独提交事务,磁盘 IO 开销大 | 改为批量写入,使用executemany,最后统一 commit |
| Excel 打开 CSV 文件中文乱码 | CSV 使用了普通 utf-8 编码 | 导出时使用utf-8-sig编码 |
| 多个进程同时写入时报 database is locked | SQLite 并发写能力有限 | 使用 WAL 模式,限制同一时刻只有一个写进程;或引入消息队列串行化写入 |
| 重复导入导致消息重复 | 消息表没有唯一约束,缺少去重逻辑 | 增加 source_id 字段和唯一索引,使用INSERT OR IGNORE |
下面是几个高频问题的详细排查逻辑。
启动时报错ModuleNotFoundError
先确认是否在虚拟环境中,再确认是否已经安装依赖:
pip list | grep PyYAML如果没有输出,执行安装命令。
数据库初始化和表结构问题
ChatArchive 启动时会自动执行建表逻辑,但也有可能因为异常中断留下不完整的库文件。如果遇到no such table错误,建议先停止应用,把现有的.db文件重命名备份,再重新运行一次,让程序重新建表。
关键词查询失效
LIKE 查询适合前缀和包含匹配,但中文场景下会有明显局限。比如搜索“with语句”时,由于消息内容里可能是“with 语句”,中间有空格,模糊匹配会可能失败。简单场景可以用LIKE '%关键词%'应付,复杂场景建议引入全文检索。
SQLite 从 3.34 版本开始支持 FTS5 的trigramtokenizer,它可以在不安装额外扩展的情况下比较好地处理中文连续字符匹配。示例建表语句如下:
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(content, tokenize = 'trigram');每次插入消息后,手动同步到 FTS 表,查询时先查 FTS 表得到消息 ID 列表,再回表查询详情。这套方案可以显著提升中文检索体验。
7. 最佳实践与工程建议
ChatArchive 项目虽小,但把它当作一个长期使用的工程工具来设计,会有很多细节值得打磨。
第一,始终坚持本地优先。
AI 聊天记录往往包含业务代码、个人思考、内部方案等敏感信息。默认把数据保存在本地,可以规避很多隐私风险。如果未来需要同步到云端,也应该用增量同步或加密同步方案,而不是直接把原始数据库上传。
第二,统一用标准化时间格式。
导入数据的时间字段统一使用 ISO 8601 格式,比如2024-06-01T10:00:00。这样排序、比较、序列化都方便,也不会因为时区问题导致时间错乱。
第三,给消息添加稳定的 source_id。
只要平台的导出数据里存在原始消息 ID,就应该保留到本地数据库。它不仅是去重依据,也是将来与官方数据对齐、做增量同步的基础字段。
第四,导出格式要做到“可回流”。
导出的 Markdown 适合人阅读,导出的 JSON 适合机器处理。建议导出的 JSON 和导入格式保持一致,这样就算数据库损坏,也能通过导出的文件重新构建整个归档库。
第五,数据库备份要规范化。
直接复制.db文件虽然也能用,但 WAL 模式下可能存在未合并的日志。更稳妥的做法是使用 SQLite 的在线备份接口:
import sqlite3 source = sqlite3.connect("data/chatarchive.db") backup = sqlite3.connect("data/chatarchive_backup.db") source.backup(backup) backup.close() source.close()这样备份出来的是一个完整、一致、可立即使用的数据库文件。
第六,检索能力要做分层建设。
第一层是当前已经实现的LIKE模糊查询,适合快速原型。第二层是 FTS5 全文检索,适合中文搜索和小型知识库。第三层是向量检索,适合“语义相似度”搜索,可以配合 embedding 模型把消息向量化后存入向量数据库。不要一开始就上重型架构,而是按数据量逐步演进。
第七,保留原始数据。
在导入器转换数据时,不要直接丢弃平台原始字段。可以把原始内容整个保存到metadata_json中,这样即使在转换逻辑里遗漏了某些字段,之后还能从元数据中找回,避免数据丢失。
8. 总结与下一步学习方向
ChatArchive 的核心价值是把分散的 AI 对话记录沉淀成结构化资产。从零开始,我实现了会话和消息的统一数据模型,使用 SQLite 完成本地存储,通过导入器接收异构数据,通过查询和导出能力让历史对话真正可用。
在这个过程里,有几个工程点非常重要:
- 数据模型要抽象得足够简单,能覆盖不同平台的对话结构。
- 存储层要封装好事务和连接管理,避免上层业务直接写 SQL。
- 导入逻辑要为字段缺失留好默认值,保证数据容错。
- 导出格式要兼顾人读和机器读两种场景。
如果你想继续改进这个项目,下面这些方向值得尝试:
- 增加 Web 界面:用 FastAPI 或 Flask 封装查询接口,让非技术人员也能通过浏览器检索聊天记录。
- 接入更多 AI 平台导出格式:从单一 JSON 格式开始,扩展支持 OpenAI、Claude 等平台的官方导出文件。
- 引入中文全文检索:用 FTS5 trigram 或 jieba 分词方案,替代现在的 LIKE 查询。
- 加入向量检索:把消息内容做 embedding,实现“根据语义找历史对话”的能力。
动手把 ChatArchive 跑起来,然后试着导入一份你自己真实的聊天记录。当你发现几个月前的一段 AI 回答还能被快速检索到的时候,应该就能理解这个工具真正解决的问题是什么了。