简介这是一份面向AI应用开发者、算法工程师及RAG技术学习者的Agentic RAG可运行源码包配套解析代理式检索增强生成的核心原理涵盖动态选择多信息源、克服单一知识源与一次性检索两大缺陷、单代理与多代理系统架构以及基于函数调用语言模型和DSPy、LangChain等代理框架的两种实施路线可帮助读者从原理到实现完整理解这一前沿技术。压缩包共3个文件以HTML演示页面、inscode在线运行配置和gitignore辅助文件组成整体仅8KB轻量紧凑便于快速部署和对照学习。当前已有139人学习下载适合用于技术预研、课程设计或企业智能化检索方案选型。通过这份源码读者可以直观查看Agentic RAG的代码组织方式结合说明性网页梳理工作原理与适用场景掌握从传统RAG向Agentic RAG升级的关键步骤并将其扩展至企业知识库、智能问答、多源信息整合等实际业务中。虽然包体不大但源码与说明相互配合提供了从概念到落地的完整示例方便在本地或在线平台直接运行体验。1. Agentic RAG到底是什么我最早接触Agentic RAG这词儿是在企业知识库做问答优化的时候。当时客户提了一个需求能不能让AI在回答问题时自己判断该查数据库还是查文档查完发现信息不够还得能换个渠道再查最后把结果汇总成报告。传统RAG做这事儿非常吃力——因为传统RAG的流程是固定的用户输入向量召回拼上下文大模型生成。你要是想让它在中间多查两步、查完不满意重新查得靠开发者在代码里写死各种if-else逻辑稍微复杂一点就失控。后来看到LangChain和LangGraph社区在提Agentic RAG英文资料一堆中文的落地讲解却少得可怜尤其带源码的更少。我花了几周时间开发了一个最小可运行版本跑通一些典型场景之后想借这篇内容把整个思路、代码、踩坑过程都拆开讲清楚。本文不会只停留在概念层面而是直接给一个能跑的源码结构你拿到手改改就能用。先做一个简单的类比传统RAG像地铁站的自助售票机——你投币、选站、取票流程固定Agentic RAG像人工售票窗口——你告诉售票员我要去一个最近能看展的地方售票员会问你想看什么展、预算多少、能接受多远然后综合判断给你方案甚至你会补充一句话他还能当场修正推荐结果。这个会问、会判断、会修正的循环就是Agentic RAG的核心特征。从工程定义上讲Agentic RAG不是某个框架的名字而是一种架构模式它将大语言模型的推理能力、外部工具向量检索、API调用、数据库查询、网页搜索以及自主决策的循环组装成一个能按需规划、按步骤执行并可以中途纠错的问答系统。传统RAG的核心流程是检索-生成Agentic RAG的核心则是一个规划-执行-验证的回路检索只是其中一环。2. 从被动检索到主动决策传统RAG到Agentic RAG的三个关键跃迁接触过Agentic RAG的人容易陷入一种误区以为它只是在普通RAG前面套一层智能路由——判断用户问题要不要检索要检索就查不要就直接回答。这没错但只是最表层的东西。真正拉开差距的是下面三个能力的跃迁。我把它们拆开讲每个都从为什么原来做不到开始。2.1 路由能力第一跳转向传统RAG里一个问题进来后系统默认执行检索。但实际业务中用户可能只是在闲聊或者用户问的是总结性问题并需要基于已提供的知识作答。如果硬做检索不仅浪费还会把不相关信息塞进上下文拉低回答质量。Agentic RAG的第一步是让模型判断这个问题是否需要走检索以及如果走检索应该走哪个路径。路由能力在我的源码里由一个router_agent实现它本质上是一段给模型的指令约束要求模型输出一个JSON结构指明intentchat、retrieve、summarize和query需要改写后的检索词。代码层面很简单但设计上要注意路由不是简单的关键词匹配而是要把路由决策交给LLM依靠它对语义的理解来判断。比如今天天气怎么样这种问题路由会判断不需要走内部知识库检索而应该走天气API工具再比如What is Agentic RAG就走文档检索。这一块对回答质量的影响是决定性的。我见过很多团队在调RAG的召回率、rerank模型效果始终上不去最后发现根因是很多问题根本不应该走向量检索。路由先把方向搞对后续才谈得上效果。2.2 多步工具调用跨文档推理的关键传统RAG的另一个缺陷是一次性检索一次性生成。你问对比一下A和B两篇论文在注意力机制上的不同传统RAG只能把所有相关内容一次性拼进去然后让模型自己总结。如果A的内容在某份PDF里B的内容在某个数据库表里则传统RAG基本无能为力因为它的检索器只能挂一种来源。Agentic RAG的第二个关键跃迁是支持多工具之间的编排。系统会先调文档检索工具找到A论文的相关段落再调数据库工具找B论文的关键参数然后把两段结果合并交给LLM做对比。这个过程在我源码里通过tools列表调度每个工具都是独立的可调用单元Agent根据任务需要决定调用顺序。我实现了一个特别典型的工具组knowledge_base_search本地向量检索、web_search模拟网页搜索、database_query模拟结构化数据查询。Agent可以自己决定先用哪个、再用哪个。这个过程有点像真人查资料你手上有一堆来源先翻书不够再上网再不行就查数据库最后汇总一个答案。这种多步工具调用的背后是LLM具备了计划分解的能力不过它并不是一次规划好所有步骤而是走一步看一步、根据上一步的执行结果动态调整下一步这也是Agentic RAG和单纯加长Prompt之间的本质区别。2.3 自纠正能力从不可靠回归可靠大模型生成内容是有概率性的第一次检索结果经常不够好传统RAG只能认命。Agentic RAG新增了一个至关重要但又容易被忽略的能力——自纠正。这个机制在我的源码里这样实现检索到结果之后不是直接丢给LLM而是先做一个grade_documents的评估对召回的每段内容打分判断它与用户查询的相关程度。如果发现没有任何一条内容的相关度超过阈值Agent会自主发起一次查询改写rewrite调整检索词后重新走检索流程最多重试两轮。如果在两轮之内找到了合格内容就继续生成答案如果两轮之后仍然没有则明确告知用户当前知识库中未找到足够信息而不是硬编一个答案。这个机制解决了我之前做RAG时最容易挨骂的一个问题——大模型一本正经地胡说八道。传统RAG在知识库没有答案时会用看似相关的信息拼凑一个回答用户根本无法分辨真假。Agentic RAG通过自纠正和评分门槛把不知道变成了一种显式的状态。为了更清楚地展示差异我列一个对比表维度传统RAGAgentic RAG检索流程固定单次检索按需多轮检索可自动重试问题理解拿原问题直接查embedding先路由判断意图再改写查询词工具来源通常只挂一个向量库可同时编排多个工具KB、API、DB失败处理无硬生成评分不达标会自动重写、重试或拒答上下文组装一次性拼入TopK按步骤收集动态裁剪与验证对开发者的要求提前写好固定流程设计Agent指令、工具和状态机这个对比一目了然Agentic RAG不是把RAG推翻重来而是在原有能力之上加了三层新能力——路由判断、工具编排、自我纠错分别对应了该不该查、怎么查、查不到怎么办三个核心问题。3. 源码架构与核心模块拆解概念说清楚之后必须落到代码上。我提供的源码是一个基于LangGraph的最小可运行版本环境Python 3.10依赖langgraph、langchain-openai、faiss-cpu、pydantic。为什么选LangGraph而不是LangChain的AgentExecutor因为LangGraph把Agent的运行流程显式建模为一个图结构节点和边都看得见摸得着调试和扩展都比隐藏循环的AgentExecutor好得多尤其当你要在循环中加路由判断、资格评估这样的逻辑时图的优势非常明显。3.1 工程结构概览项目源码目录结构如下agentic_rag/ ├── agent.py # 核心Agent节点定义路由、检索、评估、生成 ├── tools.py # 工具层封装知识库检索、网页查询、数据库查询 ├── orchestrator.py # LangGraph状态图与工作流编排 ├── config.py # 全局配置模型名、阈值、向量路径 ├── main.py # 命令行/API入口 ├── data/ # 示例知识库文档txt/md └── requirements.txt # 依赖列表这个结构刻意砍掉了很多花活儿每个文件职责单一方便你按图索骥理解代码。真实生产环境肯定还要加日志、监控、缓存、持久化但作为源码解析越精简越好。3.2 状态与节点设计LangGraph的核心概念是StateGraph它把每次Agent运行的行为抽象成一个全局状态对象各个节点读取状态、修改状态然后流转到下一个节点。我给这个Agentic RAG定义的状态结构如下from typing import TypedDict, List, Optional from pydantic import BaseModel, Field class AgentState(TypedDict, totalFalse): question: str # 用户输入的原始问题 query: str # 当前检索用的查询词可由router改写 intent: str # 路由意图chat / retrieve / summarize documents: List[dict] # 检索到的文档列表 retries: int # 当前重试次数 max_retries: int # 最大重试次数 answer: str # 最终生成的答案 tool_scores: List[float] # 各个工具调用的评分记录 intermediate_steps: List[str] # 完整执行轨迹用于日志和debug这个状态对象其实就是整个Agent的工作内存。比如retries字段是自纠正功能的核心变量——每轮重试加1超过max_retries就终止检索循环直接进入生成阶段或拒答。这种显式的重试计数器比在大模型提示词里写如果信息不足请重试要可靠得多。节点设计上我拆成六个节点router_node路由、tool_node工具调用、grade_node评估文档、rewrite_node查询词改写、generate_node答案生成、respond_node直接问答回复。节点的作用就是封装一段逻辑每个节点输入state、输出更新后的state。3.3 控制流是怎么走的整个Agentic RAG的工作流在orchestrator.py中组成了一个状态图逻辑如下用户问题进入router_nodeLLM判断意图。如果是chat直接走respond_node不调检索如果是retrieve或summarize进入tool_node。tool_node根据结果调用相关工具拿到候选文档后进入grade_node。grade_node对文档逐条打分如果最高分低于阈值且重试次数未满走rewrite_node修改查询词重新回到tool_node。如果分数达标或重试次数耗尽走generate_node将文档与问题一起送入LLM生成最终答案。from langgraph.graph import StateGraph, END def build_graph(): g StateGraph(AgentState) g.add_node(router, router_node) g.add_node(tool, tool_node) g.add_node(grade, grade_node) g.add_node(rewrite, rewrite_node) g.add_node(generate, generate_node) g.add_node(respond, respond_node) g.set_entry_point(router) g.add_conditional_edges( router, lambda state: respond if state[intent] chat else tool, {respond: respond, tool: tool}, ) g.add_edge(tool, grade) g.add_conditional_edges( grade, lambda state: rewrite if ( state[tool_scores] and max(state[tool_scores]) 0.6 and state[retries] state[max_retries] ) else generate, {rewrite: rewrite, generate: generate}, ) g.add_edge(rewrite, tool) g.add_edge(generate, END) g.add_edge(respond, END) return g这段图代码是整个Agentic RAG架构的骨架。路由节点决定是否检索tool和grade形成检索循环rewrite节点为自纠正提供落点。你要改成一个不检索、纯聊天机器人只需关掉tool节点要改成更复杂的多工具Agent只需在tool节点里多挂几个工具并在grade节点调整评分策略。可控性在这里这是普通LangChain AgentExecutor做不到的。4. 关键实现与参数设计架构图看得懂但代码只有跑起来才能证明有效。这一章讲实现细节我会把每个关键模块的原理讲透顺带解释为什么要这样设计。4.1 工具的抽象方式tools.py里把所有外部能力封装成统一接口每个工具是一个函数入参是查询字符串返回是一份文档列表。核心是knowledge_base_search它读取Faiss索引和文档元数据执行相似度检索并把结果转成统一格式。这是我从实际项目中提炼出的最简实现import faiss import numpy as np from openai import OpenAI client OpenAI() class VectorStore: def __init__(self, index_path: str, doc_texts: List[str]): self.index faiss.read_index(index_path) self.doc_texts doc_texts def search(self, query: str, k: int 4) - List[dict]: emb client.embeddings.create( modeltext-embedding-3-small, inputquery ).data[0].embedding vec np.array([emb], dtypenp.float32) scores, idxs self.index.search(vec, k) results [] for score, idx in zip(scores[0], idxs[0]): results.append({ text: self.doc_texts[idx], score: float(score), source: knowledge_base }) return results这里有个容易踩的坑Faiss返回的相似度得分是向量内积或L2距离的变换不同索引类型得分含义不同不能直接当作置信度使用。我在代码里统一做了一个softmax归一化处理把原始得分映射到0~1区间。4.2 路由与查询改写给模型一条冷静发挥的轨道路由节点和查询改写节点都是典型的Prompt工程问题。核心技巧是不让模型自由发挥而是用pydantic定义一个严格的schema要求模型输出固定JSON解析失败就重试保证后续代码拿到的永远是结构化的字段class RouterSchema(BaseModel): intent: str Field(descriptionchat/retrieve/summarize) query: str Field(description改写后的检索查询词) ROUTER_PROMPT 你是一个智能信息检索路由。 根据用户问题判断检索策略并输出JSON。 规则 1. 如果用户只是闲聊、情绪表达、或问题不依赖外部知识intentchat。 2. 如果需要从知识库/文档/数据库中获取信息才能回答intentretrieve。 3. 如果需要对多篇文档做对比总结intentsummarize并确保query包含完整比较对象。 用户问题{question} 请输出JSON{{intent: ..., query: ...}}为什么要做查询改写因为用户的自然语言经常包含代词、口语词和隐含上下文原话直接去embedding检索效果很差。例如用户问它和LangChain相比有什么优势这里它指代上文提到的某个框架直接拿去查库几乎没结果改写后的查询词应该变成Agentic RAG与LangChain的区别。4.3 评分、阈值与答案生成grade_node是我最想强调的一个部分。很多人在做Agentic RAG时忽略了一个关键事实LLM自己判断检索结果是否相关比任何规则都靠谱。所以我设计了一个独立的评估节点利用LLM判断文档相关性输出一个二分类标签相关/不相关并转换为分数。GRADE_PROMPT 你是一个文档相关性评估器。 用户问题{question} 候选文档{document} 请判断该文档是否包含回答用户问题所需的关键信息。 只输出一个数字1表示相关0表示不相关。 def grade_node(state: AgentState) - AgentState: scores [] for doc in state[documents]: resp llm.invoke(GRADE_PROMPT.format( questionstate[query], documentdoc[text] )) scores.append(float(resp.content.strip())) state[tool_scores] scores return state这里阈值取0.6是我在示例数据上反复试出来的经验值。但请注意阈值不是固定的它取决于你的知识库质量和向量模型。如果知识库和用户问题匹配度整体偏高阈值可以设到0.75如果知识库覆盖偏分散、问题类型多样0.5更合适。生产环境务必拉一段带标注的测试集做校准。生成节点不做太多技巧性设计就是把检索结果拼成上下文送进LLM但Prompt里必须限定仅基于给定文档回答信息不足请明说。这样即使检索环节给出了不够完美的结果生成环节也不会信口开河。5. 跑通之后踩过的坑与排查路径代码能跑是一回事跑得好是另一回事。这章我把源码在真实运行中遇到的几个典型问题按排查链路完整记录下来这些问题在你自己的部署中大概率也会遇到。5.1 子代理提示词没约束输出格式JSON解析直接崩最初版本中路由节点的Prompt只是说请输出意图和改写后的查询词没有指定JSON格式也没有pydantic schema。第一次demo演示时模型输出了一段自然语言该问题需要检索知识库查询词为Agentic RAG的概念。后面的JSON解析直接抛异常整个Agent卡死在router节点。排查链路很清晰第一步看报错日志定位到json.loads抛JSONDecodeError第二步往前翻发现模型原始输出不是JSON第三步检查Prompt发现确实没有格式约束。修复方案是引入with_structured_output或pydantic schema让LangChain的ChatOpenAI强制返回结构化对象。这个坑给所有Agent开发者的教训是永远不要让LLM自由输出任何中间结果都必须走结构化schema。调参的时候我建议手边准备几组典型问题每次改动Prompt后先跑一遍全量回归观察路由准确率。我自己的标准是至少20条覆盖不同意图的测试集路由准确率上95%才算合格。5.2 召回评分集中在0.55~0.65阈值设0.6导致全部走rewrite当我把评分阈值调成0.6之后发现系统的重试率飙升到80%。排查时打印所有文档的评分分布发现样例知识库中大多数合法文档的评分都集中在0.550.65之间0.6一刀切导致大量本应直接生成的查询被误判为不够相关然后走查询改写、二次检索响应时间翻了一倍。排查链路第一步统计所有成功问答的文档评分直方图发现分布右偏且方差小第二步把阈值降到0.5重试率回落到20%但出现凑合着用弱相关文档硬答的情况第三步引入动态阈值根据本次检索最高分和次高分之间的gap来判断是否需要重试如果最高分明显领先gap 0.1即使绝对分数不高也直接生成如果分数差距很小且整体偏低说明检索词确实有问题才走rewrite。这个修正让系统既不会过度重试也不会在检索明显失败时硬答。调阈值没有万能公式唯一可靠的办法就是拿你自己的数据做分布分析再决定是固定阈值还是动态规则。5.3 多轮循环导致上下文膨胀token直接打满严格来说Agentic RAG每次查询走完循环最多也就三四轮如果后续扩展成多用户复用的服务端Agent这个问题会立刻暴露对话历史越长输入给LLM的token就越多。我最初在generate_node里简单地把整个历史对话塞进上下文跑了十几轮后直接触发模型上下文上限。排查思路是画一条token增长曲线历史消息每轮增加工具调用产生的中间步骤也被塞进消息最终一轮的输入token接近8k还没算文档内容就已经打满。修复做了三件事第一把中间步骤和最终答案分开存储历史消息只保留用户问题、工具结果摘要、最终答案不保留工具内部日志第二给工具结果做截断超过500字符时保留首尾关键信息第三设置历史窗口为最近6轮更早的内容压缩成一段摘要作为全局缓存在不消耗预算的前提下保留语义。实际上第三点才是Agent长期记忆的正解不是无限堆之前的话而是压缩成可检索的记忆块。6. 后续怎么扩展源码只是骨架真正的价值在于你往里填什么。我在实际部署中总结了几条扩展建议从易到难排个序。先做多数据源接入。把tools.py里模拟的web_search和database_query替换成真实实现前者可以接Search API后者可以接关系型数据库或图数据库。扩展时要注意工具数量的增加会显著影响路由和编排的准确性——工具一多LLM就容易选错工具。我的经验是给每个工具写一段精确的description告诉模型这个工具适合解决什么问题、不适合解决什么问题效果比优化模型本身还明显。再做评估闭环。Agentic RAG的循环逻辑复杂能不能稳定收敛全靠评分和路由的准确率。我在生产项目里维护了一个eval_set.json每条样例包含问题、预期检索工具、预期答案要点每次修改Prompt或阈值后全量跑一遍回归用召回率和路由准确率来卡发布标准。没有评估闭环的Agent系统本质上是在裸奔。最后做异步和缓存。如果服务端要支撑并发访问需要把tools.py里的同步函数改成async并用cache装饰器对相同查询词做缓存。缓存键不要直接用用户原始问题而要用改写后的查询词否则同一语义不同表述的用户问题会反复穿透缓存。我自己做Agentic RAG最大的体会是不要把Agent想得太神秘它的本质就是一套带反馈控制的流程编排系统LLM负责其中的智能决策工程代码负责约束和兜底。源码跑通只是第一步把路由、评估、阈值、上下文管理这些细节打磨好才真正决定它在业务里能不能用、好不好用。如果你在部署过程中遇到什么奇怪的坑欢迎带着你的状态图和日志来找我聊排错这东西一聊就通。本文还有配套的精品资源点击获取