企业官网建设流程全解析

在实际开发场景里，AI 聊天工具用得越深，越容易遇到同一个问题：对话记录散落在网页端、桌面端、命令行和不同模型的官方后台里，过几天想找回一次 prompt 调优过程、一段故障排查结论或一套议题讨论，往往只能靠记忆力翻找。ChatArchive 要解决的正是这个问题。它的定位不是聊天前端，而是 AI 聊天记录的归档与再利用系统：把分散的对话统一收拢到本地数据库，提供结构化管理、全文检索、标签组织和 Markdown/JSON 导出能力，让历史聊天从“一次性消耗品”变成可以检索、复用和沉淀的知识资产。

本文会从零实现一个最小可运行的 ChatArchive 服务，技术栈选用 Python + FastAPI + SQLite，数据库使用 SQLite 内置的 FTS5 全文检索，并在检索部分单独处理中文查询的问题。整体不引入消息队列、不依赖外部搜索引擎，所有功能都可以在一台开发机上跑通。学完之后，你会理解聊天归档类系统最核心的收、存、查、导四个环节是如何设计的，也能知道在什么阶段需要引入向量检索、多用户和加密备份等生产级能力。

1. 先想清楚 ChatArchive 要解决什么问题

1.1 聊天记录和普通日志本质上不一样

很多人会把聊天记录当成日志来管理，这是归档系统设计里第一个需要纠正的认知。普通服务日志是单条、平铺、无对话语境的，记录的主要目标是排障；而 AI 聊天记录天然带有“回合”和“上下文”：

• 同一段对话里有 system、user、assistant 多种角色。
• 一个问题的答案往往依赖前面几条消息的上下文。
• 用户可能在同一场对话里反复调整 prompt，形成一版 prompt 的演进过程。
• 对话携带来源（OpenAI、Claude、本地模型、命令行工具）和模型标识。

如果归档时只保存 message 内容而不保存角色、时间、会话归属、模型和元数据，后续检索到的结果就失去了可追溯性。比如搜到一条 assistant 回复，却不知道当时用的什么模型、哪一天、在哪个会话里回答的，那这条记录只能当素材看，不能当工程依据。

因此在数据模型设计上，ChatArchive 一开始就要区分会话表和消息表，而不是把所有内容塞进一张宽表。

1.2 在线聊天与离线归档的职责边界

ChatArchive 不承担实时对话功能。它和在线聊天系统的边界可以这样划分：

• 在线聊天系统负责产生对话，交互要快，状态要实时。
• 归档系统负责保存对话，写入后基本不可变，核心价值是稳定和可检索。
• 归档系统需要兼容不同来源，所以导入格式必须统一。
• 归档系统要有导出能力，避免厂商锁定和本地数据丢失。

这里还要说清楚一个容易误解的地方：项目标题里的“重铸 AI 聊天荣光”，可以理解为让被丢弃的聊天记录重新产生价值。第一阶段不需要做模型调用，先把历史数据管起来；第二阶段才把归档内容当作知识库，检索结果注入新 prompt，实现基于历史对话的问答。这样分阶段落地，项目难度和风险都可控。

1.3 最小功能范围

实现 ChatArchive 之前，先圈定最小闭环要包含哪些能力。下面的表是第一阶段的功能范围：

按这个范围，ChatArchive 的核心链路是：导入一段对话 -> 拆成会话和消息落库 -> 建立全文索引 -> 通过 API 检索 -> 导出成可阅读或可迁移的格式。

2. 技术选型与环境准备：用最小依赖跑通第一版

2.1 为什么选 Python + FastAPI + SQLite

选型不是越重越好，而是要和问题规模匹配。ChatArchive 第一版是本地单机工具，数据量在几万到几十万条消息量级，SQLite 完全够用，而且能避免引入数据库服务器带来的运维成本。

各组件的作用如下：

• FastAPI：提供 REST API，自带 OpenAPI 文档，配合 Pydantic 做请求体校验，开发效率高。
• SQLite：单文件数据库，支持 WAL 模式，自带 FTS5 全文检索扩展，适合归档类轻量应用。
• 标准库 sqlite3：第一版直接用原生 SQL，避免 ORM 引入过多抽象，也让读者能看清 SQL 走向。
• uvicorn：FastAPI 的 ASGI 服务容器，本地开发和测试都方便。

如果后续数据量上来了，可以把存储层替换为 PostgreSQL，检索层替换为 Elasticsearch 或向量数据库，但接口层和导入导出格式可以保持不变。

2.2 开发环境清单

建议按下面的版本准备环境：

打开终端执行下面的命令创建项目和虚拟环境：

mkdir chatarchive cd chatarchive python -m venv .venv source .venv/bin/activate pip install "fastapi>=0.110" "uvicorn[standard]>=0.29" "pydantic>=2" pip install pytest

安装完之后可以检查一下 SQLite 版本和 FTS5 是否可用：

python -c "import sqlite3; print(sqlite3.sqlite_version)" python -c "import sqlite3; conn=sqlite3.connect(':memory:'); print(conn.execute('CREATE VIRTUAL TABLE t USING fts5(x); select 1 from t').fetchone())"

如果第二条命令报no such module: fts5，说明当前 Python 编译时未启用 FTS5，建议换用系统自带 Python 或重新安装支持 FTS5 的 Python 构建。这个问题在常见问题部分会再展开。

2.3 项目目录结构

第一版采用扁平结构，文件少、依赖清晰，方便学习和扩展：

chatarchive/ ├── app.py # FastAPI 应用与路由 ├── db.py # SQLite 连接、建库、事务封装 ├── models.py # Pydantic 请求与响应模型 ├── schema.sql # 建表 SQL ├── importer.py # JSONL 导入解析 ├── exporter.py # Markdown / JSON 导出 ├── search.py # FTS5 全文检索与 LIKE 回退 └── tests/ └── test_api.py # 接口测试

这个结构里，db.py是数据访问层，search.py独立出来是因为中文全文检索有特殊处理，单独成文件方便后续替换为向量检索实现。

3. 数据库设计：聊天归档的核心是会话与消息的时间线

3.1 为什么必须拆成会话表和消息表

聊天记录天然是“一对多”的结构：一个会话包含多条消息，消息之间有严格的先后顺序。如果不拆分，直接把所有消息存在一张表里，会出现几个问题：

• 无法高效查询“某个会话的全部消息”。
• 会话级别的信息（标题、来源、模型、标签）每条消息都要重复，造成冗余。
• 删除或归档一个会话时，需要手动删除多条记录，容易残留脏数据。

所以 schema 里用conversations保存会话元数据，用messages保存消息明细，通过外键关联。

3.2 建表 SQL 与字段解释

schema.sql内容如下：

PRAGMA journal_mode = WAL; PRAGMA foreign_keys = ON; PRAGMA busy_timeout = 5000; CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, external_id TEXT UNIQUE, title TEXT NOT NULL DEFAULT '', source TEXT NOT NULL DEFAULT 'unknown', model TEXT, tags TEXT NOT NULL DEFAULT '[]', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id INTEGER NOT NULL, role TEXT NOT NULL CHECK (role IN ('system', 'user', 'assistant', 'tool')), content TEXT NOT NULL, created_at TEXT NOT NULL, metadata_json TEXT NOT NULL DEFAULT '{}', FOREIGN KEY (conversation_id) REFERENCES conversations(id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_messages_conv_time ON messages(conversation_id, created_at); CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5( content, tokenize = 'trigram' );

字段设计里有几个关键决策需要理解：

• external_id是来源系统里的原始会话 ID，加 UNIQUE 约束是为了导入幂等。同一个会话重复导入时，可以跳过或更新，而不是产生重复数据。
• role用 CHECK 约束限定取值范围，避免脏数据进入表。
• tags以 JSON 数组的字符串形式保存。第一版不做标签规范化，只做辅助分类。
• created_at统一存 ISO 8601 字符串，且建议统一存 UTC 时间，展示时再转本地时区。混合时区是归档系统最常见的时间混乱来源。
• messages_fts使用 FTS5，tokenize 选trigram。这个选择对中文检索很重要，后面会专门解释。

3.3 外键与 WAL 的细节

建表使用PRAGMA foreign_keys = ON启用外键约束，否则ON DELETE CASCADE不会生效。每次建立连接后都要执行这个 PRAGMA，而不是只在建库时执行一次，因为外键约束是按连接级别生效的。

WAL 模式的好处是读写并发能力更好：读操作不会阻塞写操作，适合“导入历史数据的同时还要支持检索”的场景。busy_timeout = 5000设置 5 秒锁等待时间，避免多进程写入时直接报database is locked。

这段 SQL 里没有设置content=参数，messages_fts是独立的全文字段。这样实现最简单，代价是消息更新或删除时，全文索引里的数据也需要同步更新。第一版以追加写入为主，几乎不更新消息，所以这个取舍是划算的。

4. 核心代码实现：把收、存、查、导四个动作串起来

4.1 数据库连接与事务封装

db.py负责连接管理、建库和事务封装。为了控制篇幅，这里用标准库sqlite3实现，不引入 ORM：

import sqlite3 from contextlib import contextmanager DB_PATH = "chatarchive.db" def get_conn(): conn = sqlite3.connect(DB_PATH, timeout=5) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") conn.execute("PRAGMA journal_mode = WAL") conn.execute("PRAGMA busy_timeout = 5000") return conn @contextmanager def transaction(): conn = get_conn() try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def init_db(): import os if not os.path.exists("schema.sql"): raise FileNotFoundError("缺少 schema.sql") with open("schema.sql", "r", encoding="utf-8") as f: schema = f.read() with transaction() as conn: conn.executescript(schema)

这里使用上下文管理器封装事务，yield之后如果函数体没抛异常就 commit，否则 rollback。所有写操作都通过这个上下文管理器执行，能避免手写try/except/commit/rollback带来的遗漏。

4.2 存储会话与消息的 Dao 层

db.py里继续加入保存会话和消息的逻辑。保存一段完整对话时，必须在一个事务里同时写入会话、消息和全文索引，任何一步失败都不能留下半截数据：

def create_conversation_with_messages(conversation_data, messages): now = conversation_data.get("created_at") if not now: now = get_utc_now() updated_at = conversation_data.get("updated_at") or now tags_json = json.dumps(conversation_data.get("tags", []), ensure_ascii=False) with transaction() as conn: cur = conn.execute( """ INSERT INTO conversations (external_id, title, source, model, tags, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( conversation_data.get("external_id"), conversation_data.get("title", ""), conversation_data.get("source", "unknown"), conversation_data.get("model"), tags_json, now, updated_at, ), ) conversation_id = cur.lastrowid for msg in messages: message_created_at = msg.get("created_at") or now cur = conn.execute( """ INSERT INTO messages (conversation_id, role, content, created_at, metadata_json) VALUES (?, ?, ?, ?, ?) """, ( conversation_id, msg["role"], msg["content"], message_created_at, json.dumps(msg.get("metadata", {}), ensure_ascii=False), ), ) conn.execute( "INSERT INTO messages_fts(rowid, content) VALUES (?, ?)", (cur.lastrowid, msg["content"]), ) return conversation_id

这里要注意messages_fts插入的是rowid，它必须对应messages.id，否则检索联表时会错位。sqlite3的lastrowid在 insert 后可直接取到自增主键。

4.3 Pydantic 请求模型

models.py定义接口的请求结构，让 FastAPI 自动完成参数校验：

from typing import Optional from pydantic import BaseModel class MessageIn(BaseModel): role: str content: str created_at: Optional[str] = None metadata: dict = {} class ConversationIn(BaseModel): title: Optional[str] = "" external_id: Optional[str] = None source: str = "unknown" model: Optional[str] = None tags: list = [] messages: list[MessageIn]

role这里先不做枚举校验，数据库层的 CHECK 约束会兜底。如果请求里传入了非法 role，SQLite 会在这条消息写入时抛异常，事务回滚，整个会话都不会落库。这个设计是有意的：导入接口应该坚持“一条消息非法，整段对话不入库”，而不是留下残缺数据。

4.4 FastAPI 路由：保存、列表、详情、删除

app.py中实现最核心的几个接口：

from fastapi import FastAPI, HTTPException from fastapi.responses import PlainTextResponse from db import init_db, create_conversation_with_messages, get_conversation_list, get_conversation_detail, delete_conversation from models import ConversationIn app = FastAPI() @app.on_event("startup") def startup(): init_db() @app.post("/api/conversations") def create_conversation(payload: ConversationIn): messages = [m.model_dump() for m in payload.messages] if not messages: raise HTTPException(status_code=400, detail="messages 不能为空") conversation_id = create_conversation_with_messages(payload.model_dump(), messages) return {"id": conversation_id, "status": "ok"} @app.get("/api/conversations") def list_conversations(page: int = 1, page_size: int = 20): return get_conversation_list(page=page, page_size=page_size) @app.get("/api/conversations/{conversation_id}") def conversation_detail(conversation_id: int): detail = get_conversation_detail(conversation_id) if not detail: raise HTTPException(status_code=404, detail="会话不存在") return detail @app.delete("/api/conversations/{conversation_id}") def conversation_delete(conversation_id: int): ok = delete_conversation(conversation_id) if not ok: raise HTTPException(status_code=404, detail="会话不存在") return {"status": "ok"}

get_conversation_list、get_conversation_detail和delete_conversation是db.py里的查询函数。delete_conversation删除会话时，由于外键ON DELETE CASCADE已启用，messages 和 messages_fts 的旧数据需要额外处理：独立 FTS 表不受外键影响，所以删除会话后必须手动删除对应 FTS 行，或在应用层先查会话关联的所有 message id 再删除索引。这里建议在删除逻辑里补一步：

def delete_conversation(conversation_id): with transaction() as conn: rows = conn.execute( "SELECT id FROM messages WHERE conversation_id = ?", (conversation_id,) ).fetchall() ids = [r["id"] for r in rows] if ids: placeholders = ",".join("?" * len(ids)) conn.execute(f"DELETE FROM messages_fts WHERE rowid IN ({placeholders})", ids) conn.execute("DELETE FROM conversations WHERE id = ?", (conversation_id,)) return True

这里用 f-string 拼接占位符，是因为IN子句需要可变数量的参数，但参数值仍然由?占位，不会引入注入风险。

4.5 中文全文检索：为什么 FTS5 要用 trigram

SQLite FTS5 默认的unicode61分词器按空格和标点分词，对英文很友好，但对中文很不友好。中文句子没有空格，unicode61会把整句话当成一个 token，用户搜索“如何排查端口”时，整体短语匹配不到，前缀匹配也失效。

trigram分词器会把文本切成连续的三字符片段。例如“如何排查端口占用”会被分成“如何排、何排查、排查端、查端口、端口占、口占用”等多个三字 ngram，子串匹配能力显著增强。

但 trigram 也有一个限制：少于三个字符的查询词无法作为有效 token 匹配。因此在search.py里需要做长度判断和回退：

import sqlite3 def escape_phrase(term: str) -> str: return '"' + term.replace('"', '""') + '"' def search_messages(db_path, q, limit=20): conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row try: if len(q) < 3: rows = conn.execute( """ SELECT m.id, m.conversation_id, m.role, m.content, m.created_at, c.title AS conv_title, c.model FROM messages m JOIN conversations c ON c.id = m.conversation_id WHERE m.content LIKE ? ORDER BY m.created_at DESC LIMIT ? """, (f"%{q}%", limit), ).fetchall() else: phrase = escape_phrase(q) rows = conn.execute( """ SELECT m.id, m.conversation_id, m.role, m.content, m.created_at, c.title AS conv_title, c.model FROM messages_fts f JOIN messages m ON m.id = f.rowid JOIN conversations c ON c.id = m.conversation_id WHERE messages_fts MATCH ? ORDER BY rank LIMIT ? """, (phrase, limit), ).fetchall() return [dict(r) for r in rows] finally: conn.close()

两点解释：

• MATCH查询里使用双引号包裹用户输入，是把查询当作短语匹配；用户输入中的双引号必须转义为两个双引号，否则会破坏 FTS 查询语法。
• 小于 3 个字符的回退用LIKE，因为 trigram 无法匹配短词。比如用户搜“AI”，trigram 索引无法命中，直接走LIKE '%AI%'更可靠。

检索结果保留了conversation_id、role、created_at，这样前端展示时可以拼出“哪场对话里的谁在什么时候说了什么”。

4.6 导入：JSONL 统一格式

importer.py处理导入文件。第一版定义统一的 JSONL 格式，每行一个 JSON 对象：

{"external_id": "conv-001", "title": "排查端口占用", "source": "cli", "model": "gpt-4o", "created_at": "2025-01-01T10:00:00Z", "tags": ["linux", "网络"], "messages": [{"role": "user", "content": "如何查看 8080 端口被谁占用", "created_at": "2025-01-01T10:00:01Z"}, {"role": "assistant", "content": "使用 lsof -i :8080 或 ss -lptn 'sport = :8080'", "created_at": "2025-01-01T10:00:05Z"}]}

导入函数按行读取、逐条校验，并统计成功和失败行数：

import json def import_jsonl(path, create_func): success, failed = 0, 0 with open(path, "r", encoding="utf-8") as f: for line_no, line in enumerate(f, 1): line = line.strip() if not line: continue try: obj = json.loads(line) messages = obj.get("messages", []) if not messages: raise ValueError("messages 为空") create_func(obj, messages) success += 1 except Exception as exc: failed += 1 print(f"第 {line_no} 行导入失败: {exc}") return {"success": success, "failed": failed}

导入失败时不中断整个文件，而是跳过失败行并记录行号，方便使用者修复数据后重新导入。external_id的唯一约束在重复导入时会抛异常，实测需要把它也统计为“跳过”而不是“失败”，这里可以根据业务语义决定。

4.7 导出：Markdown 与 JSON

exporter.py负责把会话导出成两种格式。JSON 导出适合迁移，Markdown 导出适合阅读和沉淀。

def to_markdown(detail): title = detail["title"] or "未命名对话" meta = ( f"> 来源: {detail['source']} | 模型: {detail['model'] or '未知'}" f" | 时间: {detail['created_at']}" ) lines = [f"# {title}", "", meta, ""] for msg in detail["messages"]: role = msg["role"] content = msg["content"].strip() lines.append(f"## {role}") lines.append("") lines.append(content) lines.append("") return "\n".join(lines)

导出接口返回PlainTextResponse，并设置 UTF-8 编码：

@app.get("/api/export/{conversation_id}") def export_conversation(conversation_id: int, format: str = "markdown"): detail = get_conversation_detail(conversation_id) if not detail: raise HTTPException(status_code=404, detail="会话不存在") if format == "markdown": return PlainTextResponse(to_markdown(detail), media_type="text/markdown; charset=utf-8") if format == "json": return detail raise HTTPException(status_code=400, detail="format 仅支持 markdown 或 json")

JSON 导出直接返回detail结构，它包含会话元数据和消息数组。从 ChatGPT、Claude 等平台导出的官方数据，需要写对应格式的转换器，统一成上面的 JSONL 结构后再导入。

5. 启动服务并用 curl 跑通整个闭环

5.1 启动 FastAPI 服务

在项目根目录执行：

uvicorn app:app --reload --host 0.0.0.0 --port 8000

启动后访问http://127.0.0.1:8000/docs，可以看到 FastAPI 自动生成的 Swagger 接口文档。开发阶段用--reload，生产环境建议关掉自动重载并配合进程管理器。

5.2 创建一段对话

用 curl 新增一个会话：

curl -X POST http://127.0.0.1:8000/api/conversations \ -H 'Content-Type: application/json' \ -d '{ "external_id": "conv-001", "title": "如何排查端口占用", "source": "cli", "model": "gpt-4o", "tags": ["linux", "网络"], "messages": [ {"role": "user", "content": "如何查看 8080 端口被谁占用"}, {"role": "assistant", "content": "可以使用 lsof -i :8080 或 ss -lptn '"'"'sport = :8080'"'"' 查看进程"} ] }'

响应应类似：

{"id": 1, "status": "ok"}

这里 shell 嵌套引号比较繁琐，实际开发建议用 Swagger 文档或 Python 脚本提交。r 上面的示例如果粘贴到终端报错，可以把 JSON 保存到payload.json，再用下面的命令提交：

curl -X POST http://127.0.0.1:8000/api/conversations \ -H 'Content-Type: application/json' \ -d @payload.json

5.3 查询会话列表和详情

curl -s http://127.0.0.1:8000/api/conversations | python -m json.tool curl -s http://127.0.0.1:8000/api/conversations/1 | python -m json.tool

详情接口返回里，messages数组应该按创建时间正序排列。如果发现顺序错乱，检查导入时created_at字段是否完整，以及查询函数里是否按created_at排序。

5.4 检索并导出

检索中文和多字词：

curl -s --get --data-urlencode 'q=端口' http://127.0.0.1:8000/api/search | python -m json.tool

导出 Markdown：

curl -s http://127.0.0.1:8000/api/export/1?format=markdown

预期输出类似：

# 如何排查端口占用 > 来源: cli | 模型: gpt-4o | 时间: 2025-01-01T10:00:00Z ## user 如何查看 8080 端口被谁占用 ## assistant 可以使用 lsof -i :8080 或 ss -lptn 'sport = :8080' 查看进程

到这一步，收、存、查、导四个环节已经完整跑通。

5.5 自动化测试

接口测试可以用 FastAPI 的TestClient完成，避免每次手工启动服务。tests/test_api.py示例：

from fastapi.testclient import TestClient from app import app client = TestClient(app) def test_create_conv(): resp = client.post("/api/conversations", json={ "title": "测试对话", "model": "test-model", "messages": [ {"role": "user", "content": "你好"}, {"role": "assistant", "content": "你好，有什么可以帮你"} ] }) assert resp.status_code == 200 conv_id = resp.json()["id"] detail = client.get(f"/api/conversations/{conv_id}").json() assert len(detail["messages"]) == 2 search = client.get("/api/search", params={"q": "你好"}).json() assert len(search["results"]) >= 1

运行测试：

pytest -q

测试时建议使用独立测试数据库，避免污染开发数据。可以在db.py里用环境变量控制数据库路径，或者测试 fixture 里临时替换DB_PATH。

6. 常见问题与排查链路

6.1 检索不到数据，或 MATCH 报错

排查顺序建议：先确认是否真的写入成功了，再确认 FTS 表里有没有对应行，最后确认查询词长度和转义逻辑。这三个点按顺序检查，能覆盖大部分检索问题。

6.2 导入后消息顺序混乱

现象是详情接口返回的消息顺序和原始 JSONL 不一致。原因通常是created_at字段缺失或格式不统一，比如有的行是 ISO 8601，有的是时间戳，有的没有。

检查方式：

• 查看导入时是否打印了messages 为空或行解析异常。
• 用 Python 读取源文件，检查每行的created_at是否存在。
• 查询数据库里该会话的created_at列，对比可见值是否杂乱。

处理建议是导入函数里做统一规范化：缺省时间用当前 UTC 时间，时间戳格式转成 ISO 8601，解析失败时直接判定该行导入失败，而不是默认取当前时间掩盖污染数据。

6.3 并发写入时报 database is locked

本地工具一般单用户使用，但在导入大文件时，如果同时有多个进程或线程写入，SQLite 会报锁错误。

排查顺序：

• 是否开启了 WAL 模式。
• 是否设置了busy_timeout。
• 是否存在长时间的写事务没有提交。
• 是否有多个实例同时指向同一个数据库文件。

处理方案是把busy_timeout设为 5000 毫秒，写操作用单一事务批次提交，避免逐条自动提交。生产环境如果并发写入量很大，再考虑迁移 PostgreSQL。

6.4 FTS 表删数据不同步

这是一个隐蔽问题。由于第一版使用独立 FTS 表，删除会话时如果只删conversations和messages，messages_fts里仍然残留旧索引，检索时 join 不到messages行，导致结果丢失。

处理方案在删除逻辑里补充手动删除 FTS 索引行。更好的长期方案是改用 FTS5 external content 表并创建触发器，或者封装统一的删除函数，所有删除都走这一个入口。

6.5 导出文件中文乱码

Windows 下保存text/markdown响应时可能会乱码。原因通常是响应头没有声明charset=utf-8。上面的导出接口已经在media_type里带了charset=utf-8，保存时再用 Python 读取并写入文本文件即可：

curl -s http://127.0.0.1:8000/api/export/1?format=markdown -o conv1.md

读取时显式指定 UTF-8：

with open("conv1.md", encoding="utf-8") as f: content = f.read()

7. 生产化增强与最佳实践

7.1 发布前检查清单

本地跑通之后，如果要接到真实工作流里，建议按下面的清单逐项确认：

• 数据库备份：归档系统的数据价值高，发布前确认有没有定时备份脚本和恢复演练。
• 导入幂等性：重复导入同一份 JSONL 时，external_id冲突如何处理，要明确是跳过还是覆盖。
• 索引同步：消息更新、删除时，FTS 索引是否同步处理。
• 编码统一：所有读写文件显式指定 UTF-8，避免平台默认编码差异。
• 时区统一：入库时间全部转 UTC，导出时再转本地时区。
• 日志与监控：FastAPI 进程是否有访问日志，导入失败是否记录到独立日志文件。
• 性能验证：用一份数万条消息的真实数据压测检索接口，确认响应时间在可接受范围。
• 脱敏与安全：聊天记录可能包含敏感信息，本地数据库文件是否有加密或访问控制。
• 反向依赖：导出 JSON 是否能被原系统再次导入，迁移链路要双向可用。

7.2 检索增强：从关键词到语义

FTS5 只能解决字面检索。如果历史聊天里用户问“端口被占了”，但新搜索词是“端口冲突”，字面匹配会漏掉。第二阶段建议引入向量检索：

• 使用本地 embedding 模型把消息内容向量化。
• 用 SQLite 的sqlite-vec扩展或独立向量数据库保存向量。
• 查询时先做语义召回，再做关键词过滤或重排。

实现时要注意向量化和索引的更新策略：新消息入库后异步生成向量，不能阻塞导入主流程。

7.3 把归档数据变成对话上下文

ChatArchive 最有价值的生产用法，是让历史聊天参与新一轮对话。流程可以设计为：

1. 用户提问。
2. 从归档库检索相关历史消息。
3. 把命中结果按时间线拼成上下文片段。
4. 与用户问题一起发给大模型。
5. 模型基于历史记录回答当前问题。

这个流程本质上是检索增强生成，也就是 RAG。关键点在于检索结果不能只按相似度截断，还要保留角色和会话归属，避免把不同会话的碎片混在一块导致模型误解。

7.4 扩展方向速查

7.5 对新手最有价值的练习

ChatArchive 是一个很适合练习后端工程的项目，因为它涉及文件解析、数据建模、全文检索、接口设计、异常处理多个知识面。建议按顺序做四个练习：

1. 把 ChatGPT 官方导出数据写成转换器，统一成 ChatArchive 的 JSONL 格式。
2. 给会话列表接口增加按时间范围过滤和按标签过滤。
3. 为 search 接口增加分页和搜索结果高亮片段。
4. 把 FTS 表切换为 external content 模式，用触发器自动同步索引。

做完这四个练习，基本就能理解一个归档类系统从数据入口到检索出口的完整链路。ChatArchive 的价值不在于代码复杂度，而在于它把聊天记录从“用完即走的对话”变成了“可沉淀、可复用、可追溯的工程资产”。实际项目里接入的时候，优先保证导入幂等、时间线完整和导出可用，这三件事做好，工具就能真正用起来。