资讯动态

大模型应用工程化:RAG与多智能体的生产级实践

发布时间:2026/10/17 18:14:16 来源:尧图企业网站定制
直接进入正文。这两年做大模型应用我从单个脚本起步到现在维护多套RAG和智能体服务最大的感受是模型能力反而成了最简单的一环真正吃时间的是工程规范。同样的模型有人能做出一套稳定可交付的检索增强系统有人写出来的东西换个数据源就崩。差别不在模型选型而在有没有一套从数据流、调用链、异常处理到评测回归都约束清楚的工程框架。这篇文章把我自己沉淀下来的一套大模型工程规范完整拆开从RAG的文档解析、切块、召回、重排到多智能体的编排模式、工具治理、上下文传递再到日志、评测、成本治理。核心代码都贴出来了不是教学级别的示例是我在线上环境用过、踩过坑之后打磨过的写法。适合正在把大模型应用往生产环境推的团队也适合想弄清楚“RAG和智能体到底该怎么工程化落地”的个人开发者。读完你至少能避开我踩过的十几个坑。1. 工程分层先把规范建立在结构上1.1 五层参考架构我做大模型应用的第一版代码全部堆在一个文件里文档处理、向量化、检索、调用模型、拼提示词全在一条Python脚本里。跑Demo没问题一旦要和业务系统对接、要多人协作、要稳定服务立刻乱成一团。后来我按照分层思想重新组织代码整个系统拆成五层效果立竿见影。这五层分别是数据接入层负责文档解析、清洗、格式转换输出统一的文本块结构索引构建层负责切块、向量化、写库以及索引的策略配置检索服务层负责查询理解、召回、重排输出带相关性分数和引用来源的结果编排交互层负责提示词拼装、上下文管理、模型调用、流式输出治理监控层负责日志追踪、评测回归、成本统计、安全过滤这套分层的核心价值是隔离变化。模型API升级了只动编排交互层换了向量库只动索引层业务方想调整检索逻辑完全不用碰底层存储。我见过太多团队重构三个月都完成不了就是因为所有逻辑纠缠在一起牵一发动全身。1.2 分层要配合两套约定光分层还不够必须配合两套约定才跑得动。第一套是接口约定。层与层之间不直接调用对方的内部对象而是通过标准的数据结构传递。比如从索引层返回的检索结果统一是RetrievalResult对象字段固定为chunk_id、content、score、source、metadata。这样上层逻辑不管底层用的是某个向量库还是自研检索拿到的东西都一样。第二套是协议约定。每层对外暴露的能力必须是同步接口或统一异步接口所有内部重试、超时、降级逻辑封装在层内对外只暴露“成功或失败”的结果。我在实际项目里对这点体会很深一开始让上层直接拼ES查询语句结果每次检索规则调整都要改五六个地方后来强制收敛成search(query, top_k)这个统一入口改动面小了一个量级。分层的最终目的是让“出问题的时候能快速定位”。线上检索不准先看索引构建层的数据有没有问题再查检索服务层的召回逻辑一层层剥效率极高。2. RAG落地别只做一个漂亮的Demo2.1 文档解析决定检索天花板的一步很多团队做RAG把80%的精力花在调切块大小和选向量模型上文档解析草草处理。这是个很严重的误区。文档解析的质量直接决定检索的天花板切块和向量化只决定你能在多大程度上接近这个天花板。以最常见的PDF和Word文档为例我实际生产环境里的处理经验是PDF必须走两层解析。第一层抽文本层第二层做OCR兜底。只处理文本层的PDF遇到扫描件全废。OCR这块要注意表格类的扫描件要用版面分析先定位表格区域再做结构化提取直接整页OCR出来的文本顺序是乱的带标题结构的文档优先保留标题的层级关系。解析结果里必须包含h1、h2、h3等元信息后续切块要依靠这些标题做边界表格不要平铺转文本。平铺之后一行行的数据被切断检索时完全没法对齐。表格要转成Markdown表格尽量保住二维结构清理页眉页脚、页码、重复水印。这些噪声不清理检索时会被当成正文返回非常影响体验我见过最典型的失败案例某团队接入一批行业研报PDF解析出来倒是能搜到内容但用户问“XX市场今年规模多少”检索返回的却是研报附录里的免责声明。查下来就是页脚文字被当成正文索引了。所以第一步一定要干净。2.2 切块策略参数要跟着场景走切块这件事没有银弹只有“在当前场景下合理”的参数。我踩过的坑是从网上抄了一个固定长度切块的参数就上线结果问答质量非常不稳定。后来总结出几条实际可用经验第一标题结构完整优先于固定长度。一份文档如果有清晰的Markdown或HTML标题就按标题层级切块。切完如果块太长再在这个块内部按段落切。这样切出来的块有语义边界和用户问题的匹配度远高于纯按字数切的块。第二块与块之间要有重叠。我常用的重叠是50到100个字符或者按句重叠一到两句。重叠的目的是避免问题恰好落在两个块的接缝处两边信息都不完整导致检索命中失败。有人觉得重叠浪费存储实际多点向量而已换来的是检索稳定性划算。第三块长要根据知识领域的粒度来定。法律条文按条款切一份合同按条款层级切通用技术文档块长在600到1000个字符左右比较合适。如果你的业务中问题通常需要跨段落综合回答块长偏大一些如果问题是条款级的“xx规则是什么”块长就要偏小确保切出来的块专门对应一条规则。下面是我实际在用的标题感知切块代码基于常见文档解析结果实现核心逻辑是按标题序列找边界构造语义块并配置了重叠片段拼接from dataclasses import dataclass, field from typing import List, Optional dataclass class ParsedBlock: text: str heading_path: List[str] level: int start_pos: int def split_by_headings(parsed_blocks: List[ParsedBlock], max_len: int 900, overlap_len: int 80) - List[str]: 按标题结构聚合切块超过max_len时内部继续拆分并加入overlap chunks: List[str] [] current_parts: List[str] [] current_headings: List[str] [] for block in parsed_blocks: heading block.heading_path[-1] if block.heading_path else block_lines [] if block.heading_path ! current_headings: # 新标题层级结束当前chunk if current_parts: chunks.append(\n.join(current_parts)) current_parts [] current_headings block.heading_path block_lines.append(f# { | .join(block.heading_path)}) block_lines.append(block.text) text_block \n.join(block_lines) current_parts.append(text_block) acc_len sum(len(p) for p in current_parts) if acc_len max_len: # 合并为一个chunk后将末尾overlap_len字符保留续接 merged \n.join(current_parts) chunks.append(merged) tail find_overlap_tail(merged, overlap_len) current_parts [tail] if tail else [] if current_parts: chunks.append(\n.join(current_parts)) return deduplicate_chunks(chunks) def find_overlap_tail(text: str, length: int) - str: 从文本末尾切出重叠部分尽量从标点后开始避免从词语中间切断 if len(text) length: return tail text[-length:] for punct in 。\n: idx tail.find(punct) if idx ! -1 and idx length - 10: return tail[idx 1:] return tail def deduplicate_chunks(chunks: List[str]) - List[str]: seen set() result [] for c in chunks: if c not in seen: seen.add(c) result.append(c) return resultdeduplicate_chunks这个函数很有必要。不加它重叠片段可能在合并后和下一轮的chunk内容重复导致向量库里出现大量重复内容检索时同一篇文章占了好几个结果位。2.3 召回链路双路加策略才是标配我早期做RAG召回只用向量相似度这一路后来发现很多问题。最典型的是用户问“产品A的价格是多少”向量检索返回的都是语义相近的“产品价格说明”但真正的品类名称不一致导致命中率上不去。后来我改成双路召回效果提升非常明显。双路召回指同时跑向量召回和关键词召回。向量召回负责处理语义改写、同义表达关键词召回负责处理精确匹配、专有名词、编号、产品型号。两路结果做分数归一化再做加权融合。加权系数我一般是向量分数权重0.6关键词命中权重0.4如果关键词命中的是用户问题中的专有名词额外给一个加成系数1.2具体实现里关键词召回用倒排索引或文本搜索都行核心是匹配词根和短语。融合之后再用重排模型精排。重排模型不一定用最大的一些轻量级的交叉编码器就够用。重排的典型收益是召回阶段前50个候选中真正相关的结果从10个提升到25个以上Top5准确率能翻倍。下面是双路召回加融合的代码框架抽出接口后你可以接任意向量库和文本检索引擎def hybrid_search(query: str, top_k: int 20) - List[RetrievalResult]: # 向量召回 query_vec embedding_client.embed_query(query) vector_hits vector_db.search(query_vec, top_ktop_k) # 关键词召回 key_terms extract_key_terms(query) keyword_hits keyword_index.search(query, key_terms, top_ktop_k) # 分数归一化 norm_vector min_max_norm([h.score for h in vector_hits]) norm_keyword min_max_norm([h.score for h in keyword_hits]) # 加权融合map 结构用 doc_id 做连接键 fused fuse_by_weight(vector_hits, keyword_hits, w_vector0.6, w_keyword0.4, term_boostkey_terms) fused.sort(keylambda x: x.final_score, reverseTrue) # 重排精排 reranked rerank_model.predict(query, [f.text for f in fused[:50]]) return attach_scores(fused, reranked)融合时最需要注意的一个坑是“同一篇文章的双路分数会被重复计入”。如果不按文章的doc_id去重一个文档既在向量路命中又在关键词路命中分数叠加后会霸占榜单。我一般先做一个按doc_id分组的合并逻辑再进入加权计算。2.4 缓存与流式体验优化的关键动作RAG服务上线后如果每个问题都从头跑一遍检索加模型生成成本和延迟都扛不住。我做了两层缓存实测QPS提升明显。第一层是向量检索缓存。相同或近似问题的向量召回结果可以缓存起来但注意不能用问题原文做缓存key必须用问题归一化后的语义指纹。语义指纹可以是embedding向量的二进制哈希也可以用轻量分类模型输出的意图标签加关键词组合做key。第二层是完整回答缓存。常见问题池里的标准问法直接把答案拼接引用来源返回完全不用调模型。这里要注意缓存过期策略和内容更新联动知识库更新后涉及到的回答缓存必须主动失效。流式输出这块工程规范上要尊重一个原则先返回引用来源再流式生成正文。用户拿到引用来源后可以立刻判断准确性正文慢慢出也影响不大。这是我现在做所有问答类应用的标准做法。async def stream_with_cache(query: str, source_ids: List[str]): cache_key semantic_fingerprint(query) if await answer_cache.exists(cache_key): async for token in answer_cache.stream(cache_key): yield token return async for token in llm_client.stream(query, source_idssource_ids): yield token await answer_cache.save(cache_key, streamed_text)注意上面的代码里我把保存放在流式之后这看起来合理但实际生产中要做“第一个token返回前就记录流状态”防止流式中断导致缓存写入半截内容。我加了一个状态标识一旦流开始就写一个streaming标记流结束才允许覆盖成完整内容。否则客户端断开导致流中断缓存里存了半截答案后面所有命中缓存的人都会收到截断回复。3. 多智能体架构从单Agent到编排3.1 单Agent的局限性单Agent单一大模型驱动的自主代理能做很多事情但我在生产环境里用下来它的局限性很明显。第一是上下文长度瓶颈。一个Agent跑一个复杂任务要工具调用记录、思考过程、中间结果推理到后面经常发现前面的关键信息被截断了。第二是职责混乱。同一个Agent既要理解用户需求又要做信息检索还要写代码执行分析一旦任务复杂输出质量就迅速下滑。第三是URL难以约束。你很难给一个单Agent设定“什么该做什么不该做”的清晰边界它不是不想遵守是上下文里根本没有足够的位置放这些约束。把大任务拆给多个Agent每个Agent维护自己的上下文和工具集再由一个编排层统筹任务分配和结果汇总这是目前我看到的最稳定的工程解法。相比单Agent多智能体架构在可维护性和错误隔离上有本质提升代价是复杂度上了一个台阶。3.2 三种主流的编排模式我梳理过市面上主流的多智能体编排模式按我的实践体验排序主管-工人模式Supervisor-Worker适合大多数业务场景。一个主管Agent负责任务分解、分发、汇总多个工人Agent各司其职。主管不直接干具体活它只做计划者和协调者。这个模式清晰、可控、好排障。辩论模式Debate适合需要多角度判断的任务。比如做内容审核一个Agent模拟用户视角生成内容一个Agent做事实核查两个Agent相互质询最终由仲裁Agent给出结论。效果很好但Token消耗非常高。流水线模式Pipeline适合步骤固定的批处理任务。每个Agent只处理自己负责的环节结果传给下一个Agent比如先调研、再总结、最后写报告。流水线模式的优点是流程确定缺点是灵活性差中途一旦需要变更路径基本上要人工干预。我自己的经验是团队第一次做多智能体从主管-工人模式起步最稳。先跑通链路再考虑复杂模式不要一上来就堆十几个Agent互相聊天。3.3 工具与上下文的工程规范多智能体系统里每个Agent手里的工具集必须显式声明不能让他们自由发挥。我给每个Agent配工具时遵循三条规范工具Schema要完整。一个工具必须有名称、描述、参数列表、返回值说明、样例输入输出。描述里要写清楚“什么情况下用这个工具”和“什么情况下别用这个工具”LLM对描述的理解直接决定调用准确性工具注册带权限。工人Agent只能看到自己被分配的工具主管Agent可以看到所有工具的注册信息但也不直接调用而是负责分配到人工具调用要做审计。所有工具的入参和出参全部记录到日志这条对排查“Agent为什么做错决策”至关重要上下文传递是另一个关键点。多Agent之间共享上下文会产生严重的信息污染尤其是当两个Agent引用同一个文档片段时各自的解读方向不同放在共享上下文里会互相干扰。我现在的做法是“上下文隔离、结果上报”工人Agent只保留自己执行过程中的上下文完成后把结构化结果上传给主管主管保留全部结果摘要不保存具体执行细节。3.4 多智能体代码示例主管-工人版下面这段代码是我基于某个通用Agent SDK思路整理的主管-工人模式的骨架重点在任务分配和结果回收上。核心逻辑是主管注册任务分发函数工人注册自己的工具集通过一个内存消息队列做异步交互。from dataclasses import dataclass, field from typing import Dict, List, Callable, Any import asyncio dataclass class AgentTask: task_id: str agent_name: str instruction: str context: Dict[str, Any] field(default_factorydict) class AgentWorker: def __init__(self, name: str, tools: Dict[str, Callable]): self.name name self.tools tools async def run(self, task: AgentTask) - Dict[str, Any]: # 实际调用LLM完成工具选择与参数生成这里简化为直接执行 result await execute_with_tools( instructiontask.instruction, toolsself.tools, contexttask.context ) return {agent: self.name, task_id: task.task_id, result: result} class Supervisor: def __init__(self, workers: List[AgentWorker]): self.workers {w.name: w for w in workers} self.task_queue asyncio.Queue() async def dispatch(self, master_instruction: str): # 主管LLM分析任务并拆解成AgentTask plan await plan_with_llm(master_instruction, available_agentslist(self.workers.keys())) tasks [AgentTask(task_idi, agent_namep.agent, instructionp.instruction) for i, p in enumerate(plan)] results {} for t in tasks: worker self.workers[t.agent_name] results[t.task_id] await worker.run(t) return results这一段代码看起来简单但生产环境里真正的复杂度在plan_with_llm这一步。它要保证输出的计划是结构化的且工人Agent名称必须存在于注册表。为了这个目的我会用输出解析器和枚举约束强制LLM输出的agent字段必须从集群列表里选择防止它虚构一个不存在的Agent名称。还有一个细节工人Agent执行结束后要把中间步骤和最终结论分开返回。中间步骤只留存于日志最终结论才进结果汇总。否则主管摘要的时候会“过拟合”到某个工人Agent的详细日志上导致汇总跑偏。4. 工程治理可观测性、评测与成本4.1 日志规范把一次请求当作一条全链路大模型应用的服务链路比传统应用长很多一个有代表性的RAG加多智能体请求会经过网关、意图识别、检索服务、重排服务、主管Agent、工人Agent、模型推理、后处理。任何一个环节出问题没有规范日志根本定位不了。我强制要求每条日志必须带request_id和trace_id。request_id对应业务侧的一次请求trace_id对应该请求内部的整条执行链路。所有日志模块统一输出JSON格式字段固定包含时间戳、服务名、层级、事件类型、耗时、Token数、输入摘要、输出摘要、异常信息。下面是字段约定表格你可以直接拿去当模板字段说明示例request_id业务请求唯一标识req_8f3a2ctrace_id链路追踪标识trace_b7e19flayer当前所属层级retrieval/agent/llmevent事件类型search_start/tool_call/llm_streamlatency_ms本环节耗时234token_usageToken消耗{prompt: 1200, completion: 300}input_digest输入摘要问题: 产品保修多久output_digest输出摘要命中3个片段, Top1得分0.87日志写了还必须有人看。我每晚会跑一个日志聚合任务统计各环节耗时分布、错误率、Token消耗Top榜。这些指标是优化检索策略和Agent编排的第一手依据。4.2 Prompt与流程的版本化管理Prompt在工程上也是代码必须纳入版本管理。很多团队把Prompt写在代码文件里改一次Prompt就要发一次版业务方想在线改个话术都要排队。我的做法是Prompt全部外置到配置平台或独立的Git仓库运行时动态加载。版本化有几个关键约束Promp变更必须有版本号线上要能一键回滚同一个Prompt模板的多语言版本要分开管理关键链路比如主管Agent的任务分解Prompt变更必须走评测回归不能拍脑袋直接上。和Prompt配套的还有“流程配置”。多智能体的编排流程不要让LLM完全自由发挥要支持流程模板。比如“调研类任务”的模板固定是“扩散调研-聚焦验证-汇总报告”三步“问答类任务”的模板固定是“检索-标引-成文”。流程模板的好处是出问题时能按模板的环节逐段排查同时给LLM的发挥空间留下边界。4.3 评测集的构建与回归没有评测集的RAG项目就像没有单元测试的后端工程上线全靠运气。我从第一天开始就建评测集而且不是随便攒几条问答而是按业务场景构建标准问答集每个知识域挑10到20条经典问题带标准答案和引用来源边界测试集故意构造模糊问题、包含错别字的问题、需要跨多个文档综合回答的问题对抗集包括有答案但容易被检索误导的问题、知识库没有对应答案的问题、需要拒绝回答的敏感问题评测指标方面我除了看最终答案的准确率还看“检索命中率”和“引用来源正确率”。检索命中率指前3个检索结果里有没有包含标准答案所在的块这个指标能定位问题是出在检索层还是生成层。我每周跑一次回归每次跑完自动生成一份评测报告。报告里不只看分数还列出典型的失败case以及失败原因分类。这些分类就是优化检索策略和Prompt的直接输入。4.4 成本与延迟的日常治理大模型应用的成本不可控是一个常见问题。多智能体架构尤其如此因为一次任务会多次调用模型。我做了三件事来控制成本。第一最小化模型层级。能用轻量模型完成的环节如意图分类、工具调用的参数抽取绝不用大模型只有最终生成和复杂规划才用最强模型。这个策略能省下百分之五十以上的Token成本。第二引入Token预算机制。每个Agent分配明确的Token预算执行过程中消耗完了就停止不允许无限扩展。这里我是在编排层写了一个预算计数器每个Agent启动时初始化每次工具调用和模型推理都扣减触顶即终结并返回部分结果。第三动态降级。线上压力大的时候把重排模型从高阶模型降级为策略融合排序把完整生成降级为极简模式只输出结论多智能体编排从完全自主降级为固定模板模式。这套降级机制让我在流量翻倍时不需要对应翻倍预算。5. 生产环境常见问题与排查实录5.1 问题速查表我这几年在生产环境里遇到的高频问题整理成一张速查表每个问题都附上排查路径和解决方案。现象可能原因排查路径解决方案检索结果相关但答案错误生成时上下文截断或关键信息被后置查llm_stream日志的Prompt大小和上下文截断点压缩检索片段强制引用片段置顶检索结果不相关文档解析阶段丢失标题结构查索引层日志确认向量化的文本是否含标题信息修复文档解析流程重新索引同一个问题结果时好时坏双路召回融合权重未做归一化对比向量路和关键词路的分数分布统一z-score归一化加doc_id去重Agent陷入重复调用同一个工具工具描述未说明失败后的退出条件查工具调用审计日志看循环规律在工具描述中增加终止条件配置最大调用次数多Agent结果互相覆盖共享了全局上下文变量查trace链路中上下文写入顺序改为上下文隔离结果按任务ID上报Token消耗异常飙升工人Agent重试过多或上下文重复传入查Token统计日志定位消耗最大的环节增加重试次数上限 Token预算机制流式输出截断客户端断开导致缓存写入半截查缓存状态标识是否卡在streaming增加流中断后的缓存清理任务回答内容过时知识库更新后缓存未失效查缓存命中时长分布建设知识库版本号版本变更自动清缓存5.2 三个让我印象深刻的排障过程第一个是“检索正常但回答引用了错误的段落”。排查下来发现重排模型给相关文档打的分数没问题但生成阶段拼接引用片段时代码把所有来源按原始文档顺序排列没有按相关性重排。模型生成时参考的是Prompt中靠前的来源也就是不相关的部分。修复方案很直接按重排分数降序排列引用片段。第二个是“多Agent系统在任务拆解时经常虚构Agent名称”。有次主管Agent计划输出里出现一个叫“数据清洗专家”的Agent但注册表里根本没有这个Agent任务无限期等待。根因是规划Prompt里没有对可用的Agent名单进行枚举和格式约束。修复之后我在输出解析层加了一层“名称校验器”一旦发现虚构Agent直接拒绝该计划重新走一次规划循环。第三个是“用户问题命中知识库中的免责声明”。这条我前面提到过排障之后我们这个项目的文档解析流程强制加入了页眉页脚识别模块同时对索引前的正文做“区块有效性”判断正文为空、内容过于短、或者包含免责声明的区块直接排除。这些问题的共同点是什么都不是模型能力不够而是工程链路里某个环节的规范缺失。模型被冤枉的次数远比我们想象的多。5.3 避坑经验三条我坚持的原则第一条永远不要让LLM直接访问整个知识库。检索层必须存在而且检索失败时宁可返回“没有相关知识”也不要让模型胡编。这是底线。第二条对用户的原始输入做安全过滤不做不行。“鹅不会飞”这类假设性问题还好最怕的是输入被注入恶意指令绕过Agent的系统提示词。我在网关层做了指令注入检测同时对所有外部输入做内容安全审核有风险的直接拦截。第三条任何Prompt或流程改动都要配一个可回滚的开关。我供过一次教训优化了主管Agent的任务分解Prompt直接上线结果整整半天线上任务执行质量下滑。后来所有Prompt改动都走灰度开关先放5%流量观察再看全量。结尾一点个人经验写到这里把我在大模型工程化这条路上踩过的坑基本都整理出来了。最后说一点最实在的感受做RAG和做多智能体真正的分水岭不是模型选得多新、框架用得多花哨而是你有没有把数据流、任务流、监控流这三件事理顺。数据流不顺检索质量全看运气任务流不顺多智能体天天吵架监控流不顺线上出了问题只能靠猜。我见过不少团队投入大量精力去试各种新模型和新框架却在文档解析、日志规范化、评测集这些基础工程上缺课。结果就是Demo惊艳全场上线一周就崩盘。反过来那些把基础工程做到位的团队哪怕模型不是最新的交付质量和稳定性也远超同行。如果你现在正在做RAG或智能体项目我建议从今天开始做两件事第一把日志规范化确保每次请求都有全链路可追踪第二建一个哪怕很小的评测集开始记录每次改动的影响。这两件事在前期看起来不起眼三个月后再回头看它们会是你项目能走多远的关键支撑。希望这些经验能帮你少踩几个坑。你的项目如果遇到这里没覆盖到的具体问题欢迎留言交流我会尽量把排查思路整理出来。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑