1. 为什么知识获取管道是 AI Agent 的分水岭
做 AI Agent 开发的人,绕不开一个尴尬的现实:模型本身很聪明,但它对你私有的业务知识一无所知。你问它公司内部的报销流程,它给你编一个看起来很像那么回事的答案;你让它查某个产品的技术参数,它张口就来一段似是而非的描述。这不是模型不行,而是它的知识边界停在了训练数据截止的那一天,而你的业务知识每天都在更新。
知识获取管道要解决的就是这个问题。它的核心任务只有一句话:在 Agent 需要知识的时候,把正确、最新、可溯源的信息,准确送到模型的上下文里。而RAG(Retrieval-Augmented Generation,检索增强生成)就是目前最成熟、最工程化的一条实现路径。
我见过太多团队在搭 Agent 时把 90% 的精力花在 Prompt 调优和工具编排上,结果上线后用户一问具体业务问题就翻车。根因往往不在生成端,而在检索端——知识根本没被正确地"喂"进去。所以我把 RAG 放在这个系列的第四篇,因为它是 Agent 从"能聊天"走向"能干活"的关键基础设施。
这篇文章适合三类人:正在从 0 到 1 搭建 AI Agent 的开发者、需要给企业级应用接入私有知识库的工程师、以及想搞清楚 RAG 到底怎么回事的产品和技术负责人。我会从整体设计思路讲到稠密嵌入与稀疏嵌入的选型,再到完整的实操流程和踩坑记录,尽量把每个"为什么"都讲透。
2. RAG 知识获取管道的整体设计与思路拆解
2.1 一条完整的知识管道到底包含哪些环节
很多人对 RAG 的理解停留在"把文档切块、向量化、存进向量库、检索出来拼进 Prompt"这四步。这个理解不算错,但太粗了,粗到落地时处处是坑。一条真正能扛住生产环境的知识获取管道,至少包含六个环节:
- 数据接入:从各种来源(文件、数据库、API、网页)把原始知识捞进来
- 解析与清洗:把 PDF、Word、HTML 这些格式拆成纯文本,去掉页眉页脚、乱码、重复内容
- 切分(Chunking):把长文本切成适合检索和嵌入的片段
- 嵌入(Embedding):把文本片段转成向量,这里就涉及稠密嵌入和稀疏嵌入的选择
- 存储与索引:把向量和原文存进向量数据库,建立可快速检索的索引
- 检索与重排:根据用户 query 召回候选片段,再用重排模型精排,最后送给 LLM
这六个环节里,任何一个掉链子,最终效果都会崩。我见过最典型的案例是:切分策略没设计好,一个完整的操作步骤被从中间切断,检索时只召回了一半,模型拿着半截信息生成答案,用户照着做直接出错。
2.2 为什么是 RAG 而不是微调
这是每个团队都会问的问题。我的答案很直接:绝大多数场景下,RAG 的性价比远高于微调。
微调的本质是把知识"烧"进模型参数里,代价是每次知识更新都要重新训练,成本高、周期长,而且模型对训练时没见过的知识依然无能为力。RAG 的本质是把知识放在外部,模型只负责"阅读理解",知识更新只需要更新向量库,分钟级就能生效。
打个比方:微调像是把整本教材背下来的学生,背完就固定了;RAG 像是带着参考书进考场的学生,随时能翻到最新版本。对于知识频繁变动、需要溯源、需要权限控制的业务场景,RAG 几乎是唯一合理的选择。
当然,RAG 也不是万能的。它依赖检索质量,如果知识本身结构混乱、表述模糊,检索效果会很差。这时候就需要在数据治理上下功夫,而不是指望换个更强的模型就能解决。
2.3 稠密嵌入与稀疏嵌入:两条腿走路才稳
这是本篇最核心的技术点,也是很多 RAG 项目效果上不去的根因。
稠密嵌入(Dense Embedding)把一段文本映射成一个固定长度的高维向量,比如 768 维或 1024 维。它的优势是能捕捉语义相似性——用户问"怎么报销差旅费",即使文档里写的是"出差费用申请流程",稠密向量也能把它们拉到相近的位置。缺点是它对精确的关键词匹配不敏感,比如产品型号"XR-2000A"这种,稠密向量可能召回一堆语义相近但型号不对的内容。
稀疏嵌入(Sparse Embedding)则是传统的词频类表示,比如 BM25。它把文本表示成一个稀疏向量,维度等于词表大小,大部分位置是 0,只有出现的词才有值。它的优势是精确匹配强,用户搜什么词就匹配什么词,对专有名词、型号、代码标识符特别有效。缺点是无法理解语义,"报销"和"费用申请"在它眼里是两个完全不同的词。
我的经验是:生产级 RAG 一定要用混合检索(Hybrid Retrieval),把稠密和稀疏的结果融合起来。纯稠密检索在语义泛化上强,但会漏掉精确匹配;纯稀疏检索精确但不懂语义。两者结合,再配一个重排模型,召回率和准确率都会有明显提升。
| 维度 | 稠密嵌入 | 稀疏嵌入 |
|---|---|---|
| 表示方式 | 固定维度稠密向量 | 词表维度稀疏向量 |
| 语义理解 | 强 | 弱 |
| 精确匹配 | 弱 | 强 |
| 典型算法 | BGE、M3E、text-embedding | BM25、SPLADE |
| 适用场景 | 语义问答、模糊查询 | 型号检索、代码搜索 |
| 存储开销 | 中等 | 较大(词表维度) |
| 检索速度 | 快(ANN 索引) | 快(倒排索引) |
2.4 方案选型的几个关键决策点
在动手之前,有几个决策点必须先想清楚,否则后面返工成本很高。
第一,向量库选什么。小规模验证阶段,用 FAISS 或 Chroma 就够了,轻量、零运维。生产环境如果数据量上百万级,建议用 Milvus、Qdrant 或 Weaviate 这类专业向量库,支持分布式、支持混合检索、支持元数据过滤。如果团队已经在用 Elasticsearch,它的向量检索能力也够用,能省一套运维。
第二,嵌入模型选什么。中文场景我一般推荐 BGE 系列或 M3E,在中文语义任务上表现稳定。如果对多语言有要求,可以看 multilingual 的模型。模型维度不是越高越好,1024 维在大多数场景已经够用,维度太高会拖慢检索速度、增加存储成本。
第三,要不要上重排。我的建议是:只要召回数量超过 10 条,就应该上重排。重排模型(如 BGE-Reranker)会对召回的候选做精细打分,把真正相关的排到前面。实测下来,加了重排之后,Top-3 的命中率能提升 20% 以上。
3. 核心细节解析与实操要点
3.1 文档切分:最容易被低估的环节
切分策略直接决定了检索质量的上限。切得太粗,一个 chunk 里混了好几个主题,检索时噪声大;切得太细,一个完整语义被拆散,模型拿到的是碎片。
我常用的策略是递归字符切分 + 语义边界保护。具体做法是:优先按段落切,段落太长再按句子切,句子还长才按字符硬切。同时设置一个重叠窗口(overlap),一般取 chunk 大小的 10% 到 20%,保证跨 chunk 的语义连续性。
from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=512, chunk_overlap=64, separators=["\n\n", "\n", "。", "!", "?", ";", " ", ""], length_function=len, ) chunks = splitter.split_text(document_text)这里有几个参数需要根据业务调。chunk_size我一般从 512 起步,技术文档可以到 800,对话记录可以降到 256。chunk_overlap不要超过 chunk_size 的 20%,否则会有大量冗余。separators的顺序很重要,中文场景一定要把中文标点加进去,否则会按空格硬切,把句子切得七零八落。
注意:切分时一定要保留元数据,比如来源文件名、页码、章节标题。这些元数据在检索时可以用来做过滤,在生成时可以用来做引用溯源。没有元数据的 RAG 系统,用户问"这个答案从哪来的"你就答不上来。
3.2 稠密嵌入的实操细节
稠密嵌入的核心是选对模型和用好模型。模型选型上,我建议先在自己的业务数据上做一个小规模评测,别盲目追榜单。评测方法是:准备 50 到 100 个真实 query,人工标注每个 query 对应的正确文档,然后看不同模型在这些 query 上的召回率。
嵌入时有两个细节容易踩坑。一是归一化,大多数嵌入模型输出的向量需要做 L2 归一化,这样余弦相似度才能正确计算。二是批量处理,嵌入是计算密集型操作,一定要批量调用,batch size 根据显存调整,一般 32 到 128 之间。
from sentence_transformers import SentenceTransformer import numpy as np model = SentenceTransformer("BAAI/bge-large-zh-v1.5") def embed_texts(texts, batch_size=64): embeddings = model.encode( texts, batch_size=batch_size, normalize_embeddings=True, show_progress_bar=True, ) return np.array(embeddings)normalize_embeddings=True这一步千万别省,省了之后相似度计算会出问题,而且这种问题很隐蔽,往往要排查很久才发现。
3.3 稀疏嵌入与 BM25 的落地
稀疏嵌入这块,BM25 是最经典也最实用的选择。它的核心思想是:一个词在当前文档中出现次数越多、在所有文档中出现次数越少,它的区分度就越高。
BM25 有两个关键参数,k1 和 b。k1 控制词频饱和,一般取 1.2 到 2.0;b 控制文档长度归一化,一般取 0.75。这两个参数不用太纠结,默认值在大多数场景都够用。
from rank_bm25 import BM25Okapi import jieba # 中文需要先分词 tokenized_corpus = [list(jieba.cut(doc)) for doc in documents] bm25 = BM25Okapi(tokenized_corpus, k1=1.5, b=0.75) query_tokens = list(jieba.cut(query)) scores = bm25.get_scores(query_tokens) top_indices = np.argsort(scores)[::-1][:10]中文场景下分词是必须的,jieba 是最常用的选择。如果你的领域有大量专有名词,建议自定义词典,否则分词会把"知识图谱"切成"知识"和"图谱",影响匹配效果。
3.4 混合检索的融合策略
稠密和稀疏的结果怎么融合,有两种主流做法。
一种是加权求和,把两路归一化后的分数按权重相加。权重一般稠密给 0.6 到 0.7,稀疏给 0.3 到 0.4。这种做法的优点是简单可控,缺点是需要调权重。
另一种是 RRF(Reciprocal Rank Fusion,倒数排名融合),不看分数只看排名,把两路结果的排名做倒数加权。RRF 的好处是不用归一化、不用调权重,鲁棒性更好,我一般优先用这个。
def rrf_fusion(dense_results, sparse_results, k=60): scores = {} for rank, doc_id in enumerate(dense_results): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1) for rank, doc_id in enumerate(sparse_results): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1) return sorted(scores.items(), key=lambda x: x[1], reverse=True)k 值一般取 60,这是 RRF 论文里的推荐值,实测下来也比较稳。
3.5 重排:把真正相关的顶上来
召回阶段追求的是"不漏",重排阶段追求的是"排准"。重排模型通常是交叉编码器(Cross-Encoder),它把 query 和候选文档拼在一起输入模型,直接输出相关性分数。这种方式比向量点积精确得多,但计算量大,所以只能用在召回后的少量候选上。
from sentence_transformers import CrossEncoder reranker = CrossEncoder("BAAI/bge-reranker-large") pairs = [(query, doc) for doc in candidate_docs] scores = reranker.predict(pairs) reranked = sorted(zip(candidate_docs, scores), key=lambda x: x[1], reverse=True)重排的候选数量一般控制在 20 到 50 条,太少会漏,太多会慢。最终送给 LLM 的片段控制在 3 到 5 条,太多会超出上下文窗口,也会引入噪声。
4. 实操过程与核心环节实现
4.1 环境准备与依赖安装
先把环境搭起来。我用的技术栈是 Python + LangChain + BGE 嵌入模型 + Qdrant 向量库 + BM25 稀疏检索。这套组合在中文场景下成熟稳定,社区资料也多。
pip install langchain langchain-community sentence-transformers pip install qdrant-client rank-bm25 jieba pypdfQdrant 我选择用 Docker 起一个本地实例,方便调试。生产环境可以换成集群部署。
docker run -d -p 6333:6333 -v $(pwd)/qdrant_storage:/qdrant/storage qdrant/qdrant4.2 数据接入与解析
假设我们的知识源是一批 PDF 技术文档。解析 PDF 是个脏活,pypdf 能处理大部分标准 PDF,但扫描件需要 OCR,复杂排版可能解析错乱。
from pypdf import PdfReader def parse_pdf(file_path): reader = PdfReader(file_path) pages = [] for i, page in enumerate(reader.pages): text = page.extract_text() if text.strip(): pages.append({"text": text, "page": i + 1, "source": file_path}) return pages解析完一定要做清洗:去掉多余空行、合并断行、去掉页眉页脚。页眉页脚是 PDF 解析的重灾区,每页都重复出现,如果不清理,会被切进每个 chunk,严重干扰检索。
4.3 切分与元数据绑定
切分时把元数据一起带上,这是后面做过滤和溯源的基础。
def chunk_with_metadata(pages, splitter): chunks = [] for page in pages: for i, chunk_text in enumerate(splitter.split_text(page["text"])): chunks.append({ "text": chunk_text, "metadata": { "source": page["source"], "page": page["page"], "chunk_index": i, } }) return chunks4.4 双路索引构建
稠密索引和稀疏索引要分别建。稠密索引用 Qdrant 存向量,稀疏索引可以用内存里的 BM25,也可以存到 Elasticsearch。
from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct client = QdrantClient(host="localhost", port=6333) client.recreate_collection( collection_name="knowledge_base", vectors_config=VectorParams(size=1024, distance=Distance.COSINE), ) embeddings = embed_texts([c["text"] for c in chunks]) points = [ PointStruct(id=i, vector=embeddings[i].tolist(), payload=chunks[i]) for i in range(len(chunks)) ] client.upsert(collection_name="knowledge_base", points=points)注意size参数必须和嵌入模型的维度一致,BGE-large-zh 是 1024 维,写错了会直接报错。
4.5 检索流程的完整实现
把稠密检索、稀疏检索、融合、重排串起来,就是完整的检索流程。
def retrieve(query, top_k=5): # 稠密检索 query_vec = embed_texts([query])[0] dense_hits = client.search( collection_name="knowledge_base", query_vector=query_vec.tolist(), limit=20, ) dense_ids = [hit.id for hit in dense_hits] # 稀疏检索 query_tokens = list(jieba.cut(query)) sparse_scores = bm25.get_scores(query_tokens) sparse_ids = np.argsort(sparse_scores)[::-1][:20].tolist() # RRF 融合 fused = rrf_fusion(dense_ids, sparse_ids) candidate_ids = [doc_id for doc_id, _ in fused[:20]] # 重排 candidates = [chunks[i]["text"] for i in candidate_ids] pairs = [(query, doc) for doc in candidates] rerank_scores = reranker.predict(pairs) reranked = sorted( zip(candidate_ids, rerank_scores), key=lambda x: x[1], reverse=True, ) return [chunks[i] for i, _ in reranked[:top_k]]4.6 接入 Agent 的生成环节
检索出来的片段要拼进 Prompt,这里有个技巧:给每个片段编号,并要求模型在回答时标注引用来源。这样既方便溯源,也能抑制模型胡编。
def build_prompt(query, retrieved_chunks): context = "\n\n".join([ f"[{i+1}] {c['text']}\n来源:{c['metadata']['source']} 第{c['metadata']['page']}页" for i, c in enumerate(retrieved_chunks) ]) return f"""基于以下参考资料回答问题,并在答案中标注引用编号。 如果参考资料中没有相关信息,请明确说明"资料中未找到"。 参考资料: {context} 问题:{query} 答案:"""这个 Prompt 模板看起来简单,但"资料中未找到"这句话非常关键。没有它,模型在检索不到相关内容时会强行编造;有了它,模型会诚实地承认不知道。
5. 常见问题与排查技巧实录
5.1 检索命中率低的排查思路
检索命中率低是最常见的问题,排查要按环节来,别一上来就换模型。
| 现象 | 可能原因 | 排查方法 | 解决方向 |
|---|---|---|---|
| 语义相近但召回不到 | 嵌入模型不适配领域 | 人工评测嵌入模型 | 换领域适配模型或微调 |
| 精确词召回不到 | 缺稀疏检索 | 检查是否只有稠密路 | 加 BM25 混合检索 |
| 召回内容不完整 | 切分过细 | 检查 chunk 边界 | 增大 chunk 或加 overlap |
| 召回噪声大 | 切分过粗 | 看 chunk 内容是否混杂 | 减小 chunk 或语义切分 |
| 相关文档排后面 | 缺重排 | 看 Top-10 里有没有 | 加重排模型 |
| 中文匹配差 | 分词问题 | 检查分词结果 | 自定义词典 |
我踩过最深的坑是分词。有一次做产品型号检索,用户搜"XR2000A",jieba 把它切成"XR"、"2000"、"A",结果匹配不到文档里的"XR-2000A"。后来加了自定义词典,把产品型号整体作为一个词,问题才解决。
5.2 嵌入模型的显存与速度优化
嵌入模型跑起来吃显存,尤其是大模型。如果显存不够,有几个办法:一是换小模型,BGE-base 比 BGE-large 小一半,效果差距在多数场景可以接受;二是用半精度推理,model.half()能省一半显存;三是用 ONNX 或 TensorRT 加速,速度能提升 2 到 3 倍。
批量大小也要调。batch size 太小,GPU 利用率低;太大,显存爆掉。我一般从 32 开始试,逐步往上加,直到显存占用到 80% 左右。
5.3 向量库的性能调优
Qdrant 默认用 HNSW 索引,这个索引有几个关键参数:m控制每个节点的连接数,ef_construct控制构建时的搜索范围。m 越大、ef_construct 越大,索引质量越高但构建越慢、内存占用越大。
我的经验值是:m 取 16,ef_construct 取 100,在大多数场景是平衡点。如果数据量特别大(千万级以上),可以适当调大 m 到 32。检索时的ef参数控制搜索范围,调大能提升召回但会变慢,一般取 64 到 128。
5.4 几个容易忽略的实操心得
第一,一定要做检索评测。别凭感觉判断效果,准备一个评测集,每次改动都跑一遍,看召回率、MRR、NDCG 这些指标的变化。没有评测的优化都是瞎猜。
第二,元数据过滤能救命。如果知识库有权限区分,检索时一定要带上权限过滤条件,否则会召回用户无权查看的内容。这个在 Qdrant 里用 filter 实现,别等到出事故才想起来。
第三,缓存高频 query。很多业务场景下,用户问的问题高度重复。把高频 query 的检索结果缓存起来,能大幅降低响应延迟和计算成本。
第四,监控检索质量。上线后要持续监控,记录每次检索的 query、召回结果、用户反馈。发现某类 query 效果差,就针对性优化。
第五,别迷信单一指标。召回率高不代表效果好,如果召回的都是一堆边缘相关的内容,反而会干扰模型。要综合看召回率和准确率,必要时牺牲一点召回换准确。
5.5 从基础 RAG 到 Agentic RAG 的演进方向
基础 RAG 是"一次检索、一次生成",但真实场景往往需要多轮检索。比如用户问"对比 A 产品和 B 产品的技术参数",一次检索可能只召回 A 的信息,需要 Agent 自己判断信息不足,再发起一次针对 B 的检索。这就是Agentic RAG的思路——把检索变成 Agent 的一个工具,由 Agent 决定什么时候检索、检索什么、检索几次。
再往上是GraphRAG和本体 RAG,把知识组织成图结构,利用实体间的关系做多跳推理。这类方案适合知识关联复杂的场景,比如医疗诊断、法律条文推理。但工程复杂度也高得多,建议先把基础 RAG 做扎实,再考虑往上走。
我个人在实际操作中的体会是:RAG 的效果 70% 取决于数据质量和切分策略,20% 取决于检索策略,只有 10% 取决于模型选型。很多团队本末倒置,花大量时间对比模型,却不肯在数据清洗和切分上多花一天。把数据治理做好,用中等模型也能跑出很好的效果;数据一团糟,用最强的模型也救不回来。