AI Agent知识获取管道实战:TypeScript构建RAG检索增强生成系统
发布时间:2026/9/29 19:03:52 锦皓数字建站

1. 为什么知识获取管道是 AI Agent 落地的第一道坎做 AI Agent 开发的人绕不开一个尴尬的现实模型本身很聪明但它不知道你公司内部的业务规则、不知道你昨天刚更新的产品文档、更不知道你私有的那套数据表结构。你问它一个关于内部流程的问题它要么一本正经地胡说八道要么礼貌地告诉你我无法访问该信息。这不是模型能力问题而是知识边界问题。知识获取管道要解决的核心矛盾就一句话如何让 Agent 在推理的那一刻拿到正确、最新、可追溯的外部知识。RAGRetrieval-Augmented Generation检索增强生成是目前工程上最成熟、性价比最高的解法。它的思路并不复杂——把外部知识切块、向量化、存进索引用户提问时先检索出最相关的片段再把这些片段塞进模型的上下文里让模型看着材料答题。我在实际项目里踩过的最大误区是一开始把 RAG 当成一个库来用以为接上向量数据库就完事了。结果上线后 hit rate命中率惨不忍睹用户问退款流程要几天检索出来的却是退款政策适用范围。问题出在整条管道上切块策略、嵌入模型、检索方式、重排逻辑任何一环掉链子最终答案都会崩。所以这一篇我不打算只讲概念而是把知识获取管道拆成可落地的工程环节用 TypeScript 把关键代码写出来让你能真正跑通一条从文档到答案的链路。这篇文章适合三类人正在从 0 到 1 搭建 AI Agent 的开发者、被 RAG 检索效果折磨过的工程师、以及想搞清楚检索增强生成到底怎么落地的人。读完你应该能自己搭一条最小可用的知识管道并且知道每个环节该调什么参数、避开哪些坑。2. 把 RAG 拆开看一条管道到底有哪几个工位2.1 离线索引阶段文档进来之前先想清楚三件事很多人一上来就写代码调 embedding 接口这是典型的顺序错误。离线索引阶段真正决定成败的是三个前置决策切块粒度、元数据设计、更新策略。切块粒度直接决定检索质量。切太大一个块里混了好几个主题向量被平均掉检索时谁都匹配不准切太小语义不完整模型拿到半句话也答不好。我的经验值是中文文档按 300 到 500 字切英文按 200 到 400 token 切并且优先按语义边界切段落、标题、列表项而不是机械地按字数硬切。如果文档有清晰的标题层级就按标题切这样每个块天然带主题。元数据设计是新手最容易忽略的。每个块除了文本和向量至少要存来源文档 ID、章节路径、更新时间、块序号。为什么因为后面做引用溯源、按时间过滤、增量更新全靠这些字段。我见过有人只存了文本和向量结果要更新一篇文档时只能全量重建索引几万条数据重跑一遍 embedding成本和时间都受不了。更新策略要提前定。是全量重建还是增量更新增量更新需要给每个块算一个稳定的哈希比如文档 ID 加块内容哈希新版本进来时对比哈希只对变化的块重新 embedding。这套逻辑在文档频繁更新的场景下能省掉 80% 以上的计算量。2.2 在线检索阶段从 query 到候选块的完整链路在线阶段是一条流水线query 改写 → 向量化 → 召回 → 重排 → 组装上下文。每一步都有讲究。query 改写解决的是用户口语和文档书面语之间的鸿沟。用户问东西坏了咋退文档里写的是商品质量问题退货流程。直接拿原句去检索向量相似度可能很低。常见做法是用一个小模型或规则把 query 扩展成多个变体或者抽取关键词做混合检索。召回阶段现在主流是混合检索向量检索负责语义匹配关键词检索BM25 之类负责精确匹配。为什么两个都要因为向量检索对专有名词、型号、编号这类字面精确的内容不敏感。用户问XR-200 的保修期向量检索可能召回一堆讲保修政策的块但就是漏掉那个写着XR-200的块。关键词检索恰好补上这个短板。重排Rerank是提升 hit rate 的杀手锏。召回阶段为了不漏通常会取 top 20 甚至 top 50但真正塞进上下文的可能只有 3 到 5 个块。重排模型cross-encoder 架构会把 query 和每个候选块拼在一起打分精度远高于向量点积但速度慢所以只用在候选集上。实测下来加一层重排hit rate 通常能提升 15 到 30 个百分点。2.3 生成阶段上下文怎么塞才不让模型跑偏检索回来的块不是越多越好。上下文塞太多一是浪费 token二是引入噪声模型容易被无关内容带偏。我的做法是重排后取 top 3 到 5按相关性排序相关性最高的放最前面模型对上下文开头和结尾的内容注意力更高。还要给模型明确的指令边界。比如在 system prompt 里写清楚只根据提供的资料回答资料里没有的信息就说不知道不要编造。 这句话能挡掉相当一部分幻觉。另外把每个块的来源标注出来比如[来源1]让模型在回答时引用既方便溯源也逼着模型看着材料说话。2.4 一个容易被忽视的环节评估与反馈闭环没有评估的 RAG 就是盲人摸象。你改了切块策略效果是变好还是变坏加了重排hit rate 涨了多少这些都得靠评估数据说话。最小可用的评估方案是准备 50 到 100 条问题 标准答案 应该命中的文档块的测试集每次改动后跑一遍看两个指标——召回率该命中的块有没有被召回和答案准确率最终答案对不对。召回率低说明检索环节有问题召回率高但答案错说明生成环节有问题。这个区分能帮你快速定位故障点。3. 用 TypeScript 搭一条最小可用的知识管道3.1 环境与依赖选型为什么是这几个包我选 TypeScript 是因为它在 Agent 开发里越来越主流类型系统能在管道这种多环节串联的场景里帮你挡掉大量低级错误。核心依赖就几个xenova/transformers本地跑 embedding 模型不依赖外部 API适合做 demo 和隐私敏感场景。faiss-node或纯内存的余弦相似度实现向量索引。小规模数据几万块以内用内存数组加余弦相似度完全够用别一上来就上重型向量库。gpt-tokenizer算 token 数控制切块大小和上下文预算。提示如果你用外部 embedding API注意把 API key 放环境变量别硬编码进代码。另外 embedding 模型一旦选定索引和查询必须用同一个模型换了模型要全量重建索引这是硬约束。3.2 文档切块按语义边界切别按字数硬切先看切块的核心逻辑。我写了一个按段落聚合的切块函数思路是先按空行和标题把文档拆成自然段再把相邻的小段合并到目标长度遇到标题就强制断开。interface Chunk { id: string; docId: string; text: string; sectionPath: string; hash: string; updatedAt: number; } function splitIntoChunks( docId: string, content: string, targetSize 400, maxSize 600 ): Chunk[] { // 按标题和空行切出自然段 const rawBlocks content .split(/\n\s*\n/) .map((b) b.trim()) .filter(Boolean); const chunks: Chunk[] []; let buffer ; let sectionPath ; let index 0; const flush () { if (!buffer.trim()) return; const text buffer.trim(); chunks.push({ id: ${docId}#${index}, docId, text, sectionPath, hash: simpleHash(text), updatedAt: Date.now(), }); buffer ; }; for (const block of rawBlocks) { const isHeading /^#{1,6}\s/.test(block); if (isHeading) { flush(); sectionPath block.replace(/^#{1,6}\s/, ).trim(); continue; } // 超过最大长度就强制切 if ((buffer block).length maxSize) { flush(); } buffer (buffer ? \n\n : ) block; if (buffer.length targetSize) { flush(); } } flush(); return chunks; } function simpleHash(s: string): string { let h 0; for (let i 0; i s.length; i) { h (h * 31 s.charCodeAt(i)) | 0; } return h.toString(16); }这段代码有几个设计点值得说。targetSize是软目标到了就切maxSize是硬上限超了必须切防止某个超长段落撑爆上下文。sectionPath记录当前块属于哪个章节检索时可以作为过滤条件或展示给用户。hash用于增量更新时判断块有没有变。3.3 向量化与索引本地模型怎么选、怎么存向量化我用xenova/transformers加载一个多语言小模型。选它的理由是模型小几十 MB、本地跑、支持中英文适合做原型。生产环境如果追求效果可以换成更大的模型或外部 API但接口逻辑是一样的。import { pipeline } from xenova/transformers; let embedder: any null; async function getEmbedder() { if (!embedder) { embedder await pipeline( feature-extraction, Xenova/paraphrase-multilingual-MiniLM-L12-v2 ); } return embedder; } async function embed(text: string): Promisenumber[] { const model await getEmbedder(); const output await model(text, { pooling: mean, normalize: true }); return Array.from(output.data as Float32Array); } interface IndexedChunk extends Chunk { vector: number[]; } class VectorIndex { private items: IndexedChunk[] []; async add(chunks: Chunk[]) { for (const c of chunks) { const vector await embed(c.text); this.items.push({ ...c, vector }); } } search(queryVec: number[], topK 10) { const scored this.items.map((item) ({ item, score: cosine(queryVec, item.vector), })); scored.sort((a, b) b.score - a.score); return scored.slice(0, topK); } // 增量更新按 hash 判断是否需要重新 embedding async upsert(chunks: Chunk[]) { const existing new Map(this.items.map((i) [i.id, i])); for (const c of chunks) { const old existing.get(c.id); if (old old.hash c.hash) continue; // 没变跳过 const vector await embed(c.text); const idx this.items.findIndex((i) i.id c.id); if (idx 0) this.items[idx] { ...c, vector }; else this.items.push({ ...c, vector }); } } } function cosine(a: number[], b: number[]): number { let dot 0, na 0, nb 0; for (let i 0; i a.length; i) { dot a[i] * b[i]; na a[i] * a[i]; nb b[i] * b[i]; } return dot / (Math.sqrt(na) * Math.sqrt(nb) 1e-8); }upsert里的 hash 对比就是增量更新的关键。文档更新时重新切块逐块对比 hash只有内容变了的块才重新算向量。这个优化在文档量大、更新频繁的场景下是刚需。3.4 混合检索与重排把 hit rate 拉上去纯向量检索不够用我加了一个简单的关键词打分做混合。关键词打分用词频加逆文档频率的简化版够用就行。function keywordScore(query: string, text: string): number { const qTokens tokenize(query); const tTokens tokenize(text); const tSet new Set(tTokens); let hit 0; for (const t of qTokens) { if (tSet.has(t)) hit; } return qTokens.length ? hit / qTokens.length : 0; } function tokenize(s: string): string[] { // 中文按字切英文按词切简化处理 return s .toLowerCase() .split(/[\s,。.、;:!?]/) .flatMap((seg) (/[\u4e00-\u9fa5]/.test(seg) ? seg.split() : [seg])) .filter(Boolean); } async function hybridSearch( index: VectorIndex, query: string, topK 10 ) { const qVec await embed(query); const vecResults index.search(qVec, topK * 2); const merged vecResults.map((r) ({ item: r.item, vecScore: r.score, kwScore: keywordScore(query, r.item.text), })); // 加权融合权重按实际效果调 merged.forEach((m) { (m as any).finalScore 0.7 * m.vecScore 0.3 * m.kwScore; }); merged.sort((a, b) (b as any).finalScore - (a as any).finalScore); return merged.slice(0, topK); }权重 0.7 和 0.3 不是拍脑袋定的是我在测试集上试出来的。你的数据分布不同这个比例要自己调。调的方法就是固定测试集改权重跑评估看召回率变化。重排环节如果不想引入额外模型可以用一个折中方案把 top 候选块和 query 拼起来用一个小的 cross-encoder 打分。如果资源有限至少要做去重和多样性控制——同一篇文档的相邻块往往高度相似全塞进去等于浪费上下文。我的做法是同一文档最多取 2 个块保证来源多样性。4. 检索效果差先别怪模型按这个顺序排查4.1 命中率低的四种典型症状与对应根因检索效果差是最常见的抱怨但差有很多种。我把它分成四类症状每类对应不同的根因排查顺序不能乱。症状可能根因优先排查项该命中的块完全没进候选切块把关键信息切碎了切块粒度、语义边界候选里有正确块但排名靠后向量质量差或权重不合理embedding 模型、混合权重排名靠前但内容答非所问块内主题混杂切块策略、元数据过滤检索对了但答案还是错生成环节上下文组织问题prompt、上下文顺序排查要从上往下。先确认正确块到底有没有被召回这一步用测试集跑召回率就能看出来。如果召回率就低后面重排、生成再怎么调都是白费。4.2 一个真实的排查链路从答非所问到定位切块问题我遇到过一个典型案例。用户问企业版怎么升级系统答的是个人版功能介绍。第一反应是检索错了但打印出召回的块一看正确块其实在候选里排第 4只是重排后掉出了 top 3。继续往下查发现正确块和错误块在向量空间里非常接近因为两个块都提到了版本升级功能这些词。根因是切块时把企业版升级流程和个人版功能对比切进了相邻的块语义被混在一起了。修复方案是调整切块逻辑遇到企业版个人版这种并列小标题时强制断开保证每个块只讲一个版本。改完之后正确块的向量纯度提高排名自然上去了。这个案例说明很多检索问题本质是切块问题向量模型背了锅。4.3 重排不是万能药什么时候该加、什么时候别加重排能提升精度但不是所有场景都值得加。判断标准是召回阶段的候选集里正确块的排名是否稳定在 top 20 以内。如果在加重排能把它拉到 top 3如果正确块压根不在 top 20重排也救不回来得回去改召回。另外重排有延迟成本。cross-encoder 要对每个候选块单独推理20 个候选就是 20 次前向计算。如果对响应时间敏感要么减少候选数要么用更小的重排模型。我的经验是候选 20 个、重排模型参数量在亿级以下时延迟增加通常在几百毫秒可以接受。5. 让管道真正好用的几个工程细节5.1 上下文预算token 怎么分配才不浪费上下文窗口是有限资源得精打细算。我的分配策略是system prompt 占 10%检索到的资料占 60%对话历史占 20%留给模型输出的空间占 10%。这个比例不是死的但要有意识地去控制。具体到资料部分假设窗口是 8k token资料预算约 4.8k。每个块平均 400 token那最多塞 12 个块。但前面说了塞太多会引入噪声所以我实际只塞 3 到 5 个块剩下的预算留给对话历史和输出。宁可少塞、塞精也不要贪多。注意不同模型的 tokenizer 不一样中文的 token 数往往比字数多。用gpt-tokenizer这类工具精确计算别用字数估算否则容易超窗被截断。5.2 引用溯源让每个答案都能找到出处引用溯源不只是为了好看它是 RAG 相对纯生成的核心优势。实现方式是在组装上下文时给每个块编号prompt 里要求模型引用编号。function buildContext(chunks: IndexedChunk[]): string { return chunks .map( (c, i) [资料${i 1}]来源${c.docId} / ${c.sectionPath}\n${c.text} ) .join(\n\n); } const systemPrompt 你是一个严谨的助手。 只根据下面提供的资料回答问题。 回答时用 [资料N] 标注信息来源。 资料中没有的信息直接说资料中未提及不要编造。;这套组合拳下来用户能看到答案的依据也方便你事后审计。如果发现模型引用了不存在的编号说明它在幻觉这时候要检查是不是上下文里混入了干扰内容。5.3 增量更新与版本管理文档变了怎么办文档更新是常态全量重建索引不可持续。前面upsert已经实现了按 hash 增量更新但还有两个细节要处理。一是删除。文档被删了对应的块也要从索引里移除。做法是维护一个 docId 到块 ID 列表的映射删除文档时按映射批量删块。二是版本回滚。如果新版本索引效果变差要能快速回退。我的做法是索引带版本号每次重建生成新版本评估通过后再切换线上指向。这样出问题能秒回滚不至于手忙脚乱。5.4 评估集怎么建50 条数据起步就够评估集不用一开始就搞得很庞大。50 条高质量的问题-答案-命中块三元组就能给你足够的信号。关键是每条数据要标注应该命中哪个块这样召回率才能算。建评估集的方法从真实用户问题里采样或者自己根据文档内容设计问题。设计问题时要有意识地覆盖不同类型——事实型X 是什么、流程型怎么做 Y、对比型A 和 B 的区别。不同类型对检索的要求不一样混在一起测才能全面反映管道质量。每次改动管道换模型、调切块、改权重都跑一遍评估集记录召回率和答案准确率。时间长了你会有一份自己的调参日志知道什么改动有效、什么改动是负优化。这份日志比任何教程都值钱因为它是针对你的数据分布的。6. 从基础 RAG 往 Agentic RAG 走的几个方向基础 RAG 跑通之后你会发现它有几个天花板单轮检索、固定流程、不会主动追问。Agentic RAG 的思路是让 Agent 自己决定要不要检索、检索几次、检索什么把检索变成 Agent 的一个工具而不是固定环节。第一个方向是查询规划。复杂问题拆成多个子问题分别检索再综合。比如对比 A 和 B 的退款政策可以拆成两个子查询分别检索最后合并。这比一次性检索整个问题效果好得多。第二个方向是迭代检索。第一轮检索后Agent 判断信息够不够不够就基于已有信息生成新的查询再检索一轮。这个循环能处理需要多跳推理的问题。第三个方向是工具化检索。把检索封装成一个函数让 Agent 通过 function calling 自主调用。Agent 可以决定用向量检索还是关键词检索可以决定检索哪个知识库。灵活性上去了但也要注意控制循环次数防止 Agent 陷入无限检索。这几个方向我在后续的项目里都试过效果确实比基础 RAG 好但复杂度也上去了。我的建议是先把基础 RAG 的每个环节调扎实再往上叠 Agentic 的能力。基础不牢加再多花活也是空中楼阁。检索召回率上不去Agent 再聪明也拿不到正确的材料最后还是答错。我个人在实际操作中的体会是RAG 这条管道没有一劳永逸的配置它更像是一个需要持续调优的系统。数据在变、用户在变、模型也在变唯一不变的是用评估数据说话这个原则。每次改动前先想清楚要验证什么假设改完用数据验证别凭感觉调参。这套方法论比任何具体的参数值都重要。
锦
锦皓数字建站
深耕本土企业品牌数字化升级,专注原创端正雅致商务官网,从视觉设计到稳定运维全程保驾护航。