TypeScript 实战：从零搭建 AI Agent 知识获取管道（RAG）

1. 为什么知识获取管道是 AI Agent 的分水岭

做 AI Agent 开发的人，绕不开一个尴尬的现实：模型本身很聪明，但它对你私有的东西一无所知。你问它公司内部的报销流程，它只能编；你问它上周刚更新的接口文档，它还是只能编。这不是模型不行，而是它的知识被冻结在训练截止那一天。知识获取管道要解决的，就是让 Agent 在运行时能够"现查现用"，把外部知识动态喂进上下文。

这条管道在业内有个更常见的名字——RAG（检索增强生成）。拆开看就三件事：把知识切块存起来（索引）、根据问题把相关的块捞出来（检索）、把捞出来的内容拼进提示词让模型作答（生成）。听起来简单到像废话，但真正落地时，90% 的效果问题都出在前两步，而不是模型。

我见过太多项目，上来就调大模型 API，提示词写得花里胡哨，结果一问具体业务就露馅。后来加上 RAG，效果立竿见影，但又踩进另一个坑：检索回来的内容驴唇不对马嘴，模型拿着错误材料一本正经地胡说。所以这一篇我想把 RAG 的基础管道从头捋一遍，用TypeScript把每个环节都写出来，让你不光知道怎么调库，更知道每一步为什么这么设计。

这篇文章适合谁？如果你正在从 0 到 1 搭建 AI Agent，或者已经搭了个能聊天的壳子但苦于它不懂你的业务，那这篇就是给你准备的。我会假设你有基本的 TypeScript 基础，但不需要你懂向量数据库或嵌入模型——这些我都会从零讲清楚。读完你应该能自己搭一条可用的知识获取管道，并且知道出问题时该往哪儿查。

2. RAG 管道的整体设计与选型思路

2.1 一条完整的知识获取管道长什么样

先把全景图铺开。一条标准的 RAG 管道，从原始文档到最终回答，中间要经过这么几个阶段：

• 加载（Load）：把 PDF、Markdown、网页、数据库记录等各种来源的原始内容读进来，统一成纯文本。
• 切分（Chunk）：把长文本切成大小合适的片段。这一步看似机械，实则决定了检索的粒度。
• 嵌入（Embed）：把每个文本片段转成一个高维向量，语义相近的文本在向量空间里距离也近。
• 存储（Store）：把向量和原文一起存进向量数据库，建立索引。
• 检索（Retrieve）：用户提问时，把问题也转成向量，在库里找最相似的若干个片段。
• 重排（Rerank，可选但强烈建议）：对初步检索结果做二次精排，把真正相关的顶上来。
• 生成（Generate）：把精排后的片段拼进提示词，交给大模型生成最终答案。

这七步里，加载和切分是"离线"做的，一次处理可以反复用；嵌入和存储也是离线；检索、重排、生成是"在线"的，每次提问都要跑一遍。理解这个离线/在线的划分很重要，因为它决定了你的性能优化该往哪儿使劲——离线慢点无所谓，在线必须快。

2.2 为什么用 TypeScript 而不是 Python

RAG 的教程和库，绝大多数是 Python 生态的，LangChain、LlamaIndex 都是 Python 优先。那我为什么还要用 TypeScript 写这一篇？

原因很实际。第一，如果你的 Agent 本身是跑在 Node.js 服务里的，或者前端就要做检索，跨语言调用会带来额外的部署复杂度和延迟。第二，TypeScript 的类型系统在处理文档元数据、检索结果结构时非常舒服，能提前把很多字段拼写错误挡在编译期。第三，现在主流的嵌入模型和向量库都提供了 HTTP API 或 JS SDK，语言早就不是障碍了。

提示：选语言看你的 Agent 主战场在哪。如果整个后端是 Java 或 Python，没必要为了 RAG 单独引入 Node；但如果你的 Agent 服务本来就是 TS 写的，硬塞一个 Python 微服务进来只会让运维头疼。

2.3 方案选型：自建还是用现成服务

这里有个绕不开的决策：向量存储和嵌入，是自己搭还是用云服务？

自建的好处是数据不出内网、成本可控、可深度定制。坏处是要自己维护向量库、处理扩容、调优索引参数。云服务的好处是开箱即用、免运维，坏处是数据要出境（对很多企业是硬伤）、按量计费长期看不一定便宜。

我的建议是分阶段：原型阶段用本地方案快速验证效果，比如内存向量库或者轻量的本地文件索引，跑通了再决定要不要上生产级方案。很多团队一上来就买最贵的向量数据库，结果发现检索效果差是因为切分策略不对，跟数据库一点关系没有。先把管道逻辑跑通，存储层随时可以换。

3. 核心环节拆解与实操要点

3.1 文档加载：别小看格式清洗

加载这一步，新手最容易掉以轻心。以为读个文件就完事了，实际上原始文档里全是噪音：PDF 里的页眉页脚、Markdown 里的代码块标记、网页里的导航栏和广告。这些噪音如果不清掉，会被切进片段里，污染检索结果。

我一般会做这么几件事：统一换行符、去掉连续空行、剥离 HTML 标签、把表格转成结构化的文本描述。特别是表格，直接当纯文本读会丢失行列关系，模型看了也懵。一个实用的做法是把表格转成"列名: 值"的键值对形式，虽然啰嗦但语义清晰。

interface RawDocument { id: string; source: string; content: string; metadata: Record<string, string | number>; } function cleanText(raw: string): string { return raw .replace(/\r\n/g, "\n") .replace(/<[^>]+>/g, "") .replace(/\n{3,}/g, "\n\n") .replace(/[ \t]+/g, " ") .trim(); }

这段清洗逻辑看着朴素，但能解决大部分脏数据问题。注意metadata字段，它会在检索时派上大用场——比如你可以按来源过滤、按时间排序，这是纯向量检索做不到的。

3.2 文本切分：粒度的艺术

切分是整条管道里最需要经验的一步。切太大，一个片段里混了好几个主题，检索时语义被稀释；切太小，一句话被拦腰截断，上下文丢失，模型拿到半截话也答不好。

业界常用的策略是固定长度加重叠：每个片段比如 500 个字符，相邻片段重叠 50 到 100 个字符。重叠的意义在于，如果一句话正好落在切分点上，前后两个片段都能包含它，不至于丢信息。

但更聪明的做法是按语义边界切。优先在段落、句子、标题这些自然边界处断开，实在没有边界再按长度硬切。这样每个片段都是一个完整的语义单元。

interface Chunk { id: string; docId: string; content: string; index: number; } function splitByParagraph( doc: RawDocument, maxLen = 500, overlap = 80 ): Chunk[] { const paragraphs = doc.content.split(/\n\n+/); const chunks: Chunk[] = []; let buffer = ""; let idx = 0; for (const p of paragraphs) { if ((buffer + p).length > maxLen && buffer.length > 0) { chunks.push({ id: `${doc.id}-${idx}`, docId: doc.id, content: buffer.trim(), index: idx++, }); buffer = buffer.slice(-overlap) + "\n\n" + p; } else { buffer += (buffer ? "\n\n" : "") + p; } } if (buffer.trim()) { chunks.push({ id: `${doc.id}-${idx}`, docId: doc.id, content: buffer.trim(), index: idx, }); } return chunks; }

注意：maxLen和overlap没有万能值。中文信息密度高，500 字符往往够用；英文可能要 800 到 1000。overlap 一般取 maxLen 的 10% 到 20%。这些参数必须拿你的真实数据去试，别照抄。

3.3 嵌入与向量存储：把语义变成坐标

嵌入模型的作用，是把一段文本映射成一个固定长度的数字数组，比如 768 维或 1536 维。这个数组就是文本在"语义空间"里的坐标。语义相近的文本，坐标距离也近。检索的本质，就是算距离找最近的邻居。

距离怎么算？常用的是余弦相似度，它衡量两个向量的夹角，对向量长度不敏感，适合文本。也有用欧氏距离的，但文本场景下余弦更稳。

function cosineSimilarity(a: number[], b: number[]): number { let dot = 0, normA = 0, normB = 0; for (let i = 0; i < a.length; i++) { dot += a[i] * b[i]; normA += a[i] * a[i]; normB += b[i] * b[i]; } return dot / (Math.sqrt(normA) * Math.sqrt(normB)); }

存储层，原型阶段我推荐先用内存数组，简单直接，几百上千个片段完全够用。等数据量上来了再换向量数据库。下面是个极简的内存向量库：

class InMemoryVectorStore { private items: { chunk: Chunk; vector: number[] }[] = []; add(chunk: Chunk, vector: number[]) { this.items.push({ chunk, vector }); } search(queryVec: number[], topK = 5) { return this.items .map((it) => ({ chunk: it.chunk, score: cosineSimilarity(queryVec, it.vector), })) .sort((a, b) => b.score - a.score) .slice(0, topK); } }

这个实现没有任何优化，检索是 O(n) 全量扫描。但它的价值在于让你彻底看清检索的本质——就是算相似度排序。理解了这层，你再用专业向量库时就知道那些索引参数在优化什么了。

3.4 检索与重排：把对的捞上来

基础检索就是拿问题向量去库里找 topK。但纯向量检索有个通病：它擅长抓"语义相似"，不擅长抓"关键词精确匹配"。比如你搜一个产品型号 "XR-2000"，向量检索可能给你返回一堆讲"产品型号命名规范"的片段，因为语义上相关，但你要的是那个具体型号。

解决办法是混合检索：向量检索和关键词检索（比如 BM25）各跑一遍，把结果融合。融合算法常用 RRF（倒数排名融合），简单说就是看一个片段在两个榜单里的排名，排名越靠前综合分越高。

function rrfFuse( vectorResults: { id: string; rank: number }[], keywordResults: { id: string; rank: number }[], k = 60 ): { id: string; score: number }[] { const scores = new Map<string, number>(); const add = (id: string, rank: number) => { scores.set(id, (scores.get(id) ?? 0) + 1 / (k + rank)); }; vectorResults.forEach((r) => add(r.id, r.rank)); keywordResults.forEach((r) => add(r.id, r.rank)); return [...scores.entries()] .map(([id, score]) => ({ id, score })) .sort((a, b) => b.score - a.score); }

重排则是另一层保险。初步检索捞回 20 个候选，用一个专门的重排模型（通常是交叉编码器）对每个候选和问题做精细打分，再取前 5 个。重排模型比向量检索慢，但准得多，所以只用在候选集上，性价比很高。

4. 完整管道的实操搭建

4.1 从零跑通一条最小管道

理论讲完，我们把整条管道串起来跑一遍。假设你有一批 Markdown 文档，要做一个能回答文档内容问题的 Agent。

第一步，加载并切分所有文档：

import { readFileSync, readdirSync } from "fs"; import { join } from "path"; function loadDocs(dir: string): RawDocument[] { return readdirSync(dir) .filter((f) => f.endsWith(".md")) .map((f) => ({ id: f.replace(".md", ""), source: join(dir, f), content: cleanText(readFileSync(join(dir, f), "utf-8")), metadata: { filename: f }, })); } const docs = loadDocs("./knowledge"); const allChunks = docs.flatMap((d) => splitByParagraph(d));

第二步，调用嵌入接口把每个片段转成向量。这里我用一个通用的 HTTP 调用示意，具体端点换成你用的服务即可：

async function embed(texts: string[]): Promise<number[][]> { const res = await fetch("https://your-embedding-endpoint/v1/embeddings", { method: "POST", headers: { "Content-Type": "application/json", Authorization: `Bearer ${process.env.EMBED_KEY}`, }, body: JSON.stringify({ model: "embed-model", input: texts }), }); const data = await res.json(); return data.data.map((d: any) => d.embedding); } const store = new InMemoryVectorStore(); const batchSize = 32; for (let i = 0; i < allChunks.length; i += batchSize) { const batch = allChunks.slice(i, i + batchSize); const vectors = await embed(batch.map((c) => c.content)); batch.forEach((c, j) => store.add(c, vectors[j])); }

批处理这一步很关键。嵌入接口通常按 token 计费，逐条调用既慢又贵，批量提交能显著降低成本。32 是个比较稳妥的批大小，太大可能触发接口限制。

第三步，检索加生成：

async function answer(question: string): Promise<string> { const [qVec] = await embed([question]); const hits = store.search(qVec, 5); const context = hits .map((h, i) => `[片段${i + 1}] ${h.chunk.content}`) .join("\n\n"); const prompt = `你是一个严谨的助手。请仅根据以下资料回答问题， 如果资料中没有答案，直接说"资料中未提及"，不要编造。 资料： ${context} 问题：${question}`; const res = await fetch("https://your-llm-endpoint/v1/chat/completions", { method: "POST", headers: { "Content-Type": "application/json", Authorization: `Bearer ${process.env.LLM_KEY}`, }, body: JSON.stringify({ model: "your-model", messages: [{ role: "user", content: prompt }], temperature: 0.2, }), }); const data = await res.json(); return data.choices[0].message.content; }

到这里，一条最小可用的 RAG 管道就跑通了。注意提示词里那句"资料中没有答案就直接说未提及"，这是抑制幻觉的关键。模型天生倾向于给出一个像样的答案，你不明确禁止它编，它就会编。

4.2 参数选择背后的计算逻辑

很多人调 RAG 是凭感觉，其实几个核心参数是有计算依据的。

topK 取多少？取决于你的上下文窗口和片段大小。假设模型上下文窗口是 8K token，你的提示词模板和问题占了 1K，那留给资料的是 7K。如果每个片段平均 200 token，理论上能塞 35 个。但塞太多会稀释重点，还会引入噪音。实践中 topK 取 3 到 8 比较常见，配合重排效果最好。

片段大小怎么定？一个经验公式：片段大小约等于"一个完整论点所需的平均字数"。技术文档通常一个概念讲清楚要 300 到 600 字，那就定在这个区间。太小则语义不全，太大则检索精度下降。

overlap 取多少？一般取片段大小的 10% 到 20%。太小起不到防截断作用，太大则存储冗余、检索重复。500 的片段配 80 的重叠是个不错的起点。

4.3 元数据过滤：被低估的利器

纯向量检索有个盲区：它不知道时间、来源、权限这些"硬约束"。比如用户问"最新的政策是什么"，向量检索可能把三年前的旧政策也捞回来，因为语义上"政策"都相关。

解决办法是在检索时加元数据过滤。给每个片段打上时间戳、来源、分类标签，检索时先按条件筛，再算相似度。

function searchWithFilter( store: InMemoryVectorStore, queryVec: number[], filter: (c: Chunk) => boolean, topK = 5 ) { return store["items"] .filter((it) => filter(it.chunk)) .map((it) => ({ chunk: it.chunk, score: cosineSimilarity(queryVec, it.vector), })) .sort((a, b) => b.score - a.score) .slice(0, topK); }

这个能力在真实业务里价值巨大。多租户系统里，用户只能检索自己有权访问的文档；时效性内容里，只检索最近更新的。这些都不是向量能表达的，必须靠元数据。

5. 常见问题与排查技巧实录

5.1 检索效果差的排查顺序

检索不准是最常见的问题，但原因可能出在管道的好几个环节。我一般按这个顺序排查：

排查时有个笨办法但特别有效：把检索回来的片段原样打印出来看。很多人只看最终答案，不看中间检索结果，结果在错误的方向上优化提示词，白费功夫。先确认检索对了，再谈生成。

5.2 幻觉抑制的实战技巧

RAG 能减少幻觉，但不能消除。模型看到检索内容后，仍可能"脑补"出资料里没有的细节。几个我实测有效的技巧：

第一，提示词里明确要求"引用来源"。让模型在答案里标注每个结论来自哪个片段，这会逼它更严格地依赖资料。第二，要求"资料不足时明确说明"，而不是"尽量回答"。第三，把 temperature 调低，0.1 到 0.3 之间，减少随机发挥。第四，如果业务允许，让模型先复述检索到的关键信息，再作答，相当于强制它先"读题"。

提示：别指望一次调好。幻觉抑制是个反复迭代的过程，每次发现编造，就把那个案例加进测试集，改完提示词再跑一遍，看有没有回归。

5.3 性能与成本的平衡

RAG 的成本主要在两块：嵌入调用和生成调用。嵌入是离线一次性的，文档不变就不用重跑，成本可控。生成是在线的，每次提问都要调，是主要开销。

优化生成成本，最直接的是控制塞进上下文的片段数量。topK 从 10 降到 5，token 消耗可能减半，效果未必下降——因为重排后前 5 个往往就包含了答案。另一个技巧是缓存：相同或相似的问题，直接返回缓存答案，省掉一次生成调用。

性能上，检索本身很快（内存库毫秒级），瓶颈通常在嵌入接口的网络延迟。如果在线检索也要实时嵌入问题，那这个延迟就躲不掉。可以考虑用更小的嵌入模型，或者把常见问题预先嵌入缓存起来。

5.4 我踩过的几个坑

说几个文档里不会写、但实际会遇到的坑。

第一个坑是编码问题。中文文档如果编码没处理好，读进来全是乱码，嵌入出来的向量毫无意义，检索自然全错。加载时务必确认文件编码，UTF-8 是标配。

第二个坑是切分把代码块切断了。技术文档里代码块很重要，如果按固定长度硬切，一个函数可能被切成两半，检索到半截代码，模型也还原不出完整逻辑。解决办法是切分时识别代码块边界，把整个代码块作为一个不可分割单元。

第三个坑是嵌入模型的维度不一致。索引时用了一个模型，检索时换了另一个，维度对不上直接报错，或者维度碰巧一样但语义空间不同，检索结果全乱。索引和检索必须用同一个嵌入模型，这个要写进配置里锁死。

第四个坑是topK 太大反而更差。我一度以为多捞点总没错，结果发现塞进去的噪音片段会干扰模型判断，答案质量不升反降。后来加了重排，topK 从 20 精排到 5，效果明显变好。

6. 从基础 RAG 到 Agentic RAG 的演进方向

基础 RAG 跑通之后，你会发现它还是有点"死"：不管问题难易，都是固定检索一次、固定 topK、固定流程。但真实问题千差万别，有的需要查多个文档综合，有的一个问题要拆成几步查，有的查一次发现不够还得再查。

这就是Agentic RAG的用武之地——让 Agent 自己决定要不要检索、检索几次、用什么查询词、结果够不够。它把 RAG 从一条固定管道，变成了 Agent 可以调用的一个工具。Agent 可以先分析问题，生成更精准的查询，检索后评估结果质量，不够就换个角度再查。

实现上，这意味着检索不再是管道里的一个固定步骤，而是一个可以被 Agent 反复调用的函数。Agent 的循环里，它自己判断"我现在需要查资料吗""查回来的够用吗""要不要再查一次"。这比固定管道灵活得多，但也更复杂，需要设计好 Agent 的决策逻辑和终止条件。

从工程角度看，基础 RAG 是 Agentic RAG 的地基。你得先把加载、切分、嵌入、检索、重排这些环节都搞扎实，Agent 调用它们时才靠得住。地基不牢，Agent 再聪明也是拿着烂材料做菜。所以别急着上 Agentic，先把这一篇里的基础管道跑通、调好，再往上叠。

我个人在实际项目里的体会是，RAG 的效果提升，八成来自数据质量和切分策略，两成来自检索算法，提示词工程能贡献的其实很有限。很多人本末倒置，在提示词上反复雕花，却不肯花时间清洗数据、调切分参数。把功夫下在前面，后面会省心很多。