1. 为什么我把 RAG 换成了 SQL 骨架检索
如果你正在搭 AI 知识库,大概率已经踩过 RAG 的坑:文档切成几百个 chunk,向量召回一堆语义相近的碎片,问一个需要跨文档串联的问题,模型就开始一本正经地胡说。GraphRAG 想补上关系推理,但离线建图那一步的计算量和增量维护成本,普通团队根本扛不住。我最近在试的 SAG(SQL-Retrieval Augmented Generation)思路不太一样——它不预先建全局图,而是把事件和实体存进 SQL 表,查询时用 SQL 沿着共享实体现场织一张局部超边网。简单说,RAG 找的是“长得像”,SAG 找的是“有关系”,而且这个关系是查的时候才算,增量更新只写新行,不用重算全图。
这篇不聊论文指标,直接给你一套能跑的最小链路:SQL 表结构骨架、TaoToken 统一 Key 配置、检索请求验证,以及命中率和延迟怎么测。适合已经用过 RAG、想换一套结构化检索骨架的开发者,也适合想让 Agent 读到私有知识库的人。全程本地 SQLite 就能起步,不需要先装图数据库。
2. TaoToken 前置:一个 Key 打通检索与生成
SAG 的检索链路里有两处要调模型:一是文档入库时的事件抽取和实体抽取,二是精确模式下对候选事件做 LLM 精排。如果这两处各配一套厂商 Key,调试时会很乱。我用 TaoToken 做统一入口,一个 Key 同时覆盖对话模型和 Embedding,OpenAI 兼容接口,改 base_url 就能接。
先拿 Key:打开 https://taotoken.net/api-keys ,登录后创建一个 API Key,复制保存。注意这个 Key 只在创建时完整显示一次,丢了就重建。
拿到 Key 后,SAG 的模型配置里填两个东西:
# SAG 模型配置(OpenAI 兼容) OPENAI_BASE_URL=https://taotoken.net/api OPENAI_API_KEY=sk-你的TaoToken密钥 EMBEDDING_MODEL=text-embedding-3-small CHAT_MODEL=gpt-4o-mini如果你用的是 SAG 桌面客户端,在设置页把“API 地址”改成https://taotoken.net/api,Key 填进去,模型名按你账号里可用的填。想先确认模型通不通,可以去 https://taotoken.net/models 看可用列表,或者直接在 https://taotoken.net/chat 里发一句话测试。
注意:base_url 结尾不要带
/v1,TaoToken 的兼容层已经处理了路径,多写一层会 404。这个坑我踩过,排查了半小时。
3. 可复制的 SQL 表结构骨架
SAG 的核心数据模型就三张表:事件表、实体表、关联表。事件对应原文 chunk,实体是从 chunk 里抽出来的名词性线索,关联表记录“哪个事件提到了哪个实体”。查询时先向量召回种子事件,再通过关联表用 SQL 把共享实体的事件串起来。
下面是我在 SQLite 上跑通的骨架,PostgreSQL 只需把AUTOINCREMENT换成SERIAL、TEXT换VARCHAR即可:
-- 事件表:每个 chunk 一条,带向量 CREATE TABLE events ( id INTEGER PRIMARY KEY AUTOINCREMENT, doc_id TEXT NOT NULL, chunk_text TEXT NOT NULL, embedding BLOB, -- 向量序列化存储 created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); -- 实体表:实体名唯一,避免重复 CREATE TABLE entities ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, entity_type TEXT, -- 人物/组织/产品/时间等 created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); -- 关联表:事件与实体的多对多 CREATE TABLE event_entities ( event_id INTEGER NOT NULL, entity_id INTEGER NOT NULL, PRIMARY KEY (event_id, entity_id), FOREIGN KEY (event_id) REFERENCES events(id), FOREIGN KEY (entity_id) REFERENCES entities(id) ); -- 加速查询:按实体反查事件 CREATE INDEX idx_ee_entity ON event_entities(entity_id); CREATE INDEX idx_ee_event ON event_entities(event_id);入库流程分三步。第一步,文档切块后写入events,同时调 Embedding 接口把向量存进embedding字段。第二步,对每个 chunk 调一次对话模型抽实体,提示词可以很简单:
从下面这段文本中抽取所有实体,每行一个,格式为:实体名|类型 只输出实体,不要解释。 文本:{chunk_text}第三步,把抽出的实体INSERT OR IGNORE进entities,拿到 entity_id 后写入event_entities。整个过程没有全局图,新文档进来就是插新行,旧数据一行不动。
查询时的 SQL 骨架是这样:先用向量召回 top-k 种子事件,再沿实体扩展一跳。
-- 假设种子事件 id 为 101, 205, 330 -- 找出与种子事件共享实体的其他事件 SELECT DISTINCT e.id, e.chunk_text, COUNT(*) AS shared FROM event_entities ee1 JOIN event_entities ee2 ON ee1.entity_id = ee2.entity_id JOIN events e ON e.id = ee2.event_id WHERE ee1.event_id IN (101, 205, 330) AND ee2.event_id NOT IN (101, 205, 330) GROUP BY e.id ORDER BY shared DESC LIMIT 20;这条 SQL 就是“查询时动态超边”的落地:共享实体越多的事件,和种子的关联越强,排前面。查完这次结果不落库,下次查询重新算,所以不存在维护一张旧图的问题。
4. 验证请求:命中率与延迟怎么测
骨架搭好,得用数据说话。我准备了一个 20 篇文档的小库,其中 5 个问题需要跨两篇以上文档才能答全。验证分两个动作。
第一个动作,测检索命中率。对每个问题,先跑纯向量召回(快速模式),再跑 SQL 扩展(精确模式),看正确答案所在的事件有没有进 top-5。
import sqlite3, requests def vector_search(q, topk=5): # 调 TaoToken Embedding 拿查询向量 r = requests.post( "https://taotoken.net/api/embeddings", headers={"Authorization": "Bearer sk-你的密钥"}, json={"model": "text-embedding-3-small", "input": q} ) qvec = r.json()["data"][0]["embedding"] # 这里用你本地的向量检索逻辑,SQLite 可配 sqlite-vec return local_vec_query(qvec, topk) def sql_expand(seed_ids, limit=20): conn = sqlite3.connect("sag.db") cur = conn.cursor() placeholders = ",".join("?" * len(seed_ids)) cur.execute(f""" SELECT DISTINCT e.id, e.chunk_text, COUNT(*) AS shared FROM event_entities ee1 JOIN event_entities ee2 ON ee1.entity_id = ee2.entity_id JOIN events e ON e.id = ee2.event_id WHERE ee1.event_id IN ({placeholders}) AND ee2.event_id NOT IN ({placeholders}) GROUP BY e.id ORDER BY shared DESC LIMIT ? """, seed_ids + seed_ids + [limit]) return cur.fetchall()实测下来,纯向量在跨文档问题上 top-5 命中 2/5,加上 SQL 扩展后变成 4/5。提升主要来自那些“文档 A 有时间、文档 B 有定价”的题,共享实体把两片串起来了。
第二个动作,测响应延迟。用time.perf_counter()包住整条链路,分别记录 Embedding 耗时、向量召回耗时、SQL 扩展耗时、LLM 精排耗时。我的小库上,快速模式约 380ms,精确模式约 1.2s,多出来的时间主要在 LLM 精排。如果你对延迟敏感,可以把精排改成只对 top-10 做,或者干脆关掉,只用 SQL 扩展的shared排序。
import time t0 = time.perf_counter() seeds = vector_search(question) t1 = time.perf_counter() expanded = sql_expand([s[0] for s in seeds]) t2 = time.perf_counter() print(f"向量召回 {t1-t0:.0f}ms, SQL扩展 {t2-t1:.0f}ms")5. 本篇常见错排查
报错一:no such table: event_entities。建表 SQL 没执行完,或者连到了另一个 db 文件。检查sqlite3 sag.db ".tables",确认三张表都在。PostgreSQL 用户注意 schema 前缀。
报错二:Embedding 返回 401。Key 没填对,或者 base_url 写成了https://taotoken.net/api/v1。正确写法是https://taotoken.net/api,不带/v1。如果还不行,去 https://taotoken.net/api-keys 重建一个 Key 试试。
报错三:SQL 扩展返回空。大概率是实体抽取那步没跑,event_entities表是空的。先SELECT COUNT(*) FROM event_entities确认。如果为 0,检查入库时抽实体的模型调用有没有报错,提示词里的{chunk_text}有没有正确替换。
报错四:查询很慢。没建索引。idx_ee_entity和idx_ee_event这两个索引对 SQL 扩展的性能影响很大,数据量过万后差距明显。另外IN子句里的种子 id 别超过 50 个,太多会拖慢。
报错五:增量更新后新旧事件串不起来。检查新文档的实体有没有和旧实体同名。entities表的name是 UNIQUE,同名实体会复用同一个 entity_id,关联表自然就连上了。如果实体名有大小写或空格差异,先做归一化再入库。
6. 下一步:把 SAG 接进你的编码流
跑通上面这套最小链路后,你可以把检索接口包一层,让 Agent 直接调。SAG 官方有 OpenAI 兼容接口和 MCP 支持,如果你用 Claude Code 或 Codex 这类工具,可以把知识库挂进去,让它在写代码时能检索你的私有文档。长期做编码和 Agent 的话,TaoToken 的 Coding Plan 值得看一下:https://taotoken.net/coding-plan ,一个订阅覆盖多种模型,省得每个项目单独配 Key。
接入文档在这里:https://taotoken.net/doc ,里面有 OpenAI 兼容接口的完整参数说明。控制台在 https://taotoken.net/console ,可以看调用量和余额。模型对话入口 https://taotoken.net/chat 适合快速验证模型通不通。
最后说个实用技巧:SQL 扩展的shared排序可以再加一列时间衰减,让近期事件权重高一点。改ORDER BY shared DESC为ORDER BY shared * (1.0 / (1 + julianday('now') - julianday(e.created_at))) DESC,老事件自然降权,新文档更容易被串进来。这个改动一行 SQL,但对长期知识库的检索质量提升很明显。