从零手写RAG系统：AI工程的核心链路与实战避坑指南

我一直觉得AI工程这个岗位在过去两年里被严重低估了。外面铺天盖地的教程都在教你怎么调Prompt、怎么套一个LangChain的Demo但真到了线上你面对的是数据切分不合理导致检索结果乱七八糟、用户一句话拆成五个意图、模型超时把整个接口拖垮……这些问题没有一个是你熟练背诵Prompt技巧能解决的。我入行的时候把AI工程想象成写代码让大模型回答问题后来才发现它是一门关于检索、评估、成本、稳定性还有概率的系统工程。这篇文章我想以从零开始做AI工程为线索把我踩过的坑、验证过的方案和现在还在用的方法完整复盘一遍给那些准备独立搭建一个AI应用、尤其是要做RAG相关项目的人一份可以照着走的路线图。1. 先想明白AI工程到底在工程什么1.1 它和传统软件工程的本质差异做传统后端逻辑是确定性的参数传进去结果就是那几个分支你写一百个测试用例基本能覆盖大部分行为。AI工程不一样同样的Prompt、同样的输入模型输出可能每次都不一样而且它出错的方式你事先猜不到。这不是多写几个异常处理能解决的你得接受一个现实你的系统天然带概率性工程的目标不是消灭错误而是降低错误率并且让错误可观测、可回退、可兜底。我在第一个真实的AI项目里吃过很大的亏。当时团队想做一个内部文档问答系统我直接拿现成的框架把文档一股脑塞进向量库然后在Prompt里写请根据以下资料回答。离线随便测了几条觉得效果还行一上线就出问题问报销流程返回的是无关的制度文件问稍微长一点的问题上下文直接被截断最离谱的是有一次模型一本正经地根据不相关的检索结果编了一个答案。那段时间我每天都在翻框架源码但问题根本不在框架上而是我完全不清楚整个链路里每一个环节在做什么、在哪里可能出错。1.2 经典AI应用的基本链路一个典型的LLM应用尤其是RAG检索增强生成类应用链路大概是这样的输入处理用户问题进来先做改写、意图识别、关键词提取有时候还要判断是否涉及敏感内容。检索层根据问题去向量库或关键词索引里召回候选内容这是决定答案质量的上游。上下文组装把召回的文档片段按一定顺序和格式拼进Prompt同时控制长度预算。生成层调用大模型用组装好的上下文生成最终答案。后处理与兜底对输出做校验检查是否偏离检索内容是否满足格式要求如果模型超时或返回了不安全内容要有降级方案。你会发现调Prompt只是其中很小的一块。真正决定系统表现好坏的是检索层的质量和评估反馈的闭环。所以我才坚持建议第一个AI项目一定要自己从最底层手写一遍把切分、向量化、检索、拼装、评估这些环节全部亲自摸一遍哪怕代码丑一点因为只有你亲手踩过切分大小不对导致检索失灵的坑你才能真正理解那些成熟框架里每个参数存在的意义。2. 从零手写一个最小可用的RAG服务我的完整闭环我做一个知识库问答项目时没有直接用现成的检索编排框架而是自己搭了一套最小闭环。下面按环节拆开讲每个环节我都会给出我实际用到的代码和理由代码不追求优雅追求能跑、能改、能理解。2.1 数据切分检索质量的第一道关口很多人把切分想得太简单以为就是按字数切。实际上切分粒度直接决定检索质量切得太细单个片段缺乏上下文模型拿到的是只言片语切得太粗一个片段里混了好几个主题向量化之后会被稀释检索命中率反而下降。我用的经验值是按语义边界切同时保证片段在300到500个token之间并且相邻片段保留一定重叠。我用一套很朴素的分段逻辑先按文档结构标题、段落粗切再对过长的段落做滑动窗口切分窗口大小设为400个token重叠80个token。重叠这里很关键否则一个论点恰好被切成两半信息就丢了。import re def split_document(text, max_tokens400, overlap_tokens80): # 先用空行和标题做语义粗切 blocks re.split(r\n\s*\n|\n#{1,6}\s, text) chunks [] buffer for block in blocks: buffer block \n if count_tokens(buffer) max_tokens: # 按句子边界做细切 sentences re.split(r(?[。!?])\s*, buffer.strip()) while sentences: window while sentences and count_tokens(window) max_tokens: sentence sentences.pop(0) if count_tokens(window) count_tokens(sentence) max_tokens and window: break window sentence chunks.append(window) # 保留最后一段作为overlap注意不要重复 if sentences: sentences.insert(0, window[-min(len(window), 200):]) window buffer if buffer.strip(): chunks.append(buffer.strip()) return chunks这段代码不复杂但它体现了几个我后来才悟到的点一是必须按语义边界而不是纯字符数切二是overlap不是把上一段原样塞进下一段开头那样会重复造出语义偏移的向量更好的做法是句子级窗口滑动让切分点始终保持在尽量完整的句子上。每个chunk我还额外记录了来源文件的ID和标题这样后续检索结果可以定位出处。2.2 向量化与检索不要迷信单一方案向量模型的选择是另一个大坑。我当时对比了几种方案远程API向量、本地开源模型各有各的适用场景。如果你追求检索效果的稳定性并且数据量不大本地开源模型性价比更高也不用把文档内容传到外部服务如果涉及跨语言或超大规模相似检索API模型可能更方便。下面是当时对比的一张简化表方案维度部署成本更新成本适用场景本地开源Embedding模型768/1024左右低单卡可跑文件更新时增量重算私有知识库、数据不出域远程API向量模型视服务而定无部署按量收费无需维护模型但受接口限制快速验证、多语言场景关键词索引BM25无向量极低极低术语精确匹配、人名/编号检索关键结论是不要只做向量检索。我做过一次实验在一个企业内部制度文档集里向量检索对请假流程是什么这种语义问题表现很好但对合同编号CH-2023-0817这种精确字段向量搜索经常召回一堆语义相近但编号不对的内容。后来我在检索层加了BM25关键词召回和向量结果做合并去重整体命中率提升非常明显。很多成熟的检索系统都会做混合检索如果你从零手写最简单的方式是向量检索取topK个结果BM25取topK个结果两个集合合并后统一用相关性打分排序。import numpy as np def cosine_similarity(a, b): return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) 1e-9) def search(query_embedding, doc_embeddings, doc_ids, top_k10, threshold0.55): scores [] for idx, doc_emb in enumerate(doc_embeddings): sim cosine_similarity(query_embedding, doc_emb) if sim threshold: scores.append((sim, doc_ids[idx])) scores.sort(reverseTrue, keylambda x: x[0]) return scores[:top_k]这个检索函数故意写得朴素但它把一个很重要的点突出了相关性阈值。一开始我没设阈值结果无论问题跟知识库有没有关系向量库都会返回最相似的几条模型拿着完全无关的资料硬答编造出看起来很合理的内容。加入阈值之后没有相关内容时系统会明确告诉用户资料库中没有找到相关信息这比瞎编强一百倍。2.3 提示词模板与上下文拼装结构化的力量生成层的核心不是提示词写得多花哨而是结构清晰、让模型明确知道每一部分的角色。我使用的Prompt模板基本长这样你是一个企业知识库助手。请严格基于下面提供的资料回答问题。 如果资料中找不到答案请直接说资料库中没有找到相关信息不要自行编造。 [资料开始] doc idfile-001 chunk12报销标准……/doc doc idfile-003 chunk45请假流程……/doc [资料结束] 用户问题{question} 要求 1. 回答必须引用对应的资料编号格式为[1][2] 2. 如果多个资料存在矛盾请指出来 3. 回答控制在200字以内。拼装顺序也很重要。我的经验是系统指令在最前然后是检索到的文档片段最后放用户问题。因为大多数模型对文本尾部的内容注意力更高用户问题放在最后模型在生成时会更贴合当前提问。另外每条检索片段我强制加上document ID和chunk编号不仅方便溯源还能让模型在回答时输出引用标记这对后续人工核查答案特别有用。2.4 把完整链路串起来当你把切分、向量化、检索、拼装都做完之后一个最小的RAG服务其实就是一个函数def rag_answer(question): q_emb embed(question) hits hybrid_search(question, q_emb, top_k5) context format_hits(hits) prompt render_prompt(context, question) answer llm_generate(prompt, temperature0.2) return answer, hits别小看这个函数它就是整个AI应用的地基。我后来发现凡是地基没打好、直接上框架的项目最后排查问题的时候都要回到这一层——切分是不是坏了检索是不是召回了错误内容连接口超时了还是模型拒绝回答都得一层层查。自己手写过一遍之后你再去看那些成熟框架的文档理解速度是碾压级的因为你已经知道每个模块在解决什么问题。3. 评估AI应用里最容易被忽视却决定成败的一环3.1 离线评测集别拿FAQ当唯一标准我见过特别多团队说我们评估过了,一问怎么评估的答拿文档里的几十个FAQ试了一下。问题在于FAQ通常是干净、完整、陈述式的而真实用户提问是碎片化、模糊、带口语的。真正好用的评测集应该来自真实日志哪怕只有五十条也比自己编的完美问题有价值得多。我建立评测集的方法是分几步走。第一步从线上日志里抽出真实用户问题去掉重复和涉及隐私的内容留下大概一百到三百条第二步按类型给这些问题打标签比如事实查询流程咨询比较类模糊表达知识库外问题第三步为每条问题人工写标准答案标准答案不需要精雕细琢但要标注清楚它依赖哪些文档第四步把这些数据和答案一起版本化管理每次改动都要全量回归。这份评测集里的知识库外问题是最容易被忽视但最不能缺的一类。你要专门往评测集里塞一些知识库覆盖之外的问题比如用户突然问今天天气怎么样答案应该是拒绝回答而不是硬编。很多AI应用翻车都是翻在这里。3.2 我实际在用的评估指标从零开始做评估你不需要上来就搞复杂的指标我建议先用四个简单但实用的维度它们分别覆盖检索和生成两个层面指标计算方式说明检索命中率 Recall5标准答案依赖的文档是否出现在Top5结果中衡量检索层质量不取决于模型回答相关性人工打分或模型打分1-5分衡量最终答案对问题的覆盖程度忠实度Groundedness检查回答中事实性论断是否都有检索片段作为支撑专门防幻觉看哪些论断无源可溯拒绝率知识库外问题的正确拒绝比例不该答的坚决不答检索命中率我建议最先做因为它能干净地隔离问题如果检索就已经失败了那后面生成层怎么调都没用。忠实度的自动化我常用一个笨办法把回答拆成若干事实性短句逐句去检索片段里做包含或相似度匹配找不支撑的句子就标红。不是特别严谨但对于拦截模型自己编造这类问题非常有效。3.3 让评测跑起来的闭环评估最忌讳的是一次性工作。我现在养成的习惯是任何改动不管是改了切分窗口大小、换了向量模型、还是微调了Prompt都必须先在评测集上跑一遍对比前后得分再决定上不上线。你可以写一个很简单的评测脚本输入旧版本和新版本的回答结果输出每个维度打分的变化。我见过太多人在那里凭感觉调参数——把temperature从0.7改到0.1觉得好像更稳定了把topK从3改到5觉得好像更全了这些感觉如果没有评测数据支撑上线之后大概率回退。AI工程的进步不是靠感觉是靠样本集和分数。4. 从原型到生产延迟、成本与稳定性三大关4.1 延迟预算把时间花在最值得的地方AI应用和传统接口最大的体验差异在延迟。用户能容忍传统接口200毫秒但不会容忍一个问答接口30秒才出结果。我算过一笔典型延迟账问题向量化大约100到300毫秒检索加排序在50毫秒以内大模型生成反而是大头按输出三百个字符计算可能要两到五秒。也就是说生成占掉了大部分时间你想优化延迟主要得从生成侧下手。第一个手段是流式输出。从用户点击发送到看到第一个字控制在五百毫秒内体感就会好很多用户看到字在流动就不觉得慢。第二个手段是缓存。对高频问题做语义缓存用向量相似度判断用户当前问题和历史上某个已回答问题是不是同一个意思如果是就直接返回历史答案相似度阈值可以设置在0.92以上这样命中率虽然不算高但命中后的响应时间直接降到几十毫秒。第三个手段是路由简单问题比如报销上限是多少路由给小模型回答复杂多步骤问题才调用大模型成本更低响应更快。4.2 成本控制大模型的每一分钱都要花明白做AI工程成本不是运维账单上的一个数字它直接决定了你产品能服务多少用户。一个容易被忽略的事实是输入上下文越长每次调用花的钱越多。RAG系统如果一次性塞进五千token的资料即便只输出两百token计费也主要被输入吃掉了。因此我的做法是给每次请求设置上下文预算比如四千token封顶。检索结果按照相关性得分从高往低填超过了预算就丢弃同时会话型应用会做历史摘要——把之前多轮对话压缩成一段简短总结而不是把所有历史消息原文都塞进上下文。这样做的代价是模型偶尔会漏掉很早之前的一个细节但对大多数场景来说成本节省接近一半完全划算。4.3 降级兜底让系统在异常时仍然体面模型接口不稳定、返回内容不合规、检索结果为空这些不是万一会遇到的情况而是日常。一个成熟的AI工程必须有明确的降级链路。我给自己的系统设计了三级降级第一级当检索相似度全部低于阈值模型直接回答资料库暂无相关信息不需要生成第二级当模型接口超时或返回错误重试一次重试仍失败则返回预设的提示语并记录告警第三级当检测到用户输入存在注入风险比如要求忽略之前的指令或恶意内容直接拦截不进检索和生成链路。千万别觉得降级方案不够AI。对一个生产系统来说可靠比聪明重要。我见过某个项目追求极致回答质量结果模型供应商出问题的时候全站接口全部报错用户投诉铺天盖地。现在我把宁可说不知道不能说错当成铁律。4.4 可观测性没有日志的AI应用等于盲飞传统应用排查问题看报错堆栈AI应用排查问题看的是输入输出对。我强烈建议从第一天就建立结构化日志每条请求至少记录以下字段字段说明request_id全链路追踪IDquery_text用户原始问题retrieved_chunks实际召回的片段ID和相似度得分prompt_text最终送入模型的完整Promptresponse_text模型原始输出latency_ms每个环节耗时token_usage输入和输出token数eval_flag是否命中缓存/是否触发了降级这些日志至少有四个用途出问题时可复现成本可按用户维度分摊检索质量可以对照人工标注持续优化还能用来持续扩充评测集——线上真实问题和模型回答就是最好的评估素材。我现在的排查习惯是用户说答得不好我先去日志里定位检索环节召回了什么基本能判断是切分层的问题、向量模型的问题还是提示词的问题。5. 那些看起来合理但实战会坑人的默认选择5.1 四个必然踩过的坑做AI工程这一年多我总结了四个默认选择基本每个新手都会踩而且踩的姿势都一模一样。第一个是上下文塞得越全越好。总觉得多给资料模型就答得准结果塞了七八个片段进去模型根本分不清主次容易被不相关的片段带偏。现在我严格控制检索片段不超过五个、每个不超过四百token。第二个是离线测几条没问题就上线。离线几条你精心设计的用例根本覆盖不了线上的脏数据。我做过一次很丢脸的演示样品问题回答得特别好观众随机问了一个问题模型一本正经地编了一个不存在的条款因为没有检索到相关内容而系统也没有触发无资料拒绝的机制。第三个是向量模型选好就不用动了。文档会更新、用户的提问方式会变一段时间之后你会发现检索准确率悄悄下滑。这很正常所以要有定期评测的习惯并保留历史向量版本方便回退。第四个更隐蔽默认用户会好好说话。真实用户不会给你一个完美的长问题他可能只输入报销合同去年甚至还有错别字。我的解决方案是加了一层查询改写先让模型把用户短查询扩写成两到三个不同角度的完整问题再分别检索合并结果。实测带上改写之后知识库问答的整体召回率能提升十几个百分点。5.2 接下来怎么走从RAG到Agent和微调当你亲手把RAG闭环做扎实之后下一步的方向大概有三条。一是往Agent方向走让模型能够自己决定调用哪些工具、按什么顺序执行任务。这个时候你要重点学习的是规划能力和工具调用的可靠性设计比如要把Agent的每一步操作都记录成结构化轨迹方便回溯。二是往微调方向走如果你发现提示词怎么调整都达不到效果或者你的场景需要固定的输出格式小规模监督微调值得尝试。微调不是从零训练而是用高质量样本让模型学会你的领域格式和表达习惯。三是往工程纵深走检索重排、多路召回、向量数据库选型、模型网关、灰度发布这些都是AI工程和基础设施交叉的地带越往后越值钱。最后说一点我个人的体会AI工程入门最忌讳的就是框架焦虑。今天这个框架火明天那个工具出来跟着追永远追不完。踏踏实实从切分、向量化、检索、评估这一条线手写一遍哪怕只有一千行代码你会获得一种任何新框架都夺不走的能力——理解系统每个环节的能力。之后再遇到任何新工具你都是在评估它帮你解决了哪个环节的问题而不是被它牵着走。