SQLite FTS5 替代云向量数据库:RAG 检索管线 10 倍提升实战
原文:https://dev.to/cagrik34/i-ditched-cloud-vector-databases-for-sqlite-fts5-and-my-rag-pipeline-got-10x-better-759(作者 @cagrik34)
几个月前,我在为一个内部工程仓库调试 RAG 管线。一位开发者输入了:
"`config_v2.py` 中 worker pool #4 的超时限制是多少?"
系统后端使用了一个流行的托管向量数据库和最先进的余弦嵌入,自信地检索出了五段关于工作线程最佳实践、微服务弹性和线程池扩展模式的内容。
它完全漏掉了 config_v2.py 中定义 WORKER_POOL_4_TIMEOUT = 45 的那行单行注释。
为什么?因为从数学上讲,变量名和精确整数与一个关于架构的抽象问题之间没有很高的语义相似度。
那天晚上,我审查了基础设施账单。我们每个月花几百美元维护一组向量实例,存储的文本和代码不到 80 MB。我们增加了网络跳数、冷启动、API 速率限制和额外的运维复杂度——结果在工程师最关心的检索准确率上反而更差:精确关键词、错误码、版本号和文件路径。
我决定推倒重来,从第一性原理重新构建。
不用云数据库。不用重量级容器。只用 SQLite FTS5、稠密嵌入和 Reciprocal Rank Fusion(RRF)。
下面解释为什么这种混合架构能稳定胜过纯语义搜索,以及如何用不到 60 行干净的 Python 代码实现它。
纯语义搜索的盲区
稠密向量搜索擅长理解意图。如果用户问"如何重启服务器?",语义搜索能轻松找到一篇解释"重启你的实例"的文档。
但在生产系统中,用户不只是问抽象问题。他们搜索的是:
- 错误码:
ERR_CONN_REFUSED_0x82 - 部件/型号编号:
SKU-4982-A - 精确标识符:
def compute_rrf_rank(...) - 财务与合同数字:
$14,250 quarterly budget
稠密嵌入将这些唯一标记压缩到一个稠密潜在空间中,抹平了赋予精确标记身份的棱角。结果就是检索中的语义幻觉。
另一方面,经典的 BM25 词法搜索(Lucene 和 Elasticsearch 的基础)恰恰在向量搜索失败的地方表现出色:精确标记匹配、词频和逆文档频率。
解决方案不是二选一,而是将两者结合。
┌──────────────────────┐
│ User Query │
└──────────┬───────────┘
│
┌────────────────┴────────────────┐
▼ ▼
┌──────────────────┐ ┌──────────────────┐
│ SQLite FTS5 │ │ Dense Embeddings │
│ (BM25 Sparse) │ │ (Cosine Vector) │
└────────┬─────────┘ └────────┬─────────┘
│ │
Top-K Ranked List Top-K Ranked List
│ │
└────────────────┬────────────────┘
▼
┌───────────────────────────┐
│ Reciprocal Rank Fusion │
│ (k = 60) │
└─────────────┬─────────────┘
▼
Final Hybrid Top Hits
为什么选 SQLite?
大多数工程师忘了,SQLite 标准库中已经内置了地球上最快的全文搜索引擎之一(FTS5)。
- 进程内运行,零网络延迟。
- 无需守护进程管理或外部云凭证。
- 使用 BM25 评分,支持自定义分词器(
unicode61、trigram、前缀匹配)。 - 持久化到单个可移植的
.db文件,可以提交到 Git 或挂载到任何地方。
当你将 SQLite FTS5 与本地嵌入模型(或 Google 的 gemini-embedding-001、Cohere Embed 等快速 API)配合使用时,你就拥有了一个完整的、自包含的搜索引擎。
秘密武器:Reciprocal Rank Fusion(RRF)
当你同时运行 BM25 和余弦相似度时,会得到两组分数:
- BM25 分数是无界的正浮点数(如
8.45、14.20、2.10)。 - 余弦分数在
-1.0到1.0(或0.0到1.0)之间有界。
试图归一化并相加这些原始分数是个陷阱。如果某个查询产生了一个极大的 BM25 异常值,它会完全淹没语义引擎的信号。
这就是 Reciprocal Rank Fusion(RRF) 的用武之地。
RRF 不看任意的分数数值,而是看每个引擎结果的排名顺序:
$$RRF(d) = \sum_{m \in M} \frac{1}{k + r_m(d)}$$
其中:
- $r_m(d)$ 是文档 $d$ 在系统 $m$ 中的位置(排名 $1, 2, 3...$)。
- $k$ 是常数平滑参数(研究中的标准默认值是 $60$)。
如果一篇文档在 BM25 中排名第一、在语义搜索中排名第二,它会获得巨大的加权。如果它只在一个引擎中出现在第 20 名,其分数会平滑衰减。这种方法是确定性的,不受分数尺度不匹配的影响,且无需任何超参数调优。
代码:50 行实现可用的混合搜索引擎
以下是一个使用标准 Python 和 SQLite 的最小化完整实现,可以直接复制运行:
import sqlite3
import math
from typing import List, Dict, Tuple
class LocalHybridSearch:
def __init__(self, db_path: str = ":memory:"):
self.conn = sqlite3.connect(db_path)
self.cursor = self.conn.cursor()
self._setup_db()
self.vectors = {} # doc_id -> list[float]
def _setup_db(self):
self.cursor.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS docs_fts USING fts5(
doc_id UNINDEXED,
title,
content,
tokenize='unicode61 remove_diacritics 2'
);
""")
self.conn.commit()
def add_document(self, doc_id: str, title: str, content: str, embedding: List[float]):
self.cursor.execute(
"INSERT INTO docs_fts(doc_id, title, content) VALUES (?, ?, ?)",
(doc_id, title, content)
)
self.conn.commit()
self.vectors[doc_id] = embedding
def _bm25_search(self, query: str, top_k: int = 20) -> List[Tuple[str, float]]:
# 清理查询词以适配 FTS5 语法
clean_tokens = [f'"{w}"' for w in query.replace('"', '').split() if w.strip()]
if not clean_tokens:
return []
match_expr = " OR ".join(clean_tokens)
self.cursor.execute("""
SELECT doc_id, bm25(docs_fts) as score
FROM docs_fts
WHERE docs_fts MATCH ?
ORDER BY score ASC LIMIT ?
""", (match_expr, top_k))
# SQLite bm25() 返回值越低/越负表示匹配度越高
return [(row[0], abs(float(row[1]))) for row in self.cursor.fetchall()]
def _dense_search(self, query_vec: List[float], top_k: int = 20) -> List[Tuple[str, float]]:
def cosine_sim(a: List[float], b: List[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x * x for x in a))
norm_b = math.sqrt(sum(y * y for y in b))
return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0
scores = [(doc_id, cosine_sim(query_vec, vec)) for doc_id, vec in self.vectors.items()]
scores.sort(key=lambda x: x[1], reverse=True)
return scores[:top_k]
def search(self, query: str, query_vec: List[float], top_k: int = 5, k_rrf: int = 60) -> List[Dict]:
bm25_results = self._bm25_search(query, top_k=20)
dense_results = self._dense_search(query_vec, top_k=20)
rrf_scores = {}
for rank, (doc_id, _) in enumerate(bm25_results, start=1):
rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + (1.0 / (k_rrf + rank))
for rank, (doc_id, _) in enumerate(dense_results, start=1):
rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + (1.0 / (k_rrf + rank))
sorted_hits = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
return [{"doc_id": doc_id, "rrf_score": round(score, 4)} for doc_id, score in sorted_hits]
生产部署:构建 hybrid-rag-action
我将这套架构模式打包成了一个开源工具——Hybrid RAG GitHub Action。
看到这套架构在 GitHub Marketplace(Microsoft/GitHub 生态) 上被正式应用、验证并发布,这是一个令人自豪且倍感谦卑的里程碑。
每当仓库中新建 issue 或 pull request 时,该 Action 会:
- 对代码库进行索引(对函数定义、markdown 文档和代码文件进行 AST 感知解析)。
- 运行 FTS5 BM25 + 稠密向量嵌入余弦相似度检索。
- 用 RRF 融合两路排序结果。
- 生成自动分拣响应,引用相关代码所在的确切行号。
由于零外部数据库依赖,整条流水线直接在 GitHub Actions runner 中运行,耗时不到 3 秒,基础设施开销为零。
除了上述实际集成之外,同一架构还作为参考方案提交并分享到了 10+ 个主流开源 AI 生态和仓库中,包括 Google Gemini Cookbook、Meta Llama Cookbook 和 Stanford DSPy。
什么时候你真正需要专用向量数据库?
实事求是地说,在以下情况下你确实需要 Milvus、Qdrant 或 Pinecone:
- 你要索引 1 亿+ 条向量,无法全部装入内存或单机 NVMe 硬盘。
- 你需要跨多个可用区进行分布式水平分片。
- 你需要在大规模场景下实现多租户安全隔离。
但如果你的语料库规模在 100 万条 chunk 以下(这涵盖了约 90% 的企业内部知识库、代码库和文档站点),那么搭建云端向量集群就是严重的过度工程。
SQLite FTS5 + 内存或 SQLite 存储的稠密向量相似度检索能给你带来:
- 即时部署: 无需配置,零 docker 配置。
- 零云基础设施费用。
- 亚毫秒级检索延迟。
- 坚如磐石的精确关键词匹配,同时不牺牲语义理解能力。
总结思考
过去两年,AI 行业一直在向工程师们灌输一个观念:万物皆需向量化。
向量确实强大,但语言是微妙的。有时候,找到 ERR_404_NULL_POINTER 的最佳方式不是通过 1536 维的余弦夹角,而是通过经典的倒排索引词元匹配。
在你的下一个 RAG 项目中试试 SQLite FTS5 + RRF 吧。你也许会发现,自己根本不需要再订阅一个 SaaS 服务。
欢迎在下方评论区分享你对混合检索的想法和经验。
开源项目与链接:
- GitHub Action(Marketplace): Cagrik34/hybrid-rag-action
- GitHub 主页: @Cagrik34
原文:https://dev.to/cagrik34/i-ditched-cloud-vector-databases-for-sqlite-fts5-and-my-rag-pipeline-got-10x-better-759(作者 @cagrik34)