目录一、基于 LangGraph 实现的代理式 RAG (检索增强生成) 系统案例介绍二、编码思路三、代码实现步骤一组件导入以及初始化大模型步骤二准备知识库”并创建”检索工具步骤三设计工作流程节点”节点1决策节点 generate_query_or_respond节点2检索器工具节点 retriev节点3:问题优化节点 rewrite_question节点4答案生成节点 generate_answer步骤四添加节点步骤五定义边步骤六编译图步骤七执行图运行结果四、相关问题五、总结从 Agent 视角评价本 RAG 检索案例一、基于 LangGraph 实现的代理式 RAG (检索增强生成) 系统案例介绍还记得我们在 LangChain 篇章中编写过 LCEL 链式的 RAG 系统。整个系统就是一条直线输入问题 → 检索文档 → 组合提示词 → 生成答案 → 输出结果它适合初学者因为概念少只需要理解 Runnable, StrOutputParser 等基础概念逻辑简单线性流程易于理解和调试代码直观一眼就能看懂数据流动快速上手几分钟就能跑通整个流程而在 LangGraph 篇章中我们将会构建一个智能的文档问答系统复杂版它能够根据检索结果的质量动态调整查询策略。逻辑如下图所示举个例子这里我们提供了 “某某就业课” 相关的几篇课程宣传文档作为知识库。可以支持我们询问某某提供了哪些课程Java 开发方向的课程安排测试开发方向的主线课程有哪些C 开发方向的项目列表Redis 课程内容是什么……当用户询问 某某提供了哪些课程 时根据要求会经历模型决定调用检索工具搜索 某某课程检索到相关文档后进行评估如果文档相关直接生成答案如果文档不相关重写问题后重新检索这个设计确保了即使初次检索不成功系统也能通过重写查询来改进结果质量比传统 RAG 系统更加鲁棒。鲁棒性是指系统在面对各种内外部干扰时仍能保持其性能稳定的能力。具体来说它描述了系统在不确定性、噪声、误差或其他不利因素下的表现。鲁棒性与稳定性不同稳定性关注系统在扰动消失后的恢复而鲁棒性关注在持续扰动下的表现。在模型中鲁棒性指的是模型在陌生环境或噪声干扰下依旧能够完成预期任务的能力。在控制系统中鲁棒性指系统在一定参数摄动下维持性能的特性是现代控制系统设计的重要指标。因此相较于 LangChain 篇章中链式的 RAG是 “怎么让 RAG 跑起来”该案例将学习 “怎么就让 RAG 跑的更好、更智能”二、编码思路构建一个智能文档问答系统我们要将复杂流程分解为离散步骤且通过共享状态连接各个节点核心设计思路如下模块化设计每个节点只做一件事职责清晰质量闭环检索 → 检查 → 优化 → 再检索确保答案质量智能路由AI 自主决定下一步行动无需人工干预因此我们需要第一步准备 知识库数据加载与处理第二步创建 检索工具第三步设计 工作流程节点节点 1决策节点 generate_query_or_respond节点 2检索器工具节点 retrieve节点 3问题优化节点 rewrite_question节点 4答案生成节点 generate_answer第四步组装 工作流水线条件边 1LLM 决策是否需要进行知识库检索条件边 2检测【检索到的文档】是否与【问题】相关最终 Graph 示意图三、代码实现步骤一组件导入以及初始化大模型from typing import Literal这是 python 类型注解工具用来限制变量只能取指定的几个固定字符串值后面做条件分支判断的时候会用到。from langchain.chat_models import init_chat_modellangchain 提供的统一初始化大模型的函数不管是 deepseek、openai 都可以用这个接口快速创建聊天模型对象。from langchain_openai import OpenAIEmbeddings嵌入模型类。虽然名字带 OpenAI但是兼容所有 OpenAI 协议接口这里用来对接 deepseek 的 embedding 接口把文本转为向量。from langchain_core.messages importHumanMessage, filter_messages消息对象。HumanMessage 代表用户输入filter_messages 用来过滤历史消息防止上下文消息过长做消息裁剪。from langchain_community.document_loaders import UnstructuredMarkdownLoadermarkdown 文档加载器读取本地 md 文件解析成 Document 文档对象。前面我们已经遇到这个库需要额外安装 unstructured 依赖。from langchain_text_splitters import RecursiveCharacterTextSplitter递归字符文本分割器RAG 流程里用来把长文档切分成小块 chunk方便做向量检索。from langchain_core.vectorstores import InMemoryVectorStore内存向量库全部数据保存在内存程序退出数据就丢失适合本地调试开发不用部署真正的向量数据库服务。from langchain_classic.tools.retriever import create_retriever_tool快速把检索器包装成工具给到 agent 调用agent 就可以主动执行文档检索。from langgraph.constants import START, ENDLangGraph 内置常量代表图的起始节点和结束节点图流转的起点终点。from langgraph.graph import MessagesState, StateGraphStateGraph 是构建图的核心类用来定义节点、边、条件边MessagesState是 LangGraph 内置好的状态自带 messages 消息列表不需要自己手写 TypedDict 定义状态简化开发。from langgraph.prebuilt import ToolNode, tools_condition预构建组件。ToolNode 专门用来执行工具调用不用手写执行工具的函数tools_condition 是预写好的条件边判断函数自动判断大模型输出是否需要调用工具需要就走向 ToolNode不需要就走向 END 结束。写 Agent 的时候高频使用。from pydantic import BaseModel, Field数据校验模型做结构化输出的时候定义输出格式。init_chat_model 初始化聊天大模型。modeldeepseek‑chat 指定使用 deepseek 聊天模型。 temperature0温度设置为 0模型输出确定性最高减少随机发散写 Agent、RAG 场景一般设置 0。api_key传入 deepseek 密钥。embeddings OpenAIEmbeddings(...) 初始化嵌入模型。虽然类名叫 OpenAIEmbeddings因为 deepseek 兼容 OpenAI 接口协议所以直接复用这个类。modeldeepseek‑embedding 指定 deepseek 向量模型。 api_key 填写 deepseek 密钥。 base_urlhttps://api.deepseek.com/v1 指定接口请求地址指向 deepseek 服务端点。步骤二准备知识库”并创建”检索工具加载文档扁平化列表:paths 是一个字符串列表存放所有待读取 markdown 文档的相对路径。这里使用../代表向上跳出当前代码所在文件夹去访问 Docs 目录下的文件。运行时要保证目录层级正确。接着一行是列表推导式循环 paths 里面每一个文件路径用 UnstructuredMarkdownLoader 加载 md 文件调用 load() 得到文档对象。每一个文件加载返回的是 List[Document]所以 docs 是二维嵌套列表格式为 [[doc1], [doc2]]。然后做列表扁平化把嵌套二维列表展开成一维 docs_list全部 Document 对象放到同一个数组给后面文本分割器使用。tiktoken 文本分割器切分文档:RecursiveCharacterTextSplitter.from_tiktoken_encoder基于 tiktoken token 计数器做分割不是按字符数切割而是按大模型 token 数量切割更贴合 LLM 实际输入。encoding_namecl100k_base对应 gpt 系列的 token 编码deepseek 也兼容这套编码。chunk_size1000每一块最大 token 数量 1000。chunk_overlap50块与块之间重叠 50 个 token避免语义被截断丢失上下文。split_documents(docs_list) 把完整文档列表切分成很多小块返回 doc_splits这就是后续存入向量库的片段。构建内存向量库:InMemoryVectorStore 内存向量库数据全部存在程序内存程序关闭数据全部销毁只适合开发调试不能用于生产环境。from_documents 是类方法传入切分好的文档片段 doc_splits 以及前面初始化好的 embeddings 嵌入模型。内部会自动循环每一个文档块调用 embedding 接口生成向量把文本和向量一起存入内存向量存储。获取检索器:vectorstore.as_retriever() 把向量存储转换成检索器对象对外提供统一检索接口。search_kwargs{k:2} 代表每次检索返回相似度最高的 2 个文档片段。Agent 调用这个工具的时候最多拿到 2 条参考文档。创建检索工具:create_retriever_tool 是 LangChain 封装好的工具生成函数直接把检索器包装成 Agent 可以调用的工具。retriever 传入上面构建好的检索器实例。name 工具名字大模型通过这个名字识别要调用哪一个工具。description工具描述非常关键大模型依靠这段描述判断什么时候应该调用这个检索工具。描述写的不准确会出现该调用工具不调用、不该调用乱调用的问题。得到 retriever_tool 之后后面把这个工具交给 LangGraphAgent 就可以自主选择执行文档检索。步骤三设计工作流程节点”根据 Graph 示意图设计出以下四个节点:节点1决策节点 generate_query_or_respond节点2检索器工具节点 retrieve节点3:问题优化节点 rewrite_question节点4答案生成节点 generate_answer我们先看上面部分被注释掉的代码。原本在 LangGraph 中如果自己手写状态需要用 TypedDict 定义messages 字段使用 Annotated 加上 operator.add代表新消息做追加合并而不是直接覆盖旧列表。但我们这个案例没有手写状态类而是直接使用 LangGraph 内置提供的 MessagesState。内置的 MessagesState 内部已经帮我们写好了 messages 字段并且默认就是追加消息的更新逻辑省去自己写 TypedDict 的工作。messages 消息列表承担两个作用一是会话记忆保存整个对话历史二是维护上下文每一轮大模型、工具的输出全部塞进 messages作为下一轮的输入。节点1决策节点 generate_query_or_respond我们先来看第一个节点函数:def generate_query_or_respond(state: MessagesState):这是图的第一个业务节点函数。 节点函数第一个入参固定为 state类型标注 MessagesState代表读取当前图的状态对象。三引号里面是函数文档字符串说明这个节点的业务逻辑接收当前对话状态交给大模型由大模型自己做判断要么调用检索工具去查文档要么直接生成回答返回给用户。model.bind_tools([retriever_tool])bind_tools 把我们前面做好的检索工具绑定到大模型。绑定之后大模型就知道这个工具的名字、描述、入参格式大模型可以自主选择要不要调用该工具。.invoke(state[messages]) 把状态里面完整的历史消息列表传给模型执行。state[messages] 就是全部对话上下文包含用户提问、之前 AI 消息、工具返回结果。result ... 执行完成得到 AIMessage 对象。 这个 AIMessage 分两种情况需要检索AIMessage 内部携带tool_calls记录要调用retrieve_bit工具以及查询参数不需要检索AIMessage 只有普通 content 文本没有 tool_calls代表直接输出答案。return {messages: [result]}节点函数返回字典。 因为 MessagesState 的 messages 更新策略是 operator.add 追加模式返回的消息列表会追加到状态原有 messages 后面不会覆盖历史消息。 这里把模型输出的 AIMessage 包装成列表返回写入状态。下面我们再看接下来的两个节点函数:节点2检索器工具节点 retrievToolNode 是 LangGraph 预构建好的工具执行节点我们使用 ToolNode 之后就不需要自己手写函数了。接着传入参数是工具列表这里把之前创建的 retriever_tool 检索工具传进去。内部会自动读取状态中 AIMessage 里面的 tool_calls解析工具名字、参数调用对应的检索工具拿到检索结果之后自动生成 ToolMessage 写入状态的 messages 消息列表。当条件边流转到这个节点时就会执行文档检索把检索返回的内容放到消息历史交给后续大模型处理。节点3:问题优化节点 rewrite_question第三个节点之前我们先声明了一个重写问题提示词常量 REWRITE_PROMPT这是一个多行字符串常量作为重写用户问题的提示模板。 模板里面 {question} 是占位符后续使用 format 方法把原始用户提问填充进去。业务作用拿到用户原始问题交给大模型让大模型改写优化问题。比如用户提问简短模糊重写之后的问题语义更完整传给向量库检索可以提升检索召回效果。接下来的 def rewrite_question(state: MessagesState): 是第三个业务节点入参接收当前图状态。 文档字符串说明这个节点功能对用户原始问题做改写优化。注释 state messages 包含 [H, A, T]H 代表 HumanMessage 用户消息A 代表 AIMessage 大模型消息T 代表 ToolMessage 工具返回消息。question state[messages][0] 取出消息列表第一条消息也就是用户最开始输入的原始提问。prompt REWRITE_PROMPT.format(questionquestion)把原始 question 填充进提示词模板组装完整 prompt。result model.invoke([HumanMessage(contentprompt)])构造 HumanMessage 调用大模型让模型输出优化改写后的问题返回结果是 AIMessage 对象。return {messages: [HumanMessage(contentresult.content)]}节点返回字典。把模型改写好的文本包装成一条新的 HumanMessage追加写入状态 messages 列表。后续流程就使用这条改写后的用户消息去做检索或者交给大模型回答。关键点这里不是返回 AIMessage而是把改写后的内容包装成 HumanMessage。相当于把用户输入替换成优化之后的问题后续流程就基于优化后的问题继续执行。节点4答案生成节点 generate_answer接下来看节点4:最开始是 GENERATE_PROMPT 回答生成提示词模板这是 RAG 生成答案的提示词常量。里面定义了大模型的角色与回答约束必须依赖传入的检索上下文如果上下文找不到答案就直接回复不知道输出限制最多三句话保证回答简短。模板中有两个占位符{question} 用来填充用户问题{context} 用来填充向量库检索出来的文档片段。后续代码调用 format 把真实内容填进去组装完整请求交给大模型。def generate_answer(state: MessagesState): 是第四个业务节点负责最终组装上下文、调用大模型产出最终回答。文档字符串标注该节点功能生成答案。注释 state messages 包含 [H, A, T]H 代表 HumanMessage 用户消息A 代表 AIMessage 大模型消息T 代表 ToolMessage 工具返回消息说明状态消息列表里面存放着多轮不同类型消息。question state[messages][0].content取出消息列表第一条消息也就是用户的原始问题文本。context state[messages][-1].content取消息列表的最后一条消息。走到这个节点的时候最后一条消息就是 ToolNode 执行完检索之后返回的 ToolMessagecontent 里面存放检索出来的文档片段作为回答的参考上下文。prompt GENERATE_PROMPT.format(questionquestion, contextcontext)把用户问题、检索到的上下文填充进提示词模板得到完整 prompt。model.invoke([HumanMessage(contentprompt)])把组装好的 prompt 封装成 HumanMessage 调用大模型得到 AIMessage 对象。return {messages: [model.invoke([HumanMessage(contentprompt)])]}节点返回字典把大模型输出的 AIMessage 放入列表追加写入 state 的 messages。走完这个节点之后图就可以流向 END 结束把最终回答给到调用方。这个节点特点不走工具绑定逻辑是传统 RAG 的生成逻辑直接拿检索上下文 问题手动拼接 prompt 交给大模型生成答案。步骤四添加节点workflow StateGraph(MessagesState)实例化 StateGraph传入状态类型 MessagesState。workflow 就是工作流图纸对象刚创建完成是空图没有节点、没有边后面所有注册节点、配置流转路线都操作这个对象。workflow.add_node(generate_query_or_respond)注册第一个业务节点。只传函数对象节点名称自动使用函数名 generate_query_or_respond。该节点负责调用绑定工具的大模型产出 AIMessage判断是否需要检索。workflow.add_node(retrieve, retriever_node)注册工具节点。这里传两个参数第一个字符串 retrieve 是人为指定的节点名称第二个是 ToolNode 实例对象 retriever_node。ToolNode 不是普通 def 定义的函数不能自动获取名字所以必须手动指定节点名 retrieve后续配置边就用这个字符串引用该节点。workflow.add_node(rewrite_question)注册问题优化重写节点直接传入函数节点名字自动为 rewrite_question负责优化用户原始提问。workflow.add_node(generate_answer)注册生成最终答案节点直接传入函数节点名字自动为 generate_answer拿到检索上下文后拼接 prompt 输出最终回答。执行完全部 add_node 之后图已经登记全部业务节点但是节点互相之间没有连线不知道执行顺序接下来就要配置边来定义流转逻辑。步骤五定义边workflow.add_edge(START, end_keygenerate_query_or_respond) 是配置图的入口。START 是 LangGraph 内置的起始虚拟节点图一启动就直接流向 generate_query_or_respond 节点正式开始业务逻辑。接下来是第一处条件边workflow.add_conditional_edges。source 指定来源节点为 generate_query_or_respond判断函数使用框架内置的方法 tools_condition。tools_condition 会读取状态里最新的 AIMessage检查是否存在 tool_calls 工具调用。path_map 做结果映射如果返回 tools就流转到 retrieve 节点执行检索如果返回 end直接走到 END 结束整张图。 业务含义大模型判断需要调用工具就去检索不需要工具直接结束。GRADE_PROMPT 是文档相关性打分的提示词模板。作用是交给大模型对比检索出来的文档上下文和用户问题输出 yes 或者 no代表文档是否和问题相关。class GradeDocuments (BaseModel)定义 Pydantic 结构化输出模型。只有一个 score 字段约束输出只能是 yes 或者 no用来接收大模型返回的相关性判定结果保证输出格式稳定。def grade_documents (state: MessagesState) 是自定义条件路由函数从 retrieve 节点出来之后就会执行这个函数。user_messages filter_messages(state [messages], include_types[human])过滤消息列表只取出 HumanMessage 类型消息拿到最新一条用户问题。tool_message state [messages][-1]取消息列表最后一条也就是 retrieve 节点执行完成得到的 ToolMessage里面存放检索文档片段。prompt GRADE_PROMPT.format (contextcontext, questionquestion)把问题和检索文档填充进打分提示词。model.with_structured_output (GradeDocuments).invoke开启结构化输出强制大模型按照 GradeDocuments 模型格式返回结果。if result.score yes如果文档相关返回字符串 generate_answer路由到生成答案节点else 返回 rewrite_question路由到问题重写节点。检索出来的文档不相关就重写用户问题重新走一遍完整流程。第二处条件边workflow.add_conditional_edgessource 来源节点是 retrieve使用上面写好的 grade_documents 作为判断函数。path_map 列表对应函数的两个返回值分别映射到 generate_answer 和 rewrite_question。workflow.add_edge (start_keygenerate_answer, end_keyEND)固定边当执行完 generate_answer 生成最终回答之后流向 END 结束整个图。workflow.add_edge (start_keyrewrite_question, end_keygenerate_query_or_respond)固定边。问题重写完成之后重新回到最开始的 generate_query_or_respond 节点开启新一轮循环使用优化后的问题重新判断是否检索。到此整张图的节点、全部边、条件路由配置完成。下一步就是调用 compile 编译图得到可执行对象。步骤六编译图graph workflow.compile()调用 compile 方法编译工作流对象 workflow。编译阶段会做校验检查节点、边、条件路由是否合法有没有节点找不到、循环路径异常这类问题。编译完成之后得到 graphgraph 才是真正可以调用执行的实例workflow 只是用来定义的图纸不能直接跑。步骤七执行图graph.invoke() 是被注释掉的代码invoke 是一次性调用方式整张图全部跑完之后一次性返回最终完整状态看不到中间每一步节点输出。而下面的 graph.stream() 是流式执行方式会按照图的执行步骤每走完一个节点就产出一块 chunk 数据可以拿到每一个节点执行后的状态更新方便调试观察每一步流转。传入的字典是图的初始 statemessages 数组里面放入 HumanMessage就是用户输入的问题作为整个 Agent 的输入。外层 for chunk in graph.stream 循环迭代每一个节点执行完成返回的 chunk 对象。每一个 chunk 对应一个节点执行完毕之后对状态的变更。被注释的 print(chunk)打开之后可以打印原始完整的 chunk 字典适合调试看底层数据结构。内层 for node, update in chunk.items() 遍历 chunk。node 代表当前刚刚执行完成的节点名字update 代表这个节点对 state 做出的更新内容。print (f由节点 {node} 更新消息)打印日志直观看到现在是哪一个节点执行完毕。update [messages][-1].pretty_print ()取出本次更新消息列表里的最后一条消息调用 pretty_print 做格式化打印。可以清晰看到 AIMessage、ToolMessage、HumanMessage 不同消息类型以及对应的 content 内容方便观察 Agent 每一轮模型输出、工具调用结果。print (\n\n) 打印换行用来隔开不同节点输出控制台阅读体验更好。运行之后控制台就会按顺序打印各个节点的输出我们可以完整看到模型判断、工具检索、文档打分、问题重写、生成最终回答的全部中间过程。运行结果第一部分由节点 generate_query_or_respond 更新消息。这里输出的是 AIMessage里面携带 ToolCalls。说明大模型判断当前问题需要做检索生成了工具调用请求。工具名字为 retrieve_bitcall ID 是本次工具调用的唯一标识Args 里面的 query 参数就是大模型生成的检索查询词C开发方向的项目列表。模型把用户原始问题整理成适合向量库检索的查询语句交给后续工具节点执行。第二部分由节点 retrieve 更新消息。输出类型变成 ToolMessage。代表 retrieve 节点已经执行完成调用向量检索拿到文档结果。Name 字段对应工具名称 retrieve_bit省略号代表检索返回的文档片段内容。这条 ToolMessage 会追加到 state 的 messages 列表作为后续节点的输入材料。走到这里之后会执行条件路由函数 grade_documents对刚刚拿到的检索文档做相关性打分。在本次示例运行中文档判定结果为 yes文档和问题相关所以路由走向 generate_answer不会走到 rewrite_question 重写问题分支。第三部分由节点 generate_answer 更新消息。再次输出 AIMessage这就是整个 RAG Agent 产出的最终回答。节点拿到原始用户问题加上上一步 ToolMessage 携带的检索文档填入 GRADE_PROMPT 提示词模板交给大模型。 输出内容C开发方向的项目包括多个层次的挑战项目、标准项目和极简项目。挑战项目如微服务的即时通讯系统和视频点播系统涵盖复杂功能和技术组件标准项目包括仿QQ音乐的音乐播放客户端等适合快速掌握极简项目则如贪吃蛇游戏适合基础薄弱的同学。总共有15个不同规格的项目可供选择。本次案例没有触发重写问题的分支。如果向量库检索出来的文档和问题完全无关grade_documents 就会返回 no就会流转到 rewrite_question 节点改写查询再回到 generate_query_or_respond重新发起一轮检索。stream 流式打印的好处就在于能够完整看到每一步节点输出调试 Agent 的时候可以直观定位问题比如模型没有生成工具调用、检索文档不匹配、打分逻辑异常等。四、相关问题Q1节点都是执行对应逻辑的函数条件边也要绑定对应函数制定条件逻辑对吗是的。节点每一个业务节点就是一个函数接收 state 状态作为入参执行业务逻辑返回状态更新字典。普通业务节点、重写问题、生成答案全部都是自定义函数ToolNode 是框架封装好的特殊节点底层内部同样也是函数逻辑。条件边普通 add_edge 固定边只写起点终点无条件直接流转。add_conditional_edges 条件边必须绑定一个路由判断函数函数读取 state 状态做逻辑判断返回下一个节点名字框架根据返回值决定跳转到哪一个节点实现分支跳转。本案例里两处条件边一处绑定 tools_condition 内置函数判断 AIMessage 有没有 tool_calls另一处绑定我们手写的 grade_documents 函数做文档相关性打分。Q2AIMessage 就一定是大模型生成的消息吗是的。AIMessage 代表 AI 角色消息只能来自大模型调用返回。不管是普通回答还是携带 tool_calls 工具调用指令只要是模型输出就会包装成 AIMessage 对象。节点 1 generate_query_or_respond 输出 AIMessage里面存放 tool_calls 工具调用指令告诉框架要执行检索。节点 3 generate_answer 输出 AIMessage存放最终给用户看的自然语言答案。Q3节点 1 产出带 tool_calls 的 AIMessage交给节点 2节点 2 执行完生成 ToolMessage这个流程理解是否正确完全正确。节点 1 运行大模型输出 AIMessage对象内部 tool_calls 字段填充检索工具的名称、查询参数。条件边检测到 tool_calls 不为空流转进入节点 2 retrieveToolNode。ToolNode 读取 state 中 AIMessage 的 tool_calls调用向量检索工具执行查询。工具执行完成把检索结果封装成 ToolMessage追加写入 state 的 messages 列表。ToolMessage 专门用来存放工具执行之后返回的结果不会由大模型直接生成。Q4HumanMessage、AIMessage、ToolMessage 三者是否有特定的用法三者各司其职不能混用。HumanMessage代表用户输入消息。程序启动时传入的用户提问、重写问题之后构造的模拟用户提问都使用 HumanMessage是来自 “人类视角” 的消息。AIMessage代表大模型输出消息。可以是普通文本回答也可以携带 tool_calls 工具调用指令。ToolMessage代表工具执行返回结果消息。只能是调用工具之后产生用来存放工具返回的数据例如检索文档、联网搜索结果。state 的 messages 列表就是这三类消息不断追加完整保存整个 Agent 的全部对话、工具调用历史后续每一个节点都读取这份消息列表作为上下文。Q5前面是完全手写状态、节点、条件边框架有内置方法可以简化开发是不是可以直接使用内置方法没错。我们这个案例属于手写完整版本所有条件路由、打分逻辑全部自己实现好处是每一步都可控适合学习原理能够看懂完整链路。LangGraph、LangChain 提供了预封装的 Agent 构造函数例如 create_react_agent内部已经封装好节点、条件边、tools_condition开发者只需要传入模型、工具不用手动写 add_node、add_conditional_edges快速搭建 Agent。但是封装版本会隐藏底层执行细节调试排错难度上升。学习阶段优先手写完整版本搞懂原理项目开发可以直接使用框架内置封装提升开发效率。Q6结合流程图再梳理完整闭环流程开始 START → 节点 1LLM 绑定检索工具→ 生成 AIMessage条件边判断 AIMessage 是否存在 tool_calls无 tool_calls直接走向结束输出答案。有 tool_calls流向节点 2 执行检索工具产出 ToolMessage。节点 2 出来执行条件边运行 grade_documents 检查检索文档相关性文档相关 yes流向节点 3生成最终 AIMessage执行结束。文档不相关 no流向节点 4 重写问题生成新 HumanMessage回到节点 1开启新一轮循环。五、总结从 Agent 视角评价本 RAG 检索案例这个案例不是简单的一次性 RAG 链路而是具备自我校验、自我重试能力的 RAG Agent和普通直接 “查询‑检索‑生成” 的线性 RAG 有本质区别。优点具备检索结果自我校验能力普通 RAG 拿到检索文档直接交给大模型生成答案不会管检索出来的内容是否匹配用户问题。 本案例增加 grade_documents 打分节点Agent 会主动评判检索文档相关性。一旦文档质量差不会直接输出错误答案而是触发重试逻辑规避向量检索召回不准带来的幻觉问题。形成闭环重试工作流检索文档不相关时进入重写问题节点改写 query 之后回到起点重新执行整套 Agent 流程。 相当于 Agent 发现拿到的资料不行自己重新整理问题再去查资料而不是直接把错误材料喂给大模型。这是 Agent 和普通 RAG 最核心的差异拥有反馈、修正、循环的能力。工具调用由大模型自主决策是否需要检索这件事交给 LLM 自主判断不是写死固定一定要检索。 如果用户问题不需要知识库内容模型可以直接给出回答跳过检索全流程兼顾问答效率。全链路透明可控便于调试基于 LangGraph 节点 条件边实现每一步动作都拆成独立节点借助 stream 流式输出可以观察模型工具调用、检索返回内容、打分结果、重写后的问题。 排查问题的时候很容易定位故障是 query 写得差检索召回差还是打分逻辑不合理。非常适合学习原理也方便二次修改定制。消息状态完整保存上下文依靠 messages 消息列表维护全部历史用户提问、模型工具调用、检索返回结果全部留存。每一轮循环大模型都可以拿到完整历史不会丢失上下文信息。存在的短板与现实业务问题重试没有次数上限当前代码没有设置最大重试次数。极端情况下向量库一直召回不相关文档会无限循环重写问题→检索→打分不通过→再次重写造成死循环、消耗大量 token。生产环境必须增加循环计数达到阈值直接终止流程输出兜底提示。文档打分逻辑比较单一现在只是简单二元 yes/no 判定整批检索文档。真实业务会出现部分文档相关、部分无关的场景当前逻辑要么全部放行要么全部丢弃重写无法过滤掉无关片段、保留有效文档。重写问题策略简单rewrite_question 节点只做问题改写没有多种改写策略。只生成一条新 query不支持多 query 并行检索召回能力有限。没有异常容错处理没有处理异常场景向量库报错、工具调用失败、结构化输出解析失败。一旦中间任意环节抛异常整个图直接中断缺少兜底分支。token 消耗偏高每一次重试都要完整跑一遍大模型工具调用、检索、打分、重写问题多轮循环会叠加 token 开销成本上升响应速度变慢。适用场景 不适用场景适用场景对答案准确性要求高允许一定响应延迟知识库容易出现检索召回偏差的业务比如知识库问答、内部文档助手。 不适用场景追求极致响应速度、高并发 QPS 的线上业务大量重试会拉高耗时与成本。总结它属于增强版 RAG Agent 原型把传统 RAG 升级成拥有自我反思的智能体。 普通 RAG 是 “执行预先写好的固定步骤”而这个 Agent 可以感知检索资料质量自主决定要不要重新查询。 学习层面非常优秀完整展示 LangGraph 做 Agent 的核心思想状态保存、节点分工、条件分支、循环重试。上线使用还需要补充最大重试次数、异常捕获、更完善的打分策略、兜底输出。