1. 为什么知识获取管道是 AI Agent 落地的第一道坎
做 AI Agent 开发的人,绕不开一个尴尬的现实:模型本身很聪明,但它不知道你公司内部的业务规则、不知道你昨天刚更新的产品文档、更不知道你私有的那套数据长什么样。你问它一个关于内部流程的问题,它要么一本正经地胡说八道,要么直接告诉你“我无法回答”。这不是模型不行,而是它缺少一条把外部知识喂给它的管道。
这条管道,就是RAG(Retrieval-Augmented Generation,检索增强生成)。名字听起来学术,拆开看特别朴素:检索 + 增强 + 生成。先从你的知识库里把相关内容捞出来,再把捞出来的内容塞进提示词,最后让模型基于这些内容生成回答。整个过程的核心价值在于——让模型的回答有据可依,而不是凭空捏造。
我见过太多团队在 Agent 项目上踩的第一个大坑,就是一上来就折腾多智能体协作、工具调用编排、复杂的状态机,结果基础的知识获取管道没搭好,Agent 回答问题时满嘴跑火车,用户用两次就再也不打开了。知识获取管道是 Agent 的地基,地基不稳,上面盖什么都是危楼。
这篇内容适合谁看?如果你正在用 TypeScript 做 AI Agent 开发,想从零搭一套能用的 RAG 管道,或者你已经有了一个 RAG 原型但效果不理想想找问题,那接下来的内容应该能帮到你。我会从整体设计思路讲到具体实现细节,包括分块策略、向量化选型、检索逻辑、重排序,以及那些文档里不会写但实际开发中一定会遇到的坑。
2. RAG 管道的整体设计与核心思路拆解
2.1 一条完整的知识获取管道长什么样
很多人对 RAG 的理解停留在“把文档切一切、向量化、存进向量库、查询时做相似度匹配”这个层面。这个理解不算错,但太粗糙了。一条真正能用的知识获取管道,至少包含以下几个环节:
- 文档加载:从各种来源(本地文件、数据库、API)把原始内容读进来
- 内容清洗:去掉无关的格式标记、页眉页脚、重复内容
- 分块:把长文档切成适合检索和生成的小块
- 向量化:把文本块转成向量表示
- 存储:把向量和原文一起存进向量数据库
- 检索:根据用户查询找出最相关的文本块
- 重排序:对检索结果做二次排序,把真正有用的排前面
- 上下文组装:把选中的文本块拼成提示词的一部分
- 生成:模型基于组装好的上下文生成回答
这九个环节里,分块和检索是最容易出问题的两个点。分块切得不好,检索再准也捞不到完整信息;检索策略太单一,明明库里有的内容就是匹配不上。后面我会重点展开这两块。
2.2 为什么选 TypeScript 来做这条管道
热词里出现了 TypeScript,这不是偶然。用 TypeScript 做 RAG 管道有几个实打实的好处:
类型安全在数据处理管道里价值巨大。RAG 管道涉及大量的数据结构转换——文档对象、文本块、向量、检索结果、上下文组装结果,每一层都有不同的字段和类型。用 TypeScript 把这些类型定义清楚,编译期就能发现大部分字段拼写错误和类型不匹配的问题。我试过用纯 JavaScript 写同样的逻辑,调试时间至少多出三分之一。
生态成熟度足够。Node.js 生态里有 LangChain.js、LlamaIndex.TS 这些专门做 RAG 的框架,也有原生的向量数据库客户端。你不需要自己造轮子,但需要理解这些轮子是怎么转的,才能在出问题时知道去哪里找原因。
前后端统一语言。如果你的 Agent 产品有前端界面,用 TypeScript 可以让前后端共享类型定义。检索结果的格式、文档块的接口,前后端用同一套类型,联调时少吵很多架。
注意:TypeScript 5.x 之后一些旧的编译选项(比如
moduleResolution: "node10")已经被标记为弃用,新项目建议直接用"bundler"或"node16"。这个细节在搭建项目时如果忽略,后面升级 TypeScript 7.0 时会遇到编译报错。
2.3 方案选型的几个关键决策点
搭 RAG 管道时,有几个决策会直接影响最终效果和开发成本,我逐个说下我的选择和理由。
向量数据库选型。可选的有 Pinecone、Weaviate、Qdrant、Chroma,还有 Postgres 的 pgvector 扩展。我的建议是:如果你只是做原型验证,用 Chroma 或 pgvector 就够了,部署简单、成本低;如果要上生产且数据量在百万级以上,Qdrant 或 Weaviate 的性能和运维成熟度更好。Pinecone 是托管服务,省心但费用不低,适合不想碰运维的团队。
Embedding 模型选型。OpenAI 的 text-embedding-3-small 性价比很高,1536 维的向量,每百万 token 成本很低。如果对中文支持要求高,可以考虑 BGE 系列或 M3E 这类中文优化的模型。选型时重点看两个指标:检索命中率和向量维度。维度越高,存储和计算成本越大,但检索精度不一定线性提升。
分块策略。固定长度分块最简单,但效果往往最差。我推荐用递归分块,按段落、句子、字符逐级切分,尽量保持语义完整性。块大小控制在 500-1000 个 token 之间,块之间保留 10%-20% 的重叠,避免关键信息被切断。
检索策略。纯向量检索在语义匹配上表现好,但对精确关键词的匹配能力弱。我的做法是混合检索:向量检索 + 关键词检索(BM25),两路结果合并后做重排序。这样既能匹配语义相似的内容,也能精确命中专有名词和代码片段。
3. 核心细节解析与实操要点
3.1 文档分块:切得好比检得准更重要
分块是 RAG 管道里最容易被低估的环节。很多人随便按 500 字切一刀就完事了,结果检索出来的块要么缺头少尾,要么包含大量无关内容。我踩过的坑是:一份产品需求文档,按固定长度切分后,某个功能点的描述被切成了两半,检索时只命中了一半,模型给出的回答缺了关键约束条件,差点导致线上事故。
递归分块的具体做法是这样的:先按双换行(段落)切分,如果某个段落还是太长,再按单换行切分,还长就按句号切分,最后才按字符硬切。这样能最大程度保持语义单元的完整性。
interface ChunkConfig { chunkSize: number; // 目标块大小(token 数) chunkOverlap: number; // 块间重叠 token 数 separators: string[]; // 分隔符优先级列表 } const defaultConfig: ChunkConfig = { chunkSize: 800, chunkOverlap: 120, separators: ["\n\n", "\n", "。", "!", "?", ".", " ", ""], };分块时有个细节要注意:代码块和表格不要切开。如果你的知识库里有技术文档,代码示例被切碎后检索出来毫无意义。我的做法是在分块前先识别代码块和表格,把它们作为不可分割的单元处理,如果超过块大小限制,就单独成块并标记类型。
另一个经验是给每个块加上元数据。至少包含:来源文档名、块在文档中的位置、块的类型(正文/代码/表格)。这些元数据在检索时可以用来过滤,比如用户问的是 API 用法,就优先检索代码类型的块。
3.2 向量化:不只是调个 API 那么简单
向量化看起来简单——把文本传给 Embedding API,拿回一个浮点数数组。但实际做的时候有几个关键决策:
批量处理与速率控制。Embedding API 通常有速率限制,如果你有几千个文档块要处理,不能一次性全发出去。我的做法是用队列控制并发数,同时做好失败重试。重试时要区分错误类型:速率限制错误要退避重试,内容格式错误重试也没用,直接记录并跳过。
async function embedWithRetry( texts: string[], maxRetries: number = 3 ): Promise<number[][]> { for (let attempt = 0; attempt < maxRetries; attempt++) { try { const response = await embeddingClient.embed(texts); return response.embeddings; } catch (error) { if (isRateLimitError(error) && attempt < maxRetries - 1) { await sleep(Math.pow(2, attempt) * 1000); continue; } throw error; } } throw new Error("Embedding failed after max retries"); }向量归一化。如果你用的是余弦相似度做检索,记得把向量归一化到单位长度。有些向量数据库会自动做这件事,有些不会。不归一化的话,相似度计算会受向量模长影响,导致检索结果偏差。我一般会在存入数据库前统一做 L2 归一化。
缓存机制。开发阶段你会反复对同一批文档做向量化,每次都调 API 既慢又费钱。我的做法是用内容哈希做缓存键,把向量化结果存本地。内容没变就直接读缓存,内容变了才重新调 API。这个小优化能让开发迭代速度快好几倍。
3.3 检索策略:单一向量检索不够用
纯向量检索有个致命弱点:它对精确匹配不敏感。比如用户搜“ERR_CONNECTION_REFUSED”,向量检索可能返回一堆关于“网络错误处理”的文档,但就是找不到包含这个具体错误码的那一篇。这时候就需要混合检索来补位。
混合检索的实现思路是并行跑两路检索:
- 向量检索:把查询向量化,在向量库里找余弦相似度最高的 top-K 个块
- 关键词检索:用 BM25 或全文索引,找包含查询关键词的 top-K 个块
然后把两路结果合并,用 RRF(Reciprocal Rank Fusion)算法重新排序。RRF 的公式很简单:每个文档的得分是1 / (k + rank)的累加,k 通常取 60。这个算法不需要调参,对两路检索的分数尺度不敏感,实测下来很稳。
function reciprocalRankFusion( vectorResults: SearchResult[], keywordResults: SearchResult[], k: number = 60 ): SearchResult[] { const scoreMap = new Map<string, number>(); vectorResults.forEach((result, index) => { const score = 1 / (k + index + 1); scoreMap.set(result.id, (scoreMap.get(result.id) || 0) + score); }); keywordResults.forEach((result, index) => { const score = 1 / (k + index + 1); scoreMap.set(result.id, (scoreMap.get(result.id) || 0) + score); }); return Array.from(scoreMap.entries()) .sort((a, b) => b[1] - a[1]) .map(([id]) => allResultsMap.get(id)!); }检索数量怎么定。top-K 不是越大越好。K 太大,无关内容会稀释有效信息,模型反而容易被带偏;K 太小,可能漏掉关键内容。我的经验值是:向量检索取 top-20,关键词检索取 top-20,融合后取 top-5 到 top-8 送给模型。具体数字要根据你的文档密度和查询复杂度来调。
3.4 重排序:把真正有用的排到最前面
检索出来的 top-K 个块,相关性是有差异的。向量相似度高不代表内容真的有用。重排序就是用一个小模型(Cross-Encoder)对查询和每个候选块做精细的相关性打分,然后重新排序。
重排序模型比 Embedding 模型慢,但比大模型快得多。它只处理查询和候选块的组合,计算量可控。我一般用 BGE-Reranker 或 Cohere Rerank,效果提升很明显。实测下来,加了重排序之后,检索命中率能提升 15%-25%。
注意:重排序不是必须的。如果你的检索结果已经足够好,或者对延迟极其敏感,可以跳过这一步。但在知识库内容复杂、查询意图模糊的场景下,重排序带来的收益远大于成本。
4. 实操过程与核心环节实现
4.1 项目初始化与依赖安装
先把项目骨架搭起来。我用的是 Node.js + TypeScript 的组合,包管理用 pnpm(比 npm 快,磁盘占用也小)。
mkdir rag-pipeline && cd rag-pipeline pnpm init pnpm add langchain @langchain/openai @langchain/community pnpm add -D typescript @types/node tsxtsconfig.json的配置要注意几个点:
{ "compilerOptions": { "target": "ES2022", "module": "ESNext", "moduleResolution": "bundler", "strict": true, "esModuleInterop": true, "skipLibCheck": true, "outDir": "./dist", "rootDir": "./src" }, "include": ["src/**/*"] }moduleResolution用"bundler"而不是已弃用的"node10",这是为了兼容现代打包工具,也避免未来升级 TypeScript 7.0 时出问题。strict一定要开,RAG 管道里类型错误很隐蔽,严格模式能帮你提前发现。
4.2 文档加载与清洗
假设我们的知识库是一批 Markdown 文件。加载逻辑要处理编码问题、过滤空文件、提取元数据。
import { readFileSync, readdirSync } from "fs"; import { join, basename } from "path"; interface RawDocument { content: string; metadata: { source: string; fileName: string; loadedAt: string; }; } function loadDocuments(dirPath: string): RawDocument[] { const files = readdirSync(dirPath).filter((f) => f.endsWith(".md")); const documents: RawDocument[] = []; for (const file of files) { const filePath = join(dirPath, file); const content = readFileSync(filePath, "utf-8"); if (content.trim().length === 0) { console.warn(`Skipping empty file: ${file}`); continue; } documents.push({ content: cleanContent(content), metadata: { source: filePath, fileName: basename(file), loadedAt: new Date().toISOString(), }, }); } return documents; } function cleanContent(raw: string): string { return raw .replace(/\r\n/g, "\n") .replace(/\n{3,}/g, "\n\n") .replace(/[ \t]+$/gm, "") .trim(); }清洗这步看着简单,但实际项目中经常被忽略。我遇到过文档里混入 HTML 标签、零宽字符、重复页眉的情况,不清洗的话这些噪音会进入向量库,影响检索质量。
4.3 分块实现与参数调优
分块用 LangChain 的RecursiveCharacterTextSplitter就够用,但参数要根据你的文档特点调。
import { RecursiveCharacterTextSplitter } from "langchain/text_splitter"; interface TextChunk { content: string; metadata: { source: string; chunkIndex: number; totalChunks: number; }; } async function splitDocuments( documents: RawDocument[] ): Promise<TextChunk[]> { const splitter = new RecursiveCharacterTextSplitter({ chunkSize: 800, chunkOverlap: 120, separators: ["\n\n", "\n", "。", "!", "?", ".", " ", ""], lengthFunction: (text) => text.length, }); const allChunks: TextChunk[] = []; for (const doc of documents) { const splits = await splitter.splitText(doc.content); splits.forEach((content, index) => { allChunks.push({ content, metadata: { source: doc.metadata.source, chunkIndex: index, totalChunks: splits.length, }, }); }); } return allChunks; }参数怎么调?我的经验是:中文文档的 chunkSize 可以比英文小一些,因为中文信息密度高,800 个字符已经包含不少内容了。chunkOverlap 设成 chunkSize 的 15% 左右比较合适,太小了上下文断裂,太大了存储和检索成本上升。
4.4 向量化与存储
向量化用 OpenAI 的 embedding 接口,存储用内存向量库做演示(生产环境换成 Qdrant 或 pgvector)。
import { OpenAIEmbeddings } from "@langchain/openai"; import { MemoryVectorStore } from "langchain/vectorstores/memory"; import { Document } from "langchain/document"; async function buildVectorStore( chunks: TextChunk[] ): Promise<MemoryVectorStore> { const embeddings = new OpenAIEmbeddings({ modelName: "text-embedding-3-small", batchSize: 100, }); const documents = chunks.map( (chunk) => new Document({ pageContent: chunk.content, metadata: chunk.metadata, }) ); const store = await MemoryVectorStore.fromDocuments( documents, embeddings ); return store; }batchSize设成 100 是个折中值。太小了请求次数多,太大了容易触发速率限制。实际跑的时候要根据 API 返回的速率限制头动态调整。
4.5 检索与上下文组装
检索部分把向量检索和关键词检索结合起来。关键词检索可以用简单的 BM25 实现,也可以用数据库的全文索引。
interface RetrievalResult { content: string; metadata: Record<string, unknown>; score: number; } async function hybridRetrieve( query: string, vectorStore: MemoryVectorStore, keywordIndex: KeywordIndex, topK: number = 5 ): Promise<RetrievalResult[]> { const [vectorResults, keywordResults] = await Promise.all([ vectorStore.similaritySearchWithScore(query, 20), keywordIndex.search(query, 20), ]); const fused = reciprocalRankFusion( vectorResults.map(([doc, score]) => ({ id: doc.metadata.source + doc.metadata.chunkIndex, content: doc.pageContent, metadata: doc.metadata, score, })), keywordResults ); return fused.slice(0, topK); }上下文组装时要注意总长度控制。模型的上下文窗口有限,检索出来的内容加上系统提示词和用户问题,不能超过窗口限制。我的做法是给检索内容设一个 token 预算,比如 3000 token,按相关性从高到低填充,填满为止。
function assembleContext( results: RetrievalResult[], maxTokens: number = 3000 ): string { const parts: string[] = []; let currentTokens = 0; for (const result of results) { const tokens = estimateTokens(result.content); if (currentTokens + tokens > maxTokens) break; parts.push(`[来源: ${result.metadata.source}]\n${result.content}`); currentTokens += tokens; } return parts.join("\n\n---\n\n"); }4.6 完整查询流程串联
把上面的环节串起来,一个完整的查询流程是这样的:
async function query( question: string, vectorStore: MemoryVectorStore, keywordIndex: KeywordIndex ): Promise<string> { const results = await hybridRetrieve(question, vectorStore, keywordIndex); const context = assembleContext(results); const prompt = `基于以下参考资料回答问题。如果资料中没有相关信息,请明确说明。 参考资料: ${context} 问题:${question} 回答:`; const response = await llm.invoke(prompt); return response.content as string; }这个流程跑通之后,你就有了一个基础可用的 RAG 管道。但别急着上线,先做一轮评估,看看检索命中率和回答质量到底怎么样。
5. 常见问题与排查技巧实录
5.1 检索命中率低怎么排查
检索命中率低是最常见的问题。排查时按这个顺序来:
先看分块是否合理。把检索出来的块打印出来,看看内容是否完整。如果块被切得支离破碎,先调分块参数。我遇到过一个案例,文档里全是短段落,按默认的\n\n切分后每块只有一两句话,检索时匹配到的块信息量太少。后来把多个短段落合并成一块,效果立刻好转。
再看 Embedding 模型是否适合你的语言。用英文模型处理中文文档,效果会打折扣。如果你的知识库以中文为主,换一个中文优化的 Embedding 模型试试。
然后检查查询改写。用户的原始查询往往很短、很模糊。在检索前先用大模型把查询改写成更明确的检索语句,能显著提升命中率。比如用户问“那个报错怎么解决”,改写成“ERR_CONNECTION_REFUSED 错误的解决方法”,检索效果完全不同。
最后考虑加混合检索和重排序。如果前面几步都做了还是不行,混合检索和重排序通常能带来明显提升。
5.2 模型回答与检索内容不一致
有时候检索出来的内容是对的,但模型回答时却忽略了这些内容,或者把不同块的内容混淆了。这个问题通常出在提示词上。
我的做法是在提示词里明确要求模型只基于参考资料回答,并且要求它引用来源。引用来源这个要求很关键,它迫使模型关注检索内容,而不是依赖自己的参数化知识。
另一个技巧是把检索内容用明确的分隔符包起来,比如<context>和</context>,让模型清楚知道哪部分是参考资料,哪部分是问题。
5.3 处理速度慢的优化思路
RAG 管道的延迟主要来自三个地方:Embedding 调用、向量检索、大模型生成。优化时逐个击破。
Embedding 调用可以缓存。相同内容的向量化结果存本地,下次直接读缓存。向量检索的优化取决于你用的数据库,建好索引、控制检索范围都能提速。大模型生成这块,如果用的是流式输出,用户感知的延迟会低很多。
还有一个容易被忽略的点:并行化。向量检索和关键词检索可以并行跑,不用串行等待。上下文组装和提示词构建也可以提前准备。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方向 | 解决思路 |
|---|---|---|---|
| 检索不到相关内容 | 分块不合理或 Embedding 不匹配 | 打印检索结果看内容完整性 | 调整分块参数,换 Embedding 模型 |
| 检索到无关内容 | 向量相似度阈值太低 | 检查 top-K 结果的相关性 | 提高相似度阈值,加重排序 |
| 回答忽略检索内容 | 提示词不够明确 | 检查提示词是否强调基于资料回答 | 强化提示词约束,要求引用来源 |
| 回答内容矛盾 | 检索到冲突信息 | 检查知识库是否有过时内容 | 清理知识库,加时间过滤 |
| 处理速度慢 | 串行调用过多 | 分析各环节耗时 | 并行化、加缓存、流式输出 |
| 长文档检索效果差 | 关键信息被切散 | 检查分块边界 | 调整分块策略,增大重叠 |
5.5 几个我踩过的坑
坑一:忽略文档更新。知识库里的文档更新了,但向量库没同步,检索出来的还是旧内容。我的做法是给每个文档块加内容哈希,定期扫描文档目录,发现哈希变化就重新向量化。这个同步机制一定要做,否则知识库会越来越“脏”。
坑二:元数据丢失。分块时忘了把来源信息带到每个块上,检索出来不知道内容来自哪个文档,排查问题时两眼一抹黑。元数据一定要在分块时就带上,后面每一步都透传。
坑三:过度依赖默认参数。LangChain 的默认分块参数是给英文通用场景设计的,直接拿来处理中文技术文档效果很差。参数一定要根据自己的数据特点调,没有万能配置。
坑四:不做评估就上线。RAG 管道的效果很难靠感觉判断。我建议至少准备一组测试问题,覆盖不同类型(事实查询、推理、多跳),每次调整参数后跑一遍,看命中率和回答质量的变化。没有评估数据,调参就是盲人摸象。
6. 从基础 RAG 到 Agentic RAG 的演进方向
基础 RAG 管道跑通之后,你会发现它有一些固有的局限。比如:它只能做单轮检索,不能根据中间结果决定下一步查什么;它对所有查询用同一套检索策略,不会根据问题类型动态调整;它不能组合多个知识源的信息做推理。
这些局限催生了Agentic RAG的方向。简单说,就是把 RAG 的检索过程交给 Agent 来编排。Agent 可以决定什么时候检索、检索什么、检索几次、要不要换查询词重新检索。它还可以把检索结果和工具调用结合起来,比如先查数据库拿到订单号,再根据订单号检索相关文档。
从基础 RAG 往 Agentic RAG 演进,我建议的路径是:
第一步,把检索做成工具。把向量检索、关键词检索、数据库查询都封装成 Agent 可以调用的工具,让 Agent 自己决定用哪个。
第二步,加查询规划。让 Agent 在检索前先分析问题,决定需要哪些信息、分几步检索。复杂问题拆成多个子查询,分别检索后合并结果。
第三步,加自我反思。Agent 拿到检索结果后,评估一下这些内容是否足够回答问题。不够的话,调整查询词再检索一轮。这个反思循环能显著提升复杂问题的回答质量。
第四步,多知识源融合。当你有多个知识库(产品文档、客服记录、代码仓库)时,Agent 需要知道什么问题该查哪个库,以及如何把多个库的结果融合起来。
这条路走下来,RAG 管道就从一条固定的流水线变成了一个能自主决策的知识获取系统。但前提是基础管道足够扎实——分块、检索、重排序这些环节都调到位了,上层 Agent 的编排才有意义。基础不牢,Agentic RAG 只会让问题更复杂。
我在实际项目中的体会是,先把基础 RAG 的检索命中率做到 80% 以上,再考虑上 Agent 编排。否则你会在 Agent 层花大量时间调试,最后发现根因还是检索没做好。这个顺序不能反。