Agentic RAG实战：多跳问答、决策循环与MCP工具集成

简介一套原创智能体问答系统案例包面向人工智能应用开发者聚焦检索增强生成与智能体协同的问答场景可用于客户服务、在线教育、技术支持等业务。压缩包共二十九个文件其中源码十三个涵盖文档处理、向量检索、文本转数据库查询、工具调用等模块另有编译文件、配置、向量索引及数据库文件整体仅七百零五KB便于快速部署与二次开发。目前已有一百零四人学习下载。资源不仅提供完整程序代码还通过模块化设计展示数据预处理、模型训练、问答流程与结果评估的实现思路尤其适合想了解检索增强生成与模型上下文协议工具集成、多智能体协作的开发者参考。借助内置索引和轻量数据使用者可快速跑通智能问答流程并结合实际文档进行功能扩展。1. Agentic RAG 智能问答系统为什么「检索一次就生成」的 RAG 不够用Agentic RAG 这两年讨论度很高但真正把整套流程跑通的人不多。传统 RAG 的链路是死的——query 进来向量检索一次把 top-k 片段塞进 prompt让模型直接生成。单跳问题上这套流程够用可一旦遇到「先查 A、再拿 A 的结果去查 B」的多跳问题检索一次几乎必然漏更麻烦的是模型没有自我纠错能力检索结果不对它也硬着头皮答。我拆过的 RAG 项目里最终能落地的方案无一例外都长成了 Agent模型自己决定要不要检索、检索什么、结果够不够不够就改写关键词再查。这个压缩包是一套完整的原创案例向量库、混合检索、Agent 决策循环、MCP 工具层都齐了适合正在做知识库问答、想把固定 RAG 链路升级成可推理问答系统的人。2. Agentic RAG 的架构演进传统 RAG 在哪一环断掉Agent 就在哪一环补上2.1 传统 RAG 的固定链路缺陷先明确一个前提RAG 和 Agentic RAG 不是替代关系而是进化关系。传统 RAG 把「检索—增强—生成」当成一条固定流水线query 进来以后每一步都是预设好的。这个设计在简单文档问答里很稳但它有三个结构性缺陷。第一单次检索的召回上限很低。Embedding 匹配对措辞敏感「退货率最高的 SKU」和「哪个单品退得最多」在向量空间里距离可能很远一次检索根本召不回正确片段。第二top-k 是死参数。固定取 4 条或 5 条信息不够时不会自动多查信息冗余时也不会自动收敛。第三也是最致命的——链路没有反馈。生成阶段发现上下文矛盾或缺失整个流程不会回头补救模型只能硬答幻觉就是这么来的。这三个缺陷叠加正好解释了为什么同一个知识库传统 RAG 在「某条款是什么」这类单跳问题上表现不错一到「对比两个条款的差异」「按条件汇总多篇文档」就明显翻车。Agentic RAG 的改法不是换一个更聪明的检索器而是把「检索」从流水线上的一步变成 Agent 可以反复调用、评估、再调用的工具。下面这张对比表把差异列清楚维度传统 RAGAgentic RAG检索次数固定 1 次按需多次可改写查询重查纠错能力无检索错就答错有评估不足后重新检索多跳问题难以处理拆解子问题逐步求解可解释性只能看到 top-k 片段能看到决策轨迹和工具调用成本与延迟低高多轮调用 LLM 与工具这张表也是我判断一个 RAG 项目要不要升级成 Agentic 的取舍依据单跳问答、对延迟敏感、token 预算紧的场景传统 RAG 反而更合适一旦出现多跳、对比、需要工具配合的需求就值得上 Agent。压缩包里的案例属于后者它的检索、决策、工具三层是分开实现的下面逐一拆。2.2 决策循环检索、评估、再检索Agentic RAG 的核心不是某个更聪明的模型而是一个带反馈的循环。典型结构是三段式模型先评估当前上下文够不够回答不够就生成一个检索意图并执行拿到结果后重新评估直到信息充分或到达轮次上限。这个循环可以用一段很短的核心代码表达真实项目就是把这段逻辑配上 prompt 和工具层# agentic_loop.py —— 最小 Agentic RAG 决策循环 def ask(question: str, max_rounds: int 3) - str: context [] for round_idx in range(max_rounds): # 模型评估现有信息输出 SEARCH:改写查询 或 ANSWER:最终答案 decision llm.decide( questionquestion, contextcontext, system信息不足时返回 SEARCH: 加改写后的查询信息充足时返回 ANSWER ) if decision.startswith(SEARCH:): query decision.removeprefix(SEARCH:).strip() docs vectorstore.search(query, top_k5) context.append(docs) continue # 进入下一轮重新评估 if decision.startswith(ANSWER:): return decision.removeprefix(ANSWER:).strip() # 超出轮次上限的兜底用现有上下文强制生成 return llm.generate(question, context)逻辑说明这段代码把「检索」从一次性操作变成了循环里的动作。decide是同一个 LLM 在扮演两个角色——先是评估员判断信息够不够再是规划员产出改写后的查询。context累积每一轮检索到的片段让后续决策能看到「已经查过什么」避免同一轮里原地打转。max_rounds是整个循环的安全阀真实场景里我一般设 3超过就强制生成防止 Agent 在检索上无限绕圈。参数说明top_k5是每轮检索的召回数量知识库文档碎片化程度高就调到 8max_rounds直接决定最坏情况下的 token 消耗每多一轮就多一次 LLM 调用和一次向量检索成本敏感的场景压到 2~3。这段代码是示意压缩包里的 agent 模块在此基础上加了工具调用和 MCP 接入下一章展开。这里顺带回答一个很多人纠结的问题LangChain、Dify、CrewAI 这些框架到底选哪个。我理解框架本质上只是 harness编排壳负责把决策循环、工具注册、上下文组装这些骨架搭好而 Agent 的智能来自模型加提示词加工具设计。框架选型最该看的是工具生态和社区维护而不是哪个功能列表更长。这套案例直接用 LangChain 的 ReAct 模板加 MCP adapter因为它对工具调用的支持最成熟排查问题时的资料也最多。另外提醒一点压缩包里的问答入口是单轮设计。如果你要做多轮对话context里还要额外维护对话记忆否则上一轮检索出来的片段会污染下一轮的决策这是 Agentic RAG 做聊天机器人最常见的翻车点。2.3 工具层为什么选 MCP决策循环里有一个关键的工程问题Agent 要「调用检索」这个能力以什么形式暴露常见做法有两种一种是直接 import 检索函数注册成某框架的 tool另一种就是 MCPModel Context Protocol。这套案例选的是 MCP原因是它把知识库检索定义成了标准工具协议而不是某个框架私有的函数签名。MCP 的角色分三块Host 是持有 Agent 的应用程序Client 负责建立连接、发现工具、发起调用Server 负责实现具体工具。知识库检索就是一个典型的 MCP Server它对外暴露search_kb(query)工具并附上 JSON Schema 描述参数。Agent 运行时通过 Client 发现工具列表按需调用。MCP 角色职责案例中的对应Host运行 Agent 的应用程序main.py agent 模块Client建立连接、发现工具、调用工具mcp SDK 的 ClientSessionServer实现具体工具逻辑mcp_servers/kb_server.py选 MCP 的直接收益是工具可复用同一个知识库 Server既能接自建的 Agent也能接支持 MCP 的桌面端应用不需要为每个 Host 重写检索代码。另外工具和 Agent 进程解耦检索服务的升级、重启都不影响问答入口。代价是链路变长排查问题多一层——工具超时、Schema 不匹配这类问题第四章会具体写到。如果只是几十条数据的 demo直接 import 函数更省事知识库、Agent、前端是多人协作的工程时MCP 的隔离价值就体现出来了。3. 把压缩包跑起来环境配置、混合检索与 MCP 编排代码拆解这一章按「跑起来→看懂检索→看懂编排」的顺序拆。先说环境再说检索最后说 Agent 怎么和 MCP 工具对接每段代码都能对应压缩包里的同名模块。3.1 项目结构与启动流程解压之后先别急着装依赖把目录结构看一遍。这类 Agent 案例包通常把「检索」和「Agent」分成两个模块中间用 MCP 衔接解压后大致长这样Agentic_RAG_Agent/ ├── main.py # 问答入口接收用户问题 ├── config.yaml # 模型、向量库、top_k、max_rounds 等参数 ├── retriever/ # 向量检索 BM25 混合检索 重排序 ├── agent/ # 决策循环、prompt 模板、工具绑定 ├── mcp_servers/ # 知识库 MCP Serverstdio 传输 ├── data/vector_db/ # 预构建的向量库 └── requirements.txt启动流程分两步因为 MCP Server 是一个独立进程需要先把它拉起来再做问答cd Agentic_RAG_Agent python -m venv .venv source .venv/bin/activate pip install -r requirements.txt # 终端一启动知识库 MCP Server python mcp_servers/kb_server.py # 终端二启动问答入口 python main.py --question 上季度华东区退货率最高的SKU是什么逻辑说明main.py是用户入口只负责解析参数、调用 agent 模块真正的检索能力在mcp_servers/kb_server.py里通过标准输入输出stdio和 Agent 进程通信。Server 独立成进程的好处是检索服务的生命周期和问答入口分离后续升级向量库或重排序逻辑时不用动 Agent 代码。运行成功后终端会先打印 MCP Server 的握手日志再打印 Agent 的决策轨迹最后输出答案。如果直接看到答案但没有轨迹检查config.yaml里的 trace 开关是否打开。关键参数集中在config.yamlmodel: name: qwen-plus # 换成你有 API 权限的模型名 temperature: 0.1 retriever: top_k: 5 alpha: 0.7 agent: max_rounds: 3 max_iterations: 6 vector_store: persist_directory: ./data/vector_db embedding: BAAI/bge-m3参数说明temperature建议压到 0.1 左右Agent 的决策输出要稳定温度太高会让模型在「检索还是回答」之间摇摆。max_rounds和max_iterations是两道安全阀分别限制决策循环和 ReAct 步数。所有可变参数收敛到 yaml换模型、调检索深度时只改一个文件。requirements.txt 里锁的核心依赖是 langchain、langchain-mcp-adapters、mcp、chromadb、sentence-transformers、rank-bm25安装时注意 langchain 和 langchain-mcp-adapters 的版本要配套新版 langchain 改过 tool 接口adapter 太旧会直接报ImportError。3.2 检索器与重排序的实现Agentic RAG 对检索质量的要求比传统 RAG 高因为 Agent 的每一步决策都依赖检索结果第一轮召回的片段质量差后面所有轮次都会在错误上下文上叠加。案例里用的是混合检索加权重融合# retriever/hybrid.py —— 混合检索与权重融合 from langchain_huggingface import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma embeddings HuggingFaceEmbeddings(model_nameBAAI/bge-m3) vectorstore Chroma(persist_directory./data/vector_db, embedding_functionembeddings) def hybrid_retrieve(query: str, top_k: int 5, alpha: float 0.7): # 向量检索语义匹配覆盖措辞不同但意思相近的查询 vec_docs vectorstore.similarity_search_with_score(query, ktop_k) # BM25 关键词检索覆盖型号、编号、专有名词等精确匹配场景 kw_docs bm25_search(query, top_k) # alpha 控制语义与关键词权重返回融合后分数最高的 top_k 条 merged merge_by_score(vec_docs, kw_docs, alphaalpha) return merged[:top_k]逻辑说明embedding 用BAAI/bge-m3多语言效果稳定中文知识库用它比直接用英文为主的 embedding 少一层分词困扰。向量库用 Chroma本地文件持久化persist_directory指向预构建的data/vector_db启动时直接加载不需要重新灌库。similarity_search_with_score返回距离分数融合公式为alpha * vec_score (1 - alpha) * kw_score再按融合分数倒序取前top_k。bm25_search是压缩包里retriever/bm25.py的封装启动时对全量文档做一次分词建索引查询时按关键词命中打分。参数说明alpha0.7表示默认更信任语义匹配因为用户问题通常不是精确术语但知识库里大量存在 SKU、条款编号这类 token纯向量检索经常漏所以保留alpha在 0.6~0.8 之间调。这里补一个经验如果你的知识库是产品手册、合同条款这类专有名词密度高的文档把alpha压到 0.5 左右关键词检索的权重需要接近一半。参数默认值调整建议top_k5文档碎片化严重时调到 8alpha0.7专有名词多时压到 0.5score_threshold无低分片段过滤避免噪声进上下文注意bm25_search依赖启动时对全量文档分词建索引文档量大的场景建议用 jieba 分词后缓存索引否则每次启动都要重建几万条文档可能要等十几秒。3.3 Agent 编排与 MCP 工具接入检索器就位后剩下的就是把hybrid_retrieve暴露成 MCP 工具再让 Agent 在决策循环里调用它。这一步是这套案例和普通 RAG 最大的区别# agent/qa_agent.py —— MCP 工具加载与 ReAct Agent 编排 import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools from langchain.agents import create_react_agent, AgentExecutor async def build_agent(): # 指定 MCP Server 的启动方式本地进程 stdio 传输 server_params StdioServerParameters( commandpython, args[mcp_servers/kb_server.py], env{MCP_MODE: stdio}, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 完成 MCP 握手 tools await load_mcp_tools(session) # 拉取工具列表 agent create_react_agent(llmllm, toolstools, promptREACT_PROMPT) executor AgentExecutor(agentagent, toolstools, max_iterations6) result await executor.ainvoke({input: user_question}) return result[output]逻辑说明stdio_client拉起kb_server.py子进程并建立双向管道ClientSession.initialize()完成 MCP 协议握手load_mcp_tools把 Server 端声明的search_kb工具自动转换成 LangChain 的 tool 对象。Agent 必须跑在异步上下文里因为 MCP 的 stdio 传输本身就是异步事件循环驱动入口处用asyncio.run(build_agent())收尾。参数说明max_iterations6是 LangChain 层的安全阀和决策循环的max_rounds是两道防线——前者限制 ReAct 循环总步数后者限制每轮查询次数两层都设好才能避免无限循环。REACT_PROMPT里我明确写了一条规则只有确认上下文无法回答时才允许调用search_kb这个约束能省下大量无效检索。如果你换用别的模型记得检查它的 tool-call 格式LangChain 会自动适配大部分主流模型但个别本地模型需要单独配 tool schema否则工具调用永远解析失败。4. 避坑与排查Agentic RAG 落地中的五个实际问题这一章的内容全是实际跑包时踩过的血泪经验。Agentic RAG 比传统 RAG 多了一个决策层所有问题几乎都集中在这一层检索结果不可信、Agent 行为不可控、工具链路不稳定。下面按检索层、决策层、工程层三条线写五个坑。4.1 检索层命中但跑题专有名词查不到第一个坑检索命中但回答跑题。现象search_kb返回的片段里确实包含关键词但 Agent 生成的回答引用了错误段落答非所问看起来像幻觉。日志里能看到检索到的片段和最终答案引用的片段对不上。原因top-k 里混进了语义相近但主题不同的噪声片段。比如问题问「售后流程」检索到「售后退款」和「售后政策」两篇文档Agent 把两者混在一起作答。更深一层的原因是决策循环只评估了「检索到了什么」没有校验「检索结果是否直接支撑回答」。解决在检索后加一道重排序。我一般用 cross-encoder 对 top-k10 的候选重新打分再取前 5噪声片段基本被压到最末尾。排查时打开 trace 日志对比检索到的 doc_id 和答案引用的 doc_id不一致就是这里的问题。如果不想引入额外模型至少把alpha调到 0.8同时检查config.yaml里是否有score_threshold参数过滤掉低分片段再进上下文。第二个坑型号、编号这类专有名词检索不到。现象问「SKU-2024-088 的库存阈值是多少」检索返回空Agent 直接说知识库没有这个信息。原因Embedding 对随机字符串不敏感「2024-088」在向量空间里几乎没有区分度纯向量检索必然漏。解决这是案例里保留 BM25 混合检索的直接原因。alpha0.7意味着有 30% 权重落在关键词匹配上SKU 这类 token 靠 BM25 才能命中。如果你接手的是专有名词密度高的知识库把alpha压到 0.5让关键词检索和语义检索各占一半权重。排查时先看 bm25 日志里有没有命中如果 BM25 也没命中问题就出在分词需要给 jieba 用户词典补充这类术语。4.2 决策层反复检索烧 token输出格式不稳定第三个坑Agent 反复检索一次问答烧掉上万 token。现象决策日志里SEARCH:出现 6 次以上每轮都把 top-k5 的片段全文塞进上下文最终答案只有两句话成本接近传统 RAG 的十倍。原因提示词没有约束检索条件Agent 拿到任何问题都先检索一轮甚至改写后重复检索同一个意图加上context不清空每轮累积的文档越来越多后面几轮 prompt 越来越长。用日志统计 token 时会发现最贵的不是检索本身而是把检索结果塞回 prompt 后的二次编码。解决两条线同时堵。第一决策 prompt 里明确写「若上下文已有足够信息禁止再次检索」第二把max_rounds从 3 压到 2 试一次如果回答质量没明显下降说明检索轮次本来就是虚高的。另外我给决策循环加了查询缓存30 分钟内相同改写的查询直接复用检索结果能砍掉约三成重复检索。第四个坑模型不遵循输出协议SEARCH:前缀偶尔缺失。现象决策返回一句自然语言「我需要查询一下」代码里startswith(SEARCH:)匹配不上直接走到兜底生成回答质量断崖式下跌。原因LLM 输出格式不稳定换模型之后前缀、标点、大小写都可能变。所有靠前缀解析的 Agent 都躲不开这个坑它不是玄学是概率问题。解决别只用startswith做硬解析用正则兼容多种写法更稳妥的做法是把决策结果约束成结构化 JSONimport json, re def parse_decision(raw: str) - dict: # 兼容带 json 代码块和裸 json 两种输出 m re.search(r\{.*\}, raw, re.S) return json.loads(m.group(0)) # 示例输出{action: SEARCH, query: 华东区退货率最高的SKU} decision parse_decision(llm.decide(question, context))逻辑说明正则取出第一个 JSON 对象action字段决定走检索还是回答query字段带改写后的查询。这样解析的鲁棒性比字符串前缀高很多换模型时基本不用改代码。注意re.S标志必须带上否则 JSON 里的换行会截断匹配。如果模型经常输出无效 JSON可以在 decide 的 system 里给一个 few-shot 示例格式错误率能再降一截。4.3 工程层MCP 工具调用超时与冷启动第五个坑第一次提问总是超时第二次就正常。现象main.py启动后第一次调用search_kb报工具调用超时重试一次就能过。这个问题在本地开发时不容易发现一旦部署成服务用户第一次点击就中招。原因MCP Server 进程是惰性初始化的。stdio_client拉起子进程后ClientSession.initialize()只完成协议握手embedding 模型和向量库的实际加载发生在第一次真正调用工具时。bge-m3 加载和 Chroma 打开本地库都要几秒刚好撞上默认超时。解决给 Agent 入口加一次预热调用。在build_agent()拿到 tools 之后立刻执行一次executor.ainvoke({input: __warmup__})让 Server 把模型和向量库全部加载完成再进入真实问答。如果生产环境对首次延迟敏感就把 MCP Server 常驻Agent 侧只建连接不拉起进程避免每次问答都重新冷启动。提示MCP Server 首次启动加载 embedding 模型会比后续慢 5~10 秒这是正常现象不是代码问题。用预热调用能把这个延迟从用户路径上抹掉。排查时先看 MCP Server 的 stderr 输出模型加载报错比如显存不足、权重文件损坏都会打在这里Agent 侧只会看到一个超时。我的排查顺序固定是先看检索日志再看决策日志最后看 MCP Server stderr。三层日志对不上就沿着 trace 一层层找大多数问题在第一步就能定位。坑快速定位手段检索命中但跑题对比检索 doc_id 与答案引用 doc_id专有名词漏检看 BM25 日志是否命中反复检索烧 token数决策日志里 SEARCH 次数前缀解析失败打印 decision 原文看格式首次调用超时预热调用后重测5. 验证 Agent 行为与生产化先做对这三件事压缩包跑通只是第一步真正判断这套 Agentic RAG 能不能用我每次都会先做三件事。第一件建评测集。至少 30 条覆盖三类问题单跳直答、多跳推理、知识库无答案。无答案类最容易暴露问题——Agent 对查不到的信息应该明确说「不知道」而不是拿无关片段硬凑。评测时把问题、检索轨迹、答案对齐成一行翻车时直接能看到是哪一轮决策出了问题。第二件把 Agent 的决策轨迹落成日志。每次问答输出一个 jsonl记录每一轮的 action、query、检索到的片段 ID、token 数{round: 1, action: SEARCH, query: 华东区退货率最高的SKU, doc_ids: [doc_023, doc_041], tokens: 812}这套案例的 agent 模块预留了 trace 钩子把输出重定向到文件即可。没有轨迹日志的 Agentic RAG 在线上就是黑匣子回答错了根本没法定位是检索问题还是决策问题。第三件做缓存和并发控制。检索结果按改写后的 query 做 10 分钟缓存能省掉大量重复检索同时给 Agent 入口加信号量限流控制同时运行的问答任务数。Agentic RAG 的 token 消耗是传统 RAG 的好几倍并发一上来如果不加控制账单先崩。我之前在一个内部工具上就是忘了加缓存几百个用户涌进来一天烧掉的 token 比平时一周还多。从那以后我每次上线 Agent 类项目都强制走这三步——评测集、轨迹日志、缓存限流——再开放给用户。希望这份拆解帮得到你。本文还有配套的精品资源点击获取