☰
Agentic RAG实战:架构原理、源码实现与常见坑
2026/10/10 3:43:16 网站建设 项目流程

简介:这是一份面向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结构,指明intent(chat、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, total=False): 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中组成了一个状态图,逻辑如下:

  1. 用户问题进入router_node,LLM判断意图。如果是chat,直接走respond_node,不调检索;如果是retrieve或summarize,进入tool_node。
  2. tool_node根据结果调用相关工具,拿到候选文档后进入grade_node。
  3. grade_node对文档逐条打分,如果最高分低于阈值且重试次数未满,走rewrite_node修改查询词,重新回到tool_node。
  4. 如果分数达标或重试次数耗尽,走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( model="text-embedding-3-small", input=query ).data[0].embedding vec = np.array([emb], dtype=np.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(description="chat/retrieve/summarize") query: str = Field(description="改写后的检索查询词") ROUTER_PROMPT = """你是一个智能信息检索路由。 根据用户问题判断检索策略,并输出JSON。 规则: 1. 如果用户只是闲聊、情绪表达、或问题不依赖外部知识,intent=chat。 2. 如果需要从知识库/文档/数据库中获取信息才能回答,intent=retrieve。 3. 如果需要对多篇文档做对比总结,intent=summarize,并确保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( question=state["query"], document=doc["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.55~0.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负责其中的智能决策,工程代码负责约束和兜底。源码跑通只是第一步,把路由、评估、阈值、上下文管理这些细节打磨好,才真正决定它在业务里能不能用、好不好用。如果你在部署过程中遇到什么奇怪的坑,欢迎带着你的状态图和日志来找我聊,排错这东西,一聊就通。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询