做 RAG 应用的人通常都会遇到同一个困惑:知识库里明明有答案,召回的结果却对不上问题。我刚开始做的时候也很天真,以为把文档倒进某个检索服务就行,结果用户用自然语言一问,关键词匹配瞬间露馅。后来在 LangChain 里把向量存储这条链路彻底跑通,才真正理解“利用向量存储实现基于自然语言的相似度检索”到底在解决什么问题。这一篇是 RAG 实践系列的第三篇,我会把向量存储这件事单独拆开:从最基础的原理,到最小可运行的代码,再到工程化落地时容易踩的坑。适合正在搭 RAG、被召回质量折磨的人参考。
1. 先搞明白:语义检索到底比关键词检索多解决了什么
1.1 关键词匹配为什么会在同义改写面前失灵
很多人第一次做知识库问答,第一反应是“用数据库模糊查询不就行了”。真的不行的原因很简单:关键词匹配建立在“用户问题里出现的词,在文档里也会原样出现”的假设上。但用户不会照着内部文档说话。
举个真实场景。我的知识库里有一篇文档叫《员工离职交接操作指引》,里面写“需要归还工牌、电脑、门禁卡”。用户来问的是“我要走了,公司要我交什么东西”。这句话和文档标题之间几乎没有重复词,按词重叠打分的那套算法会给出很低的分数,哪怕这篇文档就是用户要的答案。
更麻烦的是中文里的同义表达特别多。“报销”和“走费用流程”、“退款”和“钱退回来”、“逾期”和“超期没有付”,这些说法在字符层面完全不一样,但在语义层面指向同一件事。关键词检索没有办法处理这种差异,于是召回质量就只能取决于用户会不会“猜中”文档里的词。这显然不是一个可持续的方案。
1.2 向量的核心假设:语义相近 = 坐标相近
向量检索的思路是把文本变成一串数字,让这串数字去表达文本的含义。所谓 embedding,就是用一个模型把一段文字映射到高维空间里的一个点。这个空间里的坐标不是随便给的,模型在训练的时候见过大量“上下文相似”的句子,于是会把语义相近的文本推近,把语义无关的文本拉远。
所以“利用向量存储实现基于自然语言的相似度检索”这句话,翻译过来就是:把文档片段和用户问题都扔进同一个向量空间,然后计算问题对应的点离哪些文档片段对应的点最近。最近的那些片段,就是候选答案。
这里要提醒一句:向量相似度不是模型“理解”了文本,而是模型利用了海量文本中的共现规律。比如“退款”和“退款到账时间”在大量语料里经常出现在相似位置,模型就会觉得它俩靠近。这种统计规律的好处是泛化能力强,坏处是它并不擅长处理数字、否定、精确条件这类逻辑关系。后面我会专门讲这些失败模式。
向量之间的相近程度一般用余弦相似度来算:
cosine(A, B) = (A · B) / (|A| * |B|)余弦相似度只关心方向,不关心向量的长度,所以它对文本长度不那么敏感。这也是它在文本检索里比欧氏距离更常用的原因。
2. 用 LangChain 搭一个最小可用的向量检索链路
2.1 先把文档切成适合检索的块
向量检索有一个前提:你不能把整本手册塞进一个向量里。整篇文档的语义太复杂,一个向量很难表达清楚,检索时也容易错位。所以第一步是分块。
我在中文场景下一般用 LangChain 的文本分割器,同时把中文标点加进分隔符,否则很多分块会把一句话从中间劈开。
from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter raw_text = open("handbook.txt", encoding="utf-8").read() splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80, separators=["\n\n", "\n", "。", ";", "!", "?"], ) chunks = splitter.split_text(raw_text) docs = [ Document(page_content=chunk, metadata={"chunk_id": i}) for i, chunk in enumerate(chunks) ]chunk_size=500和chunk_overlap=80是我在通用文档上的默认值。500 个 token 左右的块,既能给大模型提供相对完整的上下文,又不至于让 embedding 把太多无关信息压进一个向量里。overlap的作用是让相邻块之间保留一部分上下文,避免一句话刚好被切到两块里,导致哪一块都说不清楚。
2.2 自己写一个极简向量库,看清核心逻辑
很多人一上来就接重型向量数据库,结果出了问题也不知道是嵌入问题、分块问题还是数据库配置问题。为了把链路看清,我先写了一个极简向量库,只实现两个方法:add_documents和similarity_search。不要小看这两个方法,LangChain 里所有向量存储的抽象核心就是它们。
import math class MiniVectorStore: def __init__(self, embeddings): self.embeddings = embeddings self._documents = [] self._vectors = [] def add_documents(self, documents): for doc in documents: vector = self.embeddings.embed_documents([doc.page_content])[0] self._documents.append(doc) self._vectors.append(vector) def _cosine(self, a, b): dot = sum(x * y for x, y in zip(a, b)) na = math.sqrt(sum(x * x for x in a)) nb = math.sqrt(sum(x * x for x in b)) return dot / (na * nb) if na and nb else 0.0 def similarity_search(self, query, k=4): query_vector = self.embeddings.embed_query(query) scored = [ (self._cosine(query_vector, vec), idx) for idx, vec in enumerate(self._vectors) ] scored.sort(key=lambda x: x[0], reverse=True) return [self._documents[idx] for _, idx in scored[:k]]这段代码做的事情很简单:文档入库时先算一次 embedding,把向量和原始文档一起存起来;检索时把用户问题也变成向量,逐个和库里所有向量算余弦相似度,排序后返回得分最高的 k 个文档。
这个实现虽然粗糙,但它的结构正好对应 LangChain 里向量存储的通用接口。真实生产环境不会用这种线性扫描,因为数据量一上来速度会很难看,但理解了这个核心逻辑,再去用那些带 ANN 索引的向量库,你就能看懂它们到底在优化哪一步。
2.3 跑通第一次自然语言检索
接入 embedding 之后,第一次检索其实非常简单:
embeddings = YourEmbeddings(...) # 替换成你的 embedding 模型 store = MiniVectorStore(embeddings) store.add_documents(docs) for doc in store.similarity_search("甲方逾期了要怎么赔偿?", k=3): print(doc.page_content)如果你的文档里有一句“乙方逾期付款的,应按合同总金额每日万分之五支付违约金”,这句会被检索出来。用户用的“赔偿”和文档里的“违约金”在字面上不是一个词,但向量空间里它们离得很近。
这里有一个小坑:文档入库时我调用的是embed_documents,用户查询时调用的是embed_query。很多 embedding 模型对查询和文档这两种输入采用不同的指令前缀,分开调用是为了让模型知道“这是查询”还是“这是要被检索的文本”。如果你自己封装 embedding,千万别图省事只用一个方法。
3. Embedding 模型与向量库的选型逻辑
3.1 Embedding 模型:先看语言对齐,再看参数规模
选 embedding 模型时,最容易犯的错是“谁的参数多选谁”。参数多不代表检索效果好。对中文场景来说,更重要的是模型在中文语料上的对齐程度,以及它输出的向量维度是否匹配你的向量库和后续算法。
我一般按这三个维度去选:
第一,是否覆盖你的目标语言。中文知识库用中文友好型模型,英文技术文档用英文模型,混合语料要专门看多语言支持,否则中文会明显吃亏。
第二,向量维度是否合理。常见有 300 多、700 多、1000 多甚至更高的维度。维度越高,表达粒度越细,但存储成本、计算成本都会上升。中小规模知识库没必要追求超高维度。
第三,API 还是本地部署。本地小模型部署简单、隐私可控、单次调用成本低,但效果通常不如商用 API;商用 API 效果好,但要考虑数据出境、调用延迟和费用。放到实际项目里,数据敏感程度往往比“效果最好”更重要。
下面是我在项目里常用的比较维度:
| 类型 | 部署成本 | 检索效果 | 隐私可控 | 适合阶段 |
|---|---|---|---|---|
| 本地小模型 | 低,单人可跑 | 中等偏上 | 高 | 原型验证、私有数据 |
| 商用 API | 高一些,按量计费 | 强 | 取决于服务方 | 效果优先、数据合规 |
| 领域微调模型 | 很高,需要训练 | 领域内最强 | 高 | 专业术语极重的场景 |
3.2 向量库:文件型、独立服务、云托管怎么选
有了 embedding,下一步就是选向量存储。很多人一上来就部署独立服务,其实多数场景用不上。
如果你的数据量在十万级以内,检索频率也不高,文件型本地向量库完全够用。它把索引保存在本地文件里,进程重启后可以直接加载,不需要额外维护服务。好处是简单,坏处是并发和水平扩展能力弱。
数据量到了百万级,或者会有多个服务并发访问同一个索引,就需要独立向量数据库服务。这类服务通常提供 HNSW 一类的近似最近邻索引,检索速度和并发能力都强很多,代价是要处理部署、内存、索引重建、备份这些问题。
再往上就是托管向量库,运维成本最低,但数据和费用都要考虑清楚。我的建议是:不要在原型阶段为了“架构先进”引入重型组件,先用能跑通链路的方案,等数据量真正上去再迁移。
3.3 一个可参考的配置组合
假设我有一套中文制度文档,大概 20 万个分块。我会先用一个本地部署的中文 embedding 小模型,维度在 700 左右,向量库用文件型本地索引即可。按每个向量 700 个 float 计算,每个向量约 2.8KB,20 万块大约是 560MB 内存,即使全部加载也完全可以接受。
检索参数上,我一般先召回k=10,再做一轮重排,最终只把得分最高的 3 到 5 个片段交给大模型。这样能避免“只取 top-1 结果但 top-1 恰好不对”的尴尬。重排可以简单用 LLM 打分,也可以用专门的交叉编码器模型,但那是另一个话题了。
4. 检索质量翻车排查:常见失败模式与对策
4.1 分块粒度不匹配导致上下文残缺或语义稀释
检索质量出现问题,先别怀疑 embedding 模型,多半是分块参数没调对。
分块太小,比如一句话一个块,确实容易命中,但给到大模型的上下文往往缺前因后果。用户问“这个方案要审批吗”,如果命中的块只写了“需要审批”,没有写明“由谁审批、多久审批完”,大模型只能靠猜。
分块太大,比如把整章塞进一个块,一个向量要表达十几个主题,查询语义被稀释,结果就是什么都像,什么都不准。
我的做法是:先按文档结构分,再按长度微调。制度、合同这类有明确章节的文档,可以优先按二级标题切块;FAQ 这类短问答,块可以小到 200 token 左右;技术方案这种长段落,块可以放到 800 token。不要一个参数打天下。
4.2 元数据被忽略,检索结果缺乏过滤能力
向量检索不是只能靠语义相似度。LangChain 的向量存储接口里通常都支持filter参数,用 metadata 做前置过滤,能极大减少无效候选。
我踩过的一个坑是:知识库里有往年多版制度文件,用户问“今年的考勤规则是怎么规定的”。如果不过滤年份,最相似的可能是去年的一份文档,因为语义高度接近,但年份不对。后来我把year、doc_type、source这些字段放进 metadata,检索时先按条件过滤,再算相似度,准确率立刻上了一个台阶。
代码层面,真实 LangChain 向量库的调用大概长这样:
docs = vectorstore.similarity_search( "今年的考勤规则是怎么规定的", k=5, filter={"year": 2025, "doc_type": "制度"} )所以分块时不要只存page_content,metadata 要尽可能完整:
| 字段 | 说明 |
|---|---|
| source | 来源文件名 |
| chunk_id | 块序号 |
| section | 所属章节 |
| year | 生效年份 |
| doc_id | 文档唯一标识 |
4.3 相似度分数不能直接当置信度
另一个容易误导人的东西是相似度分数。很多人看到分数 0.8 就以为是“80% 正确”,这是完全错误的。
不同 embedding 模型的分数分布天差地别。有的模型正常相似文本在 0.7 到 0.9 之间,有的模型可能全集中在 0.4 到 0.6。统一设一个 0.7 阈值,同一个项目里可能过滤掉所有有效结果,也可能放进来一堆无关内容。
所以我的经验是:不要在代码里硬编码阈值,而是先跑一批人工标注的 query 和正确答案,看这个模型下“正确命中”和“错误命中”的分界在哪里。并且在实际链路里,不要把 top-1 直接交给大模型,最好先召回 k 个再重排。
还要意识到,向量检索对某些类型的 query 天然不友好。否定关系就是典型例子。“我不想要自动续费”和“我想要自动续费”在向量空间里的距离可能比想象中近得多。数字条件也一样,“价格大于 500”和“价格小于 500”的问题,向量模型很难精确区分。这种场景下,要么做查询改写,要么搭配其他检索手段,不要指望向量检索包打天下。
5. 从相似度检索到完整 RAG 生成链路
5.1 retriever 是向量库和大模型之间的桥梁
向量库本身只负责找候选片段,真正回答用户问题的是大模型。LangChain 里用一个retriever对象把两者连接起来。它本质上就是把相似度检索包装成一个可调用的接口,输入问题,输出文档列表。
retriever = store.as_retriever(search_kwargs={"k": 5}) docs = retriever.invoke("客户问退款多久到账")这里的好处是,后面如果要切换检索策略,比如从similarity改成mmr,或者加一层自定义重排,只要替换 retriever 的构造方式,RAG 链路的其他部分不用动。
5.2 让上下文真正可控:组装、截断、引用
检索到的文档不能直接整堆丢进 prompt。要控制长度、控制格式、控制引用。
我会先把命中的文档按“来源 + 内容”拼起来,这样大模型在回答时可以引用具体文档位置,我也可以在最终答案里带出溯源信息。
def format_docs(docs): return "\n\n".join( f"[来源:{doc.metadata.get('source', '未知')}] {doc.page_content}" for doc in docs )另外必须在 prompt 里说清楚边界。我发现稍微强势一点的约束非常有帮助,比如“只根据提供的资料回答,资料里没有的内容直接说明不知道”。否则大模型一遇到模糊问题就开始自由发挥,RAG 的防幻觉能力就白做了。
5.3 一个可复用的 LCEL 示例
新版 LangChain 我更推荐用 LCEL 写链路,因为中间每一步都可以插入自定义处理,调试起来很直观。
from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableParallel, RunnablePassthrough prompt = ChatPromptTemplate.from_template( """ 你是一个企业知识库问答助手。请只根据下面提供的资料回答问题。 如果资料中没有相关信息,请直接回答“没有找到相关资料”。 资料: {context} 问题: {question} 回答: """ ) rag_chain = ( RunnableParallel( context=retriever | format_docs, question=RunnablePassthrough(), ) | prompt | llm | StrOutputParser() )RunnableParallel会让 retriever 和问题本身并行执行,retriever 拿到文档后进入format_docs格式化,最终和原始问题一起填进 prompt。llm换成你自己项目里配置的语言模型实例即可,关键是这个链路可以在任意一步插入日志、重排、过滤,而不需要重写整条链。
6. 离线工程化收尾:持久化、增量更新与可观测性
6.1 持久化:别每次都重新 embedding 全量文档
原型阶段我经常直接内存里建库,服务一重启就全部重新嵌入。文档一多,每次重建都是十几分钟起步。必须把向量索引持久化到磁盘。
LangChain 里不同向量库的持久化 API 不一样,有的是在构造时指定persist_directory,有的是调用save_local/load_local。核心逻辑是一样的:先检查索引文件是否存在,存在就加载,不存在才重建。
import os if os.path.exists("index/vector_store"): store = LoadVectorStore("index/vector_store", embeddings) else: store = BuildVectorStore(embeddings) store.add_documents(docs) store.persist("index/vector_store")这样日常重启服务就不会产生额外成本。记得把索引目录纳入备份范围,否则误删索引文件等于要重新算所有 embedding。
6.2 增量更新:用 doc_id 去重,避免索引膨胀
知识库是活的,文档会改、会删、会加。如果每次全量重建,成本太高;如果只做增量,又要处理“同一个文档的新版本和旧版本同时存在”的问题。
我会给每个文档块生成一个稳定 ID,一般用source + section + chunk_id或者对全文做 hash:
import hashlib def build_doc_id(content): return hashlib.md5(content.encode("utf-8")).hexdigest()增量更新的流程是:先存一份doc_id -> 向量存储内部ID的映射。新增文档时,先查这个 doc_id 是否已存在;存在就更新对应向量,不存在就插入;删除来源文件时,把它对应的所有 doc_id 找出来逐一删除。
这个映射不用搞得很复杂,一个 JSON 文件或一张 SQLite 表就够了。它本身不是热点数据,真正怕的是索引里堆积大量过期向量,导致检索时总召回旧文档。
6.3 可观测性:把检索命中记录下来,才能持续调优
最后一点是我最想提醒的:检索链路一定要留下日志。
我在项目里的做法是,每次检索都把 query、recall 出来的 top-k 文档、相似度分数、最终选用的片段、最终答案一起落日志。落在同一个 request_id 下,出问题的时候能直接回放整条链路。
季度复盘时,我再挑 20 个人工标注的难题,跑一遍老版本链路和新版本链路,看 recall@5 有没有提升。没有这种回归集,所有“优化”都只是在自我感动。很多时候你以为调了模型参数有效果,其实只是换了测试用例而已。
我在实际项目里最大的体会是:向量存储只是 RAG 的发动机,真正决定跑多远的,是分块、metadata、重排和观测这些看起来不那么酷的环节。把这条链路每一步都留下记录,后面再想优化,就有了底数,而不是靠猜。