最近一直在折腾RAG相关的项目,从最早的单Agent裸检索,到后来做成多代理混合检索架构,中间踩的坑足够写满一页纸。今天这篇就把整套系统的设计思路、核心代码和调优经验完整放出来,目标是让完全没接触过Agent编排的人也能照着搭一套能用的多代理混合RAG系统。
先说清楚这套东西是什么:它是一个由路由、检索、重排、生成四个代理角色组成的问答系统,底层同时使用向量检索和关键词检索两条通道召回资料,再经过重排精筛后交给大模型生成答案。跟传统RAG最大的区别在于,它把“检索”这件事拆成了多个可独立调优的环节,每个环节由专门的代理负责决策,而不是一股脑把资料塞给模型。适合谁看?想从单RAG升级的开发者、正在做企业知识库问答的工程师、以及对Agent架构感兴趣但不知道从哪下手的初学者。
1. 为什么要做多代理混合RAG系统
1.1 传统RAG的五个“翻车”场景
我在做第一版知识库问答系统的时候,用的是最常规的套路:文档切块、向量化、存向量库、检索TopK丢给大模型。这套流程在demo阶段跑得挺欢,一旦放到真实数据上就开始原形毕露。
第一个问题是语义相似但不相关。向量检索看的是“意思像不像”,不是“问题能不能被这个片段回答”。比如用户问“离职补偿怎么算”,向量库里跟“离职流程”“补偿标准”相关的片段都会被召回来,哪怕它们根本不含计算规则。TopK取多了答案被噪声带偏,取少了又漏掉关键信息。
第二个问题是专有名词和缩写检索失效。向量模型对“CNAME记录”“TPS接口”“HVAC系统”这类专业缩写非常不敏感,你把它编码成语义向量后,跟全称的相似度往往低得可怜。这时候必须靠关键词精确匹配兜底。
第三个问题是多跳问题拆不开。用户问“A产品的价格调整后,对B方案的预算有什么影响”,传统RAG只会拿整句话去检索,召回的全是A和B各自的独立资料,模型缺乏把两段信息串起来推理的引导,答案经常只覆盖一半。
第四个问题是一问多义不知道怎么路由。比如“如何部署”这个问题,在运维文档和开发文档里有完全不同的答案。单Agent架构只能二选一,召回结果总是“有相关性但不够精确”。
第五个问题是答案没有校验环节。模型用好几个片段拼答案,一旦片段之间逻辑冲突,或者检索结果本身就错了,最后生成的答案就跟着错,而且错得理直气壮。
1.2 多代理与混合检索如何对症下药
针对上面这些问题,我重构了系统,核心思路就三句话:让专门的代理管专门的事,让不同的检索方式互补,让答案经过校验再出口。
路由代理先看用户问的是什么类型的问题——是事实查询、流程指引、对比分析还是闲聊。不同类型的问法对应不同的检索策略和提示词模板,这比一套prompt通吃所有问题要精准得多。
检索代理同时跑两路:一路向量检索抓语义相似的片段,一路关键词检索做精确匹配。两路结果合并去重后,召回质量直接从“单一相似度”提升到“语义+字面双重命中”。
重排代理拿到合并后的候选集,逐条计算与问题的相关性分数,把最靠谱的片段排到前面。这一步不是简单调参,而是让一个专门的模型来判断“这个片段到底能不能回答当前问题”。
生成代理负责最后一步:拿着重排序后的片段生成答案,并且强制要求模型先自检——如果某个片段与问题无关就丢弃,如果片段之间存在矛盾就标记出来。这相当于在出口加了一道质检关卡。
2. 系统架构拆解:四个代理角色如何协作
2.1 路由代理:先判断问题类型再决定检索策略
路由代理是整个系统的入口。它的职责不是检索,而是做“意图判断”。我把它实现成一个轻量分类器,输入用户问题,输出一个结构化决策结果,包含问题类型和检索参数建议。
问题类型我分了四类:事实型(某某是什么、在哪里)、流程型(怎么操作、步骤是什么)、对比型(A和B有什么区别)、综合型(需要多片段推理)。每一类对应不同的检索TopK和提示词模板。比如事实型问题TopK取3就够了,综合型问题我会把TopK拉到8,并且允许跨文档片段。
实现上不一定要单独训练分类模型,直接用带职能描述的LLM调用即可。关键是提示词要写清楚,并且要求模型只输出JSON。我用的决策指令大概是这样的思路:“你是系统路由模块,请判断用户问题所属类型,从给定的类型列表中选一个,不要回答具体内容,只输出JSON格式的type和reason。”
这里有个细节:路由代理的模型不一定要用最强的旗舰模型,用一个速度快的轻量模型就够,因为分类任务本身不复杂。省下来的预算可以留给生成代理。
2.2 检索代理:向量+关键词双通道召回
检索代理是整套系统的核心引擎。我在这层做了两路召回,而不是单纯靠向量检索一路走到底。
第一路:向量检索。文档被切成片段后,用嵌入模型编码存入向量库。查询时把用户问题编码成向量,用余弦相似度召回TopN。这一路负责抓“语义相近”的内容,擅长处理同义异形表达。
第二路:关键词检索。对用户问题做分词和关键词提取,用BM25算法在文档库中做词频检索。这一路负责抓“字面精确匹配”的内容,擅长处理专有名词、缩写、产品型号这类向量模型不敏感的信息。
两路结果合并后,用片段ID做去重,保留每一路的分数。这里有一个重要的经验:不要让某一路的得分被另一路淹没。向量相似度和BM25分数量纲完全不同,直接相加没有意义。我采用的方案是分别做Min-Max归一化,再按权重融合:
def fuse_scores(vector_hits, keyword_hits, alpha=0.6): """ 融合向量检索与关键词检索的结果。 vector_hits: {doc_id: similarity_score} keyword_hits: {doc_id: bm25_score} alpha: 向量得分的权重,关键词得分为 1 - alpha """ merged = {} all_ids = set(vector_hits.keys()) | set(keyword_hits.keys()) vec_min = min(vector_hits.values()) if vector_hits else 0 vec_max = max(vector_hits.values()) if vector_hits else 1 kw_min = min(keyword_hits.values()) if keyword_hits else 0 kw_max = max(keyword_hits.values()) if keyword_hits else 1 for doc_id in all_ids: v_sim = 0.0 k_sim = 0.0 if doc_id in vector_hits: v_sim = (vector_hits[doc_id] - vec_min) / (vec_max - vec_min + 1e-9) if doc_id in keyword_hits: k_sim = (keyword_hits[doc_id] - kw_min) / (kw_max - kw_min + 1e-9) merged[doc_id] = alpha * v_sim + (1 - alpha) * k_sim return sorted(merged.items(), key=lambda x: x[1], reverse=True)融合之后返回TopK候选片段。这里的alpha我一般取0.6到0.7,因为大多数场景语义匹配比字面匹配更可靠,但具体数据需要实测微调。后续会专门讲怎么调这个参数。
2.3 重排代理:精排决定质量上限
很多人忽略重排这一步,觉得反正召回了一批片段,都丢给大模型让它自己挑不就行了。实际上大模型面对5到10个片段时,注意力会被不相关的片段干扰,尤其是排在前面但相关性差的片段,很容易被误当成依据。
重排代理在我的系统里做两件事:逐条打分和相对排序。逐条打分就是针对每个候选片段,让模型或专门的排序模型判断“这段内容能否回答用户问题”,输出一个0到1的相关性分数。相对排序是在打分基础上,按分数重新排列片段顺序,只保留超过阈值的片段。
用LLM做重排的思路比较直接,把每个片段单独拿来问模型:“请判断以下资料能否回答用户问题,只输出分数0到1,不要解释。”这样做的代价是调用次数多,但它有两个好处:一是可以结合问题上下文做判断,而不是单纯依赖向量相似度;二是能识别出“内容相关但无法回答问题”的片段,比如一篇文档只在开头提到了某个关键词,实际内容跟问题无关。
如果对性能要求高,可以考虑使用嵌入模型做交互式编码判断,但我的实测结论是:用LLM重排,虽然慢那么几百毫秒,但最终答案质量提升非常明显,尤其是在候选片段数量多、噪声大的场景下。
2.4 生成代理:校验答案,拒绝胡说
生成代理是最后一道关卡。它拿到重排后的片段和路由代理传过来的问题类型,拼装成最终的提示词,交给大模型生成答案。
我在这里额外加了一个答案校验步骤,这是提升可靠性的关键。提示词里会明确要求模型:回答只基于提供的资料片段;如果资料片段中没有对应信息,直接说“资料中未找到相关内容”,不要编造;如果发现多个片段之间存在矛盾,要显式标注出来。生成之后再做一次简单的规则校验:答案如果包含“不确定”“未找到”这类措辞,就直接返回低置信度标识,前端可以据此展示相关提示,而不是让用户以为系统给出了权威答复。
生成提示词模板大致如下:
def build_generation_prompt(question, question_type, passages): type_instruction = { "fact": "请基于资料给出简明准确的事实性回答,注明信息来源片段。", "process": "请按步骤顺序给出操作流程,引用资料中的步骤说明。", "comparison": "请分别列出A和B的关键区别,并标注依据片段。", "synthesis": "请综合多份资料进行推理,明确指出每部分结论来自哪个片段。", }.get(question_type, "") passages_text = "\n\n".join([ f"[片段{i+1}] {p}" for i, p in enumerate(passages) ]) prompt = f"""你是一个严谨的知识库问答助手。 用户问题:{question} 以下是相关资料片段: {passages_text} {type_instruction} 要求: 1. 只能使用上述片段中的信息回答,不得编造。 2. 如果片段信息不足,直接回复"资料中未找到相关内容"。 3. 如果片段之间信息冲突,请指出冲突点。 4. 在答案末尾列出用到的片段编号。 """ return prompt这个模板看起来简单,但每一项要求都是实践中被坑出来的。尤其“列出用到的片段编号”这一点,对后续排查检索质量特别有用。
3. 零基础搭建实录:核心源码与配置全解析
3.1 工程结构与依赖准备
先把工程结构搭好。我的项目分五个模块,各司其职,避免把所有逻辑塞进一个文件里:
multi_agent_rag/ ├── config.py # 全局配置:模型、路径、参数 ├── document_loader.py # 文档读取与切分 ├── retriever.py # 向量检索 + 关键词检索 ├── agents.py # 路由、重排、生成代理实现 ├── orchestrator.py # 多代理编排主逻辑 └── main.py # 命令行/HTTP调用入口依赖方面不需要装太多东西,核心就四类:一个向量存储工具(用于本地索引管理)、一个嵌入模型SDK、一个大模型SDK、以及基础的数据处理库。向量存储如果数据量在几十万条以内,用本地的轻量方案就够,不需要上分布式。
环境准备阶段我建议直接用虚拟环境隔离,避免跟其他项目的依赖打架。安装完成后,第一件事不是写代码,而是先跑通“文档加载→切分→存入向量库→查询”这四条命令,确认每个环节都能工作,再往上加代理逻辑。
3.2 文档切分与向量索引构建
文档切分是RAG系统里最容易被低估的环节。切得太碎,上下文丢失;切得太粗,片段噪声大。我实测下来的经验是:不同类型文档用不同切分策略,不能一套参数通吃。
对于结构化文档(含标题层级),按标题和段落边界切分,每个二级标题下的内容作为一个片段,保留标题作为元数据。对于纯文本或网页正文,按固定窗口滑动切分,窗口大小设为400到600个字符,重叠率设20%到30%。多出来的重叠部分虽然冗余,但能有效避免关键句被拦腰截断。
切分完成后,把片段ID、内容、元数据(来源文件名、标题、页码)一起存入向量库。索引构建的关键配置有两个:一是嵌入模型的维度要和向量库配置一致;二是建议使用余弦距离作为度量方式,因为嵌入模型训练的比对逻辑通常是余弦相似度。
# document_loader.py 核心逻辑简版 def split_documents(raw_docs, chunk_size=500, overlap=100): chunks = [] for doc in raw_docs: content = doc["content"] # 按段落先粗分,避免跨语义块 paragraphs = content.split("\n\n") buf = "" for para in paragraphs: if len(buf) + len(para) < chunk_size: buf += para + "\n\n" else: if buf: chunks.append(buf.strip()) # 处理超长段落:按窗口滑动切分,保留重叠 if len(para) > chunk_size: start = 0 while start < len(para): chunks.append(para[start:start + chunk_size]) start += chunk_size - overlap buffer = para + "\n\n" if buf: chunks.append(buf.strip()) return chunks这里核心思路就一句话:能按段落切就别用固定窗口,必须先保留语义完整性。
3.3 混合检索与多代理编排主代码
编排层是整个系统的中枢。它按顺序调用各个代理,串联起完整回答流程。我把核心编排逻辑简化成下面这个结构,重点看流程,不是看具体API:
# orchestrator.py class MultiAgentRAGOrchestrator: def __init__(self, retriever, router_agent, rerank_agent, generate_agent): self.retriever = retriever self.router_agent = router_agent self.rerank_agent = rerank_agent self.generate_agent = generate_agent def answer(self, question): # step1: 路由代理判断问题类型与检索参数 route = self.router_agent.route(question) question_type = route["type"] top_k = route.get("top_k", 5) # step2: 检索代理混合召回 vector_hits = self.retriever.vector_search(question, top_k=top_k * 2) keyword_hits = self.retriever.keyword_search(question, top_k=top_k * 2) fused = fuse_scores(vector_hits, keyword_hits, alpha=route.get("alpha", 0.6)) candidates = fused[:top_k * 2] # 留出重排的淘汰空间 # step3: 重排代理精排 reranked = self.rerank_agent.rerank(question, candidates, top_k=top_k) # step4: 生成代理生成答案 + 校验 answer = self.generate_agent.generate(question, question_type, reranked) return { "question": question, "answer": answer["text"], "confidence": answer["confidence"], "source_chunks": reranked, "route_info": route }注意这里的一个细节:检索代理召回的数量是最终需要的两倍。因为重排会淘汰掉一部分噪声片段,如果只召回最终数量,重排就没有筛选空间了。
路由代理返回的参数里还有alpha这个字段。我实际做的时候,不同问题类型确实适合不同的融合权重。比如流程型问题高度依赖精确的操作步骤,关键词权重应该提高;对比型问题语义匹配更重要,向量权重提高。这个动态调整机制比全局固定一个α合理得多。
3.4 调用入口与运行效果演示
主入口文件保持薄薄一层,只做参数解析和结果打印:
# main.py def main(): # 初始化各组件 loader = DocumentLoader() docs = loader.load("./data/documents") chunks = split_documents(docs) retriever = HybridRetriever(chunks) retriever.build_index() router = RouterAgent() reranker = RerankAgent() generator = GenerateAgent() orchestrator = MultiAgentRAGOrchestrator( retriever, router, reranker, generator ) question = input("请输入问题:") result = orchestrator.answer(question) print("\n路由信息:", result["route_info"]) print("\n生成的答案:\n", result["answer"]) print("\n置信度:", result["confidence"])实际跑起来的输出效果类似这样:
请输入问题:HVAC系统在什么温度区间内属于节能运行状态? 路由信息: {"type": "fact", "top_k": 5, "alpha": 0.65} 生成的答案: 根据资料[片段2]与[片段7],HVAC系统在室内温度设定为24℃至26℃、 室外温度不低于-5℃的条件下,处于节能运行区间。 低于此区间时,系统会启动辅助加热,能耗显著上升。 置信度: 0.92注意答案末尾自动带出了片段编号,这就是提示词模板里“列出用到的片段编号”这一要求的效果。一旦发现答案引用了错误片段,我可以直接回溯到对应文档去检查。
4. 关键参数调优与成本控制实战
4.1 检索参数怎么调才合理
检索参数有几个关键变量:切分窗口大小、向量召回数、关键词召回数、融合权重、重排保留数。每个参数的理论作用我都摸了一遍,实测结论如下。
切分窗口大小:我用一组标准企业文档做对照,窗口200字符时,答案偏向“精确但碎片化”,模型经常找不到完整上下文;窗口800字符时,片段噪声明显增大,模型容易引用无关内容。500字符配上20%到30%重叠是相对稳的甜点区间。
召回数量:检索代理按最终需求的两倍召回。也就是说最终给模型5个片段,检索阶段就分别从向量和关键词各召回5到8个,合并去重后送给重排。为什么不是三倍四倍?因为重排代理面对候选数量过多时,排序稳定性和耗时都会变差,两倍是一个性价比平衡点。
融合权重动Alpha:在0.6到0.7附近先跑一批问题,把回答得不好的case拎出来看:如果错误答案明显来自关键词命中的噪声片段,就把alpha调高;如果是因为专有名词漏召回导致回答不全,就把alpha调低。这个过程至少要迭代三轮。
重排保留数:我固定为最终答案片段数的1.2到1.5倍。原因是生成代理需要一点冗余来对照信息一致性,但太多冗余又会让提示词过长、模型注意力分散。
4.2 代理并发与超时设置
多代理系统串行调用时,延迟是叠加的。路由一次调用加重排每片段一次调用再加生成一次调用,如果全部串行,一次问答的模型调用次数可能高达6到10次。在并发低的内部系统中能接受,如果要做成对外服务,就必须考虑并发和超时。
我做的优化方案是:路由代理和检索并行。因为路由决策只需要用户问题本身,不依赖检索结果;检索也不需要知道问题类型,两路可以同时跑。等两者都完成后,再拿路由结果去校准检索配置(比如不同的TopK策略可以在检索结果上做过滤,而不需要重新查库)。
重排这一步天然需要串行,因为候选片段是分批打分并排序。我在这里做了一次批处理优化:把多个片段拼进一个Prompt里让模型按列表打分,而不是一个片段调一次。实测打分质量略有下降(模型容易受先后顺序影响),但调用次数从N次降到1次,速度提升显著。如果对质量要求极高,可以保留逐条打分模式。
超时设置方面,路由代理给5秒,重排每个轮次给8秒,生成给30秒。超时后直接fallback到默认参数或提示用户稍后重试,不能让请求悬死在那里。
4.3 成本优化三板斧
多代理系统的成本比单RAG高,这是事实,但优化空间也大。我的经验是三个方向。
第一板斧:用轻量模型处理简单任务。路由分类、关键词提取、片段相关性初筛,这些任务都用小模型,只有最终生成答案才用旗舰模型。实测下来,准确率损失在可接受范围内,成本却降了六成以上。
第二板斧:缓存高重复查询。大多数知识库问答场景中,用户问的问题存在大量重复。我直接在编排层做了一层语义缓存,命中后直接返回历史答案,连路由都不跑。缓存key是规范化后的问题文本,配合一个相似度阈值判断近似重复。
第三板斧:控制重排规模。重排是调用次数的大头。候选片段从10个降到6个,重排调用就少40%。怎么降到6个不损失质量?在混合召回阶段就做一个粗筛:融合分数低于最大分数30%的片段直接丢弃,不进入重排。这个粗筛规则对大多数场景都适用。
5. 线上踩坑实录:常见问题与排查方法
5.1 检索效果差,召回乱七八糟
表现:答案引用的片段明显跟问题没关系,或者回顾关键信息。
排查路径我一般按三步走。第一步看路由信息,确认问题类型判断有没有错。路由把流程型问题判成事实型,检索策略就会跑偏。第二步看融合分数,把Top10片段的向量分和关键词分单独打印出来,如果某个片段两路分数都低却被选中了,多半是归一化或权重有问题。第三步看原始文档,有时候不是检索的错,是源文档本身就是一团乱麻,一个片段里塞了多个不相关话题,这时候需要回头改切分策略。
我遇到过最典型的一次:用户问“支付回调地址怎么配置”,向量检索召回了“回调”“地址”相关的片段但都是开发环境配置的内容,关键词检索召回了“配置”相关的操作手册但没覆盖支付通道。融合后正确答案排到第六位,被重排干掉了。最后发现是切分窗口太大,把支付回调的配置步骤跟通道列表混在了一个片段里。把切分窗口缩小并增加重叠后,问题迎刃而解。
5.2 多代理间配合失灵
表现:路由说类型A,检索却按类型B的参数跑;或者重排结果分数普遍偏低,生成代理拿不到足够多的高质量片段。
这类问题九成出在接口约定不一致上。路由代理输出的是JSON,检索代理读的却可能是另一个字段名;重排代理期望输入列表结构,检索代理传过去的是字典。多代理系统的第一课,就是要先定好每个代理之间的数据Schema,然后让每个代理的输入输出都严格符合这个Schema。
我的做法是,在每个代理函数入口加一段断言式检查,字段缺失就直接抛异常而不是静默处理。这样问题一出现就能定位,而不是等到最终答案质量变差才去猜哪里出了问题。另外,代理之间的Prompt里如果要传递结构化数据,要求模型只输出JSON,并且在外面套一层解析容错逻辑。
5.3 响应慢、卡顿与内存问题
多代理系统最常见的性能瓶颈是重排阶段的逐条模型调用。如果你发现一次问答要耗时十几秒,先别急着上缓存或换GPU,直接把重排从逐条调用改成批处理,通常立竿见影。
内存方面,最大的坑是文档切分后所有片段都常驻内存。几十万片段的文本和向量,在小内存机器上很容易把进程压垮。解决方案是把向量索引落盘,检索时只加载索引文件;片段原文存在SQLite或直接读源文件,只有真正需要送往重排的片段才加载进内存。
5.4 专项避坑速查表
| 症状 | 可能原因 | 快速解法 |
|---|---|---|
| 答案引用错误片段 | 切分窗口过大混入噪声 | 缩小窗口并增加重叠 |
| 专有名词查不到 | 关键词召回权重过低 | 调低alpha,提高BM25权重 |
| 不同问题类型答案风格混乱 | 路由判断不稳定 | 给路由Prompt加few-shot示例 |
| 重排分数全部偏低 | 提示词太严苛 | 放宽打分标准,增加“勉强相关”档位 |
| 答案缺少信息来源 | 生成提示词没要求引用片段 | 强制输出片段编号 |
| 并发一高就超时 | 重排串行调用过多 | 批处理重排,加粗筛前缀 |
6. 一些最后想说的话
整套系统从零搭起来,核心代码量其实不大,难的是想清楚每一个环节到底要承担什么职责,以及数据在不同代理之间怎么流转。我个人的体会是,多代理架构的价值不在于“用了Agent所以更高级”,而在于它把RAG链路中互相纠缠的问题拆解成了可独立优化的单元——检索调不好就只动检索,路由不准就只动路由,不用像以前那样一改就牵动全身。
最后再分享一个小技巧:上线前准备一组固定的验收问题集,每个问题标注了预期答案和预期片段。每次改完任何参数,先跑一遍验收集,对比答案质量和引用的片段编号有没有变差。这套回归测试机制省了我太多次线上翻车,强烈建议你从一开始就建立起来。后续想扩展的话,可以在重排和生成之间插入一层记忆模块,让系统记住用户之前问过什么,实现多轮对话中的上下文衔接,这就是另一个话题了。