最近圈子里有个说法单一RAG已经不够看了多代理混合RAG才是企业知识库问答的新方向。我第一次看到这个词的时候心里也打了个问号这不就是把检索和生成拆成几步再串起来吗直到自己动手搭完一版才发现里面值得展开讲的细节远比想象的多。这篇文章我想把完整的技术链路、源码结构和落地过程中踩过的坑一次性讲清楚多代理到底代理了什么、混合检索到底在混合什么、为什么路由节点能让整体效果上一个台阶以及一个零基础的人该如何在半小时内把整套系统跑起来。如果你也被“AI智能体”“多代理”这些词绕得头晕又不想停在概念层面那这篇文章应该挺对症。整个项目我选择了一套相对克制、不烧钱的方案编排用LangGraph检索用Elasticsearch的BM25加Qdrant向量库双通道重排用本地交叉编码器模型层通过OpenAI兼容接口按需分配便宜模型和强模型。这套组合跑下来内部问答的准确率从单路检索的74%左右提到了90%而且单轮查询成本基本控制在几分钱级别。下面按我搭建时的顺序来拆解。1. 多代理混合RAG到底在解决什么问题1.1 单条检索链路的瓶颈知识库问答为什么经常翻车我们最开始做的知识库问答就是标准的单路向量检索把文档切块、embedding入库、用户提问后算相似度、取TopK丢给大模型生成。流程很简单但上线之后问题一个接一个。最典型的是对角线式飘移。比如用户问“我买的这个设备保修期内坏了怎么办”向量检索可能返回一整篇售后政策的开头里面没有一句提到“保修期内维修流程”再比如用户问“如何修改登录密码”向量检索会去寻找语义上像“登录密码修改步骤”的片段但如果文档里写的是“账号安全设置说明”召回结果就变成了看似相关、实则绕弯的内容。关键词检索则走向另一个极端。它对型号、编号、报错代码这类精确串非常敏感但面对“设备频繁重启是什么原因”这种自然语言表达纯BM25经常召回不到深层语义相关的片段。还有一种情况是知识库里混着PDF、Word、Markdown、表格单一通道很难同时覆盖结构化数据和非结构化文本。换句话说任何单一检索策略都是在赌一种相似度假设。向量模型擅长语义但遇到专有名词、约定俗称的简写会漂移BM25擅长精确匹配但理解不了同义改写。混合检索的意义就是同时保留两种“猜法”再用重排把错误答案压下去。1.2 多代理不是多个聊天机器人而是职责拆分的决策节点很多人听到“多代理”第一反应是像Manus那样一串Agent互相聊天其实在企业知识库场景里完全不需要那么戏剧化。我这里的多代理是把一次问答拆成五个独立的决策节点请求入口、路由代理、检索代理、验证代理、生成代理。每个节点只干一件事。路由代理判断问题该走知识库还是直接闲聊检索代理按路由结果决定用关键词、向量还是混合召回验证代理检查召回片段和答案之间是否真的对得上生成代理负责组织语言并引用来源。这样拆的好处是任何一个环节出问题都能单独调优而不是在整条链路上盲猜。我用一张表把这几个角色的职责写清楚后面讲代码时会反复提到它们节点核心职责输入输出路由代理判断问题类型用户原始提问category: chat/vector/keyword/hybrid检索代理执行混合召回问题与路由结果候选文档片段列表验证代理校验证据与答案一致性答案 检索片段打分或拒绝回答生成代理生成最终答案验证通过后的片段带引用的回答所以整体数据流是这样的用户问题先进路由路由把问题分成闲聊、精确查询、语义查询、需要混合召回的复杂查询四类然后根据类型走不同检索路径召回结果先RRF融合、再交给重排模型卡阈值验证通过后进入生成阶段。整个链路是有分支、有回退、有校验的而不是一条直线走到底。我把这种架构称为“分层决策 多路召回 验证兜底”。听起来唬人拆开其实就是上面那张表后面每一章都是对这张表的代码化。2. 环境准备与框架选型为什么我选了LangGraph2.1 多代理编排框架对比LangGraph、CrewAI、AutoGen我最早用的是CrewAI写几行代码就能让角色互相“商量”demo效果很炫。但到了需要精确控制检索流程的时候问题就来了CrewAI的Agent本身带有自主决策倾向它会自己决定调用哪个工具、按什么顺序调用这在一个知识库问答链路里是不可控的。我希望路由是路由、检索是检索每一步都是可观测、可回滚的。AutoGen我也扫了一眼它的对话式代理更适合多角色辩论场景用来做RAG链路显得杀鸡用牛刀。最终我选了LangGraph理由很直接它把流程定义成状态图每个节点就是一个普通Python函数节点之间的边可以是条件边、并行边运行的时候能看到每一步state变化。这对排查“为什么这次检索没走混合通道”太有帮助了。对比表格如下框架控制粒度是否支持并行分支适合场景上手成本LangGraph细节点即函数支持Send并行精确编排RAG链路中需要理解图概念CrewAI粗Agent自主决策有限自主任务拆解低AutoGen对话流支持多Agent聊天多角色辩论/协作中高实际用下来LangGraph的学习曲线没有想象中陡。核心概念只有StateGraph、节点、边、条件边四个代码量不多但能把整个流程画成一张明确的拓扑图。项目里我锁定的是0.2.x版本这个阶段的API相对稳定后面跑源码时要注意别装成最新的大版本不同版本之间导入路径差异挺大。2.2 模型分层与成本预算不是所有环节都用同一个大模型零基础搭建最容易犯的错是让所有环节都调用同一个强模型。路由分类、检索片段打分、答案校验这些任务对模型能力要求并不高完全可以用便宜模型完成只有最终生成答案那一步才需要强模型来组织语言、融合多片段信息。我的模型分三层。第一层是路由和验证用gpt-4o-mini这类轻量模型单次调用大概消耗300到700个token第二层是重排用本地部署的交叉编码器bge-reranker-v2-m3完全不走外部API第三层才是答案生成用当前项目里配置的主生成模型每次大概消耗1000到1500个token。整套跑下来一次完整问答的token消耗在2000到2500之间。换算成钱的感受是在内部知识库量级下一天几千次查询月成本基本控制在几十元人民币以内。如果你们公司已经买了某些OpenAI兼容网关的配额成本还能更低。具体的环境变量配置长这样# .env 示例 OPENAI_BASE_URLhttps://your-gateway.example.com/v1 OPENAI_API_KEYsk-xxxx PRIMARY_MODELgpt-4o LIGHT_MODELgpt-4o-mini EMBED_MODELBAAI/bge-m3 RERANK_MODELBAAI/bge-reranker-v2-m3 ES_URLhttp://localhost:9200 QDRANT_URLhttp://localhost:6333 COLLECTION_NAMEenterprise_docs这里要特别提醒一句OPENAI_BASE_URL一定要配对。很多人喜欢本地起一个模型网关但不同网关对/v1路径的处理不一样如果填错代码里client.chat.completions.create会一直报404或401。我的建议是先在命令行里用curl测通网关接口再填进环境变量能省掉一晚上的排查时间。3. 核心代码拆解从路由代理到混合检索的完整实现3.1 路由代理让模型用结构化输出做决策路由代理是整个系统的入口它决定问题走哪条路。我用的方案是让轻量模型返回一个JSON字段而不是让它自由发挥。自由发挥的结果往往是模型在回复里夹带解释后面解析全靠正则非常脆弱。结构化输出的做法是给它一个很明确的system prompt并开启response_format{type: json_object}。核心代码如下import json from openai import OpenAI client OpenAI( base_urlos.getenv(OPENAI_BASE_URL), api_keyos.getenv(OPENAI_API_KEY), ) ROUTER_PROMPT 你是企业知识库的路由器。请判断用户问题属于哪一类只返回JSON对象。 类别定义 - chat: 寒暄、闲聊、与知识库无关的问题 - vector: 需要语义理解才能找到答案的问题例如如何提高设备散热效率 - keyword: 包含明确型号、编号、报错代码、产品名的问题例如BSC-2000报错E403 - hybrid: 既包含精确表达又需要额外语义理解的问题例如BSC-2000温度过高怎么办 输出格式{category: vector, reason: 简短说明} def route_question(question: str) - str: resp client.chat.completions.create( modelos.getenv(LIGHT_MODEL), messages[ {role: system, content: ROUTER_PROMPT}, {role: user, content: question}, ], response_format{type: json_object}, temperature0, ) data json.loads(resp.choices[0].message.content) return data.get(category, hybrid)temperature0很重要。路由决策需要确定性不能这次判成hybrid下次同样的问题判成keyword。另外我还在代码里加了一道规则兜底如果问题里出现“你好”“谢谢”这类词并且长度小于10个字直接走chat不再调用模型。这样寒暄类请求消耗的是零token而不是每次都发一次模型请求。3.2 混合检索BM25关键词召回与向量语义召回的RRF融合路由决定走哪条路后检索层是核心中的核心。先说为什么必须混合。向量模型对“BSC-2000”“E403”这类精确字符串往往不太敏感因为embedding层会把它们映射到语义空间里导致最相似的片段可能不是含这个型号的片段而BM25对自然语言问句的召回率又太低。混合检索的目的就是让两类检索器互补。我的实现里BM25走Elasticsearch向量检索走Qdrant两者各自返回TopK文档的排名然后用RRFReciprocal Rank Fusion把排名融合成一个分数。RRF公式很简单每个文档的分数是它在各通道排名倒数之和k通常取60。def rrf_fuse(ranked_lists: list[list[str]], k: int 60) - dict[str, float]: scores: dict[str, float] {} for ranked in ranked_lists: for rank, doc_id in enumerate(ranked): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return scores def retrieve_hybrid(question: str, top_k: int 20) - list[str]: # 通道1: ES BM25 关键词召回 es_keyword_ids es_bm25_search(question, top_k20) # 通道2: Qdrant 向量语义召回 vector_ids qdrant_vector_search(question, top_k20) # 两路排名融合 fused rrf_fuse([es_keyword_ids, vector_ids]) # 按融合分数降序取 top_k return sorted(fused, keyfused.get, reverseTrue)[:top_k]这里有个细节ES侧的索引必须做好中文分词。默认的standard分词器对中文基本无效会把一句话拆成一堆单字BM25召回质量会非常差。我用的方案是给ES装IK分词插件并在mapping里指定analyzer: ik_max_word。如果只是本地小规模验证也可以用rank_bm25这个Python库在内存里做但生产环境数据量一上来还是ES更省心。向量检索侧的embedding我用的BAAI/bge-m3维度1024中英文混合效果都比较稳。Qdrant的collection创建时直接把vectors的size设成1024distance用Cosine。写入知识库时每篇文档切块后同时写入ES和Qdrant并保留同一个doc_id。融合的时候就是拿doc_id做对齐这一步一定要提前设计好否则两边ID对不上RRF直接失效。3.3 重排、生成与答案验证从“找到相关”到“确认可靠”RRF融合拿到的TopK只能是“候选相关”不能直接当证据。模型生成答案前我加了一个本地重排步骤用交叉编码器对候选片段和问题重新打分。交叉编码器比双塔向量模型更精细它会同时看问题和片段分数更接近“这段到底能不能回答问题”。重排后我保留分数超过0.35的前3到5个片段低于阈值的直接丢弃。这个阈值需要按自己知识库的文档风格调整文档写得越规范阈值可以拉得越高。代码大概是from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-v2-m3, use_fp16True) def rerank(question: str, doc_ids: list[str]) - list[tuple[str, float]]: docs [id_to_text[i] for i in doc_ids] pairs [[question, d] for d in docs] scores reranker.compute_score(pairs) ranked sorted(zip(doc_ids, scores), keylambda x: x[1], reverseTrue) return [(did, float(s)) for did, s in ranked if float(s) 0.35]生成代理的prompt只有三句话的约束只依据给定的检索片段回答如果片段不足以回答问题明确说“资料库暂未覆盖”每个结论后面标注来源片段编号。碎片化的指令比长篇大论更有效。生成完答案后还有一道验证代理这也是很多人忽略的一步。我用轻量模型对“答案”和“检索片段”做一致性打分如果分数低于0.6就拒绝输出返回“当前资料库无法确认这个问题的答案”。这一步直接降低了幻觉率代价只是每次多花几百token。代码如下def verify_answer(question: str, answer: str, contexts: list[str]) - bool: resp client.chat.completions.create( modelos.getenv(LIGHT_MODEL), messages[ {role: system, content: 判断答案是否严格基于给定资料且没有编造。只返回分数0-1。}, {role: user, content: json.dumps({ question: question, answer: answer, contexts: contexts, })}, ], response_format{type: json_object}, temperature0, ) score float(json.loads(resp.choices[0].message.content)[score]) return score 0.6验证代理的prompt里我还会额外要求它quote出答案中对应原文的片段。这样一旦判定不过查看日志时可以直接看到是哪个片段误导了生成排查成本大幅降低。4. 完整工作流编排把零散模块串成可复现的多代理流程4.1 LangGraph状态机路由分支、并行检索与条件边模块都写好后真正让整个系统跑起来的是LangGraph的编排层。我先定义一个AgentState它就是一个普通的TypedDict承载用户问题、路由类型、检索结果、最终答案这些字段。然后按之前的分工注册节点边用条件边实现分支。一段最小可跑的编排代码如下from typing import TypedDict from langgraph.graph import StateGraph, END from langgraph.constants import Send class AgentState(TypedDict): question: str category: str contexts: list[str] answer: str def route_node(state: AgentState) - AgentState: state[category] route_question(state[question]) return {category: state[category]} def retrieve_parallel(state: AgentState) - AgentState: # 根据路由决定检索通道组合 cat state[category] if cat keyword: ids es_bm25_search(state[question], top_k20) elif cat vector: ids qdrant_vector_search(state[question], top_k20) else: ids retrieve_hybrid(state[question], top_k20) return {contexts: rerank(state[question], ids)} def chat_node(state: AgentState) - AgentState: return {answer: 我是企业知识库助手可以问我产品使用、运维、售后等问题。} def generate_node(state: AgentState) - AgentState: answer generate_answer(state[question], state[contexts]) if verify_answer(state[question], answer, state[contexts]): return {answer: answer} return {answer: 当前资料库无法确认该问题的答案请补充更详细的文档后再试。} builder StateGraph(AgentState) builder.add_node(route, route_node) builder.add_node(retrieve, retrieve_parallel) builder.add_node(chat, chat_node) builder.add_node(generate, generate_node) builder.add_edge(route, retrieve) builder.add_edge(route, chat) builder.add_edge(retrieve, generate) builder.add_edge(chat, END) builder.add_edge(generate, END) graph builder.compile()如果只想保留最少分支上面这段就够了。但我在项目里还做了一步并行把“keyword”和“vector”两类检索同时触发而不是由路由二选一。LangGraph里实现并行要用Send接口动态创建多个检索子任务等所有子任务返回后把结果合并到contexts。这一步能把p95延迟从3.1秒压到2.8秒左右代价是实现复杂度略高。4.2 可观测性每个节点的耗时、Token与召回结果都要留下痕迹多代理链路最大的风险是某个节点悄悄“黑盒化”。我见过不少项目线上检索质量下降但没人知道是路由分错了类、ES索引没更新、还是重排模型卡在了本地加载。所以我在每个节点都埋了最小化的日志。代码里不需要引额外的监控平台直接在函数装饰器里打日志就行import time from functools import wraps def log_node(name: str): def deco(fn): wraps(fn) def wrapper(state): start time.time() result fn(state) dt int((time.time() - start) * 1000) print(f[node:{name}] cost{dt}ms state_keys{list(state.keys())}) return result return wrapper return deco实际运行时的日志长这样[node:route] cost312ms state_keys[question, category] [node:retrieve] cost1042ms state_keys[question, category, contexts] [node:generate] cost1388ms state_keys[question, category, contexts, answer]我用这套日志在线上抓出过一个很隐蔽的问题某天开始所有问题都被路由成keyword因为light model所在的网关把temperature0理解成了概率截断导致输出格式不稳定后来在路由调用里加了输出格式校验和一次重试才彻底解决。如果没有日志这种问题大概率会被归类为“最近大模型有点笨”。5. 实测效果与踩坑经验成本、延迟、幻觉和路由误判5.1 三组检索策略的真实对比我把同一份内部文档库分别用三种方案跑了一遍测试集测试集包含240条真实历史问答涵盖产品参数、售后政策、故障排查、寒暄四类。方案准确率上下文命中率平均单轮Tokenp95耗时单路向量检索74%52%18002.1s混合检索无路由83%78%21002.6s多代理混合RAG90%96%20503.1s上下文命中率指的是检索出的片段中至少有一个片段真正包含答案关键信息的比例。单路向量检索这项只有52%意味着将近一半的问题根本没有被检索到正确资料这种情况下强模型再厉害也只能猜。加上路由和混合召回后这个指标到了96%答案准确率自然跟着上去。还有一个反直觉的数据多代理方案虽然多了路由和验证两个环节但平均Token数并没有比无路由的混合检索高多少。原因是路由把大量无关问题分流到了chat节点省掉了检索、重排、生成一整条链路的Token消耗。所以“多加两个代理”不等于“多烧钱”只要分流合理总成本反而可能下降。5.2 我踩过的几个坑及解决方式第一个坑是中文BM25召回质量惨不忍睹。直接用ES默认standard分词器时“保修期内怎么办”会被拆成“保”“修”“期”“内”“怎”“么”“办”检索结果几乎全是噪音。解决的方案是ES里装IK分词插件索引mapping里指定ik_max_word重新创建索引后召回质量立竿见影。如果你不想折腾ES用rank_bm25配合jieba先跑通流程也是可以的但对中文长文档效果会差一些。第二个坑是路由误判。我最初只给路由模型四个分类名没有示例。结果“我们产品怎么保修”被判成chat白白走了闲聊分支。后来在system prompt里每个分类补了两条few-shot示例并且要求模型必须输出JSON误判率才降下来。这类问题很难完全归零但好在这只是路由层后面验证代理还会兜底不影响最终答案正确性。第三个坑是并行节点传参。LangGraph的Send并行创建子任务时直接传递一个大对象会报错比如把整个ES返回结果塞进state节点就会超限。我的做法是先传文档ID列表子任务完成后再回到主流程统一合并。这个细节如果不提前处理并行检索的Demo跑通之后一上线就会炸。第四个坑是重排模型加载慢和并发限流。bge-reranker-v2-m3模型文件接近2GB首次加载可能要30秒我改成懒加载并在并行检索时加了信号量限制同时只有两个检索任务调用重排器。外部大模型API也有可能限流这类平台并发超过3就会有429报错项目里最好统一用一个HTTP客户端做重试。第五个坑是生成截断。一次回答如果被截断在中间用户看到的是一个没有结尾的答案观感非常差。我后来把生成阶段的max_tokens调到1200同时在prompt里写入“回答不超过400字”。这样模型看到指令会尽量精简但token上限又留足了缓冲实测没有再出现截断。6. 源码使用与快速部署零基础怎么在半小时内跑起来6.1 项目目录、依赖与启动顺序虽然现在不能直接贴出完整仓库链接但目录结构可以完整复现。整个项目取名multi_agent_rag核心文件不多普通人打开不会慌multi_agent_rag/ ├── .env.example ├── requirements.txt ├── config.py ├── routers.py ├── retrievers/ │ ├── es_bm25.py │ ├── qdrant_vec.py │ ├── hybrid.py │ └── rerank.py ├── workflow/ │ ├── graph.py │ └── nodes.py ├── app.py ├── build_index.py └── data/ └── 示例知识库.md依赖也收敛得比较干净核心就这些langgraph0.2.44 langchain-openai0.2.6 openai1.40.0 elasticsearch8.14.0 qdrant-client1.12.1 FlagEmbedding1.2.10 jieba0.42.1 python-dotenv1.0.1部署步骤按这个顺序来第一步clone项目并创建.env第二步用docker compose起Elasticsearch和Qdrantdocker compose up -d es qdrant第三步把企业自己的Markdown或PDF扔进data/目录第四步运行python build_index.py完成切块、embedding和索引写入最后python app.py --server启动一个FastAPI服务。整个过程没有依赖前端工程也没有复杂的模型部署唯一需要下载的模型就是bge-reranker。第一次跑通之后建议做三件事把ES索引名改成自己的产品名把路由示例换成自己业务里的真实提问再把验证阈值调到0.6左右观察误杀率。这三件事做完这套系统才算真正从Demo变成可用工具。6.2 扩展方向再接IM机器人、多知识库隔离与热点缓存如果基础版本跑顺了后续的扩展方向我按收益从高到低排个序。第一优先级是热点问题缓存用简单的字典或Redis把“问题归一化后的hash”映射到答案命中就直接返回能省掉大量重复检索成本。第二优先级是多知识库隔离在Qdrant collection或ES索引层面区分不同业务线路由代理返回的category里再增加kb产品线A或kb产品线B的字段避免跨业务串答案。第三优先级才是接企业微信或钉钉机器人给用户一个更自然的交互入口。我个人不那么推荐一上来就做“让Agent自己决定调用什么工具”这种自由模式。多代理混合RAG的价值在于可控一旦Agent可以自由决策可控性就没了。合理的方式是保持这套路由加检索的骨架在生成节点里增加一个“调用业务API”的可选步骤比如查库存、查订单状态这已经是很好的增强。我在内部问答场景里跑了一周之后最大的体会不是“多代理”“混合RAG”这些概念有多酷而是把路由、混合检索、验证这三个动作接对以后系统立刻从“能回答”变成了“敢回答”。如果你现在也准备搭一套我的建议是别一上来就追求复杂的并行检索先用最简单的两路混合加一个路由规则跑通慢慢再补验证代理和重排。最后分享一个小技巧验证代理的prompt里一定要让模型quote出原文片段这样一旦召回质量有问题你可以直接看到是哪个片段误导了生成排查成本会低非常多。
