为什么需要 LangGraph在第 2 周的学习中我们使用create_agent()一行代码就构建出了能调用工具的智能助手。这种简洁性是 LangChain 高层 API 的优势它隐藏了 ReAct 循环、工具调用解析、消息管理等复杂细节。然而当你需要构建的不仅仅是提问→思考→回答的线性流程而是多步骤、有条件分支、需要人工审批、甚至要在失败后从断点恢复的复杂工作流时这种黑盒抽象就成了桎梏。这就是 LangGraph 诞生的背景它不是一个更高层的 Agent 封装而是一个低层的、透明的编排框架让你精确控制 Agent 执行的每一步。LangGraph 的核心灵感来源于 Google 的 Pregel 系统和 Apache Beam它把 Agent 工作流建模为一个有向图节点负责执行具体逻辑调用 LLM、执行工具、进行判断边负责决定下一步走向哪里。这种设计让你可以用 Python 函数定义节点的行为再用简单的图构建 API 把它们串联起来最终得到一个完全可控、可观测、可持久化的状态机。正如官方文档所强调的LangGraph 非常底层且专注于 Agent 编排你可以脱离 LangChain 单独使用它也可以与 LangChain 的模型和工具生态无缝集成。在 2026 年的 LangChain 生态中LangGraph 已经从可选扩展成长为框架的核心基础设施官方推荐的create_agent()底层正是基于 LangGraph 构建的。这个信号告诉我们有状态的、可定制的 Agent 工作流已经成为 LLM 应用的主流范式。核心概念图、状态、节点与边LangGraph 的世界由四个核心概念构成它们在官方 Graph API 文档中有详尽的阐述。StateGraph这是 LangGraph 提供的主要图类它接收一个用户自定义的状态 Schema 作为类型参数所有节点和边的读写都围绕这个共享状态进行。状态 Schema 通常使用 Python 的TypedDict或dataclass来定义也可以使用 Pydantic 的BaseModel获得递归数据校验能力Pydantic 性能略低于 TypedDict。状态中的每个字段都有独立的 reducer 函数决定节点返回的更新如何与现有状态合并默认行为是覆盖但通过Annotated类型你可以指定自定义 reducer比如使用operator.add将新值追加到列表而不是替换。由于消息列表是 Agent 场景中最高频的状态形式LangGraph 提供了预置的MessagesState它包含一个messages字段类型为Annotated[list[AnyMessage], add_messages]其中add_messages是一个智能 reducer新消息会被追加到列表但如果消息 ID 已存在比如人工编辑了某条消息它会执行原地更新而非重复添加同时还能自动将字典格式的消息反序列化为 LangChain 的HumanMessage、AIMessage等对象。你可以直接扩展MessagesState来添加额外的状态字段fromlanggraph.graphimportMessagesStateclassState(MessagesState):documents:list[str]llm_calls:int这段代码展示了MessagesState的扩展方式直接继承它并添加自定义字段。MessagesState本身只定义了messages一个字段但在实际 Agent 中我们往往还需要追踪其他信息比如检索到的文档列表、LLM 调用次数等。通过继承方式添加的documents和llm_calls字段会自动获得默认的覆盖式 reducer而messages字段继承了父类的add_messagesreducer 行为。这种设计让你不需要重复声明消息列表的 reducer 逻辑只需关注业务特有的状态字段。Node节点是图中执行实际工作的 Python 函数。每个节点函数接收当前图状态作为第一个参数还可以接收config包含thread_id等运行时配置的RunnableConfig和runtime包含上下文、流写入器、心跳维护等运行时工具并返回一个状态更新字典注意节点不需要返回完整的状态 Schema只需返回需要更新的字段。节点可以调用 LLM、执行工具、运行任意 Python 代码。LangGraph 不限定节点内部做什么它只关心节点返回的更新如何影响状态以及下一步走向哪里。你通过add_node(node_name, node_function)将节点注册到图中节点名称就是后续连接边时使用的标识符。如果你不显式指定节点名称LangGraph 会使用函数名作为默认名称。在底层节点函数被转换为RunnableLambda从而自动获得批处理和异步支持。Edge边定义了节点之间的路由关系。最简单的边是普通边通过add_edge(node_a, node_b)建立一条从 A 到 B 的固定路径每次 A 执行完毕后无条件进入 B。更强大的路由机制是条件边通过add_conditional_edges(node_a, routing_function)实现动态路由routing_function接收当前状态返回一个字符串或字符串列表LangGraph 根据返回值决定下一步执行哪个节点。你可以额外传入一个字典作为路由映射表将函数返回值映射到节点名称。条件边是实现 ReAct 循环、分支决策和工具调用的核心机制。除了普通边和条件边LangGraph 还提供了两个特殊的虚拟节点START和END。START表示图的入口点add_edge(START, first_node)指定了图启动时首先执行哪个节点。END表示终止节点add_edge(last_node, END)标记某个节点执行后流程结束。这两个虚拟节点让图的起点和终点定义变得显式和可读。重要的是一个节点可以有多条出边所有目标节点将在同一个 super-step 中并行执行。但对于每个节点你应该选择一种路由机制要么使用普通边实现静态路由要么使用条件边或Command实现动态路由不要混用两者否则两条路径都可能被执行使图行为难以预测。此外LangGraph 还提供了两个高级路由原语。Command允许节点在返回状态更新的同时指定下一步路由通过Command(update..., gototarget_node)将状态更新和控制流合并在一步中完成。Send则支持 map-reduce 模式让你从一个节点向多个下游节点发送不同的状态片段当节点数量在运行前不可知时尤其有用。构建第一个 LangGraph 工作流让我们按照最新官方文档的实践方式从头构建一个基于 ReAct 循环的 Agent。首先确保安装了 LangGraphpipinstall-Ulanggraph第一步是定义工具和模型。我们沿用前两周的模式定义几个简单的tool工具然后使用model.bind_tools(tools)将工具绑定到模型上让模型在需要时自动生成 tool_callfromlangchain.toolsimporttoolfromlangchain.chat_modelsimportinit_chat_modelfromdotenvimportload_dotenv load_dotenv()modelinit_chat_model(Qwen/Qwen2.5-7B-Instruct,model_provideropenai)tooldefmultiply(a:int,b:int)-int:Multiply a and b.returna*btooldefadd(a:int,b:int)-int:Add a and b.returnabtooldefdivide(a:int,b:int)-float:Divide a and b.returna/b tools[add,multiply,divide]tools_by_name{tool.name:toolfortoolintools}model_with_toolsmodel.bind_tools(tools)这里有几个关键设计值得注意。tool装饰器将普通 Python 函数转换为 LangChain 工具对象函数的 docstring 会自动成为工具的描述信息description模型正是通过这个描述来决定何时调用哪个工具因此 docstring 必须准确描述工具的用途和参数含义。tools_by_name字典将工具名映射到工具对象这是后续在工具执行节点中按名称查找工具的桥梁由于模型返回的 tool_call 只包含工具名称和参数而不包含工具对象本身这个字典就充当了工具注册表的角色。model.bind_tools(tools)是关键的绑定操作它告诉模型你有这些工具可用此后模型在推理时会自动判断是否需要调用工具并在需要时生成对应的 tool_call。注意bind_tools返回的是一个新的模型实例而非修改原模型这是 LangChain 不可变设计模式的体现。第二步是定义图的状态。我们使用MessagesState并扩展一个llm_calls计数器来追踪 LLM 调用次数fromlangchain.messagesimportAnyMessagefromlanggraph.graphimportadd_messagesfromtyping_extensionsimportAnnotated,TypedDictclassMessagesState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]llm_calls:int这里我们手动定义了MessagesState而非直接使用 LangGraph 预置的版本目的是展示add_messagesreducer 的使用方式。Annotated[list[AnyMessage], add_messages]是 Python 类型注解的高级用法第一个参数list[AnyMessage]是基础类型第二个参数add_messages是 reducer 函数LangGraph 在每次节点返回状态更新时会调用这个 reducer 来合并新旧消息。add_messages的行为比简单的列表追加更智能它会用消息 ID 来判断是新消息还是已有消息的更新新消息直接追加到列表末尾而已有消息通过 ID 匹配则原地替换内容。llm_calls字段没有指定 reducer因此默认行为是覆盖每次节点返回{llm_calls: N}时旧值会被直接替换为新值。这也解释了一个重要规则节点只需要返回需要更新的字段不需要返回完整状态。第三步是定义两个核心节点。LLM 调用节点负责向模型发送消息并获取响应注意这里我们需要包含系统提示而 LangGraph 节点函数内部可以直接构造消息列表并调用模型。工具执行节点检查最后一条 AI 消息中是否包含tool_calls如果有则逐个执行对应的工具将结果封装为ToolMessage返回fromlangchain.messagesimportSystemMessage,ToolMessagedefllm_call(state:dict):LLM decides whether to call a tool or not.return{messages:[model_with_tools.invoke([SystemMessage(contentYou are a helpful assistant tasked with performing arithmetic.)]state[messages])],llm_calls:state.get(llm_calls,0)1}deftool_node(state:dict):Performs the tool call.result[]fortool_callinstate[messages][-1].tool_calls:tooltools_by_name[tool_call[name]]observationtool.invoke(tool_call[args])result.append(ToolMessage(contentobservation,tool_call_idtool_call[id]))return{messages:result}llm_call节点的核心逻辑在于消息的组装方式。[SystemMessage(...)] state[messages]将系统提示放在消息列表最前面确保模型在每次推理时都能看到系统指令这与 OpenAI 的 messages API 格式完全一致。注意我们返回的是一个包含单条 AI 消息的列表messages: [...]而不是直接返回消息对象。这是因为add_messagesreducer 期望接收一个消息列表来与现有列表合并。llm_calls的更新使用了state.get(llm_calls, 0) 1在首次调用时提供默认值 0 防止 KeyError随后每次调用递增计数。tool_node的实现体现了一个 LLM 响应可能包含多个 tool_call的设计模型可以一次请求并行调用多个工具因此我们用for循环遍历state[messages][-1].tool_calls列表。state[messages][-1]取最后一条消息即llm_call节点刚生成的 AI 消息.tool_calls是该消息中模型请求的工具调用列表。每个 tool_call 包含三个关键字段name工具名称、args调用参数、id调用唯一标识。tools_by_name[tool_call[name]]通过名称查找工具对象tool.invoke(tool_call[args])执行工具ToolMessage(content..., tool_call_id...)将执行结果封装为工具消息其中tool_call_id必须与原始 tool_call 的 id 对应这样模型才能将结果与请求匹配起来。第四步是定义条件路由函数。这是 ReAct 循环的核心决策点检查最后一条消息是否包含工具调用如果包含就返回tool_node继续执行工具否则返回END终止循环fromtypingimportLiteralfromlanggraph.graphimportStateGraph,START,ENDdefshould_continue(state:MessagesState)-Literal[tool_node,END]:Decide if we should continue the loop or stop.messagesstate[messages]last_messagemessages[-1]iflast_message.tool_calls:returntool_nodereturnEND这个函数虽然只有几行代码但它承载了 ReAct 循环的核心决策逻辑。返回类型Literal[tool_node, END]不是普通的类型注解它同时服务于两个目的一是让类型检查器验证返回值是否合法二是让 LangGraph 在编译时知道这个条件边可能路由到哪些节点从而正确渲染图结构和检查完整性。last_message.tool_calls是判断依据当模型决定调用工具时AI 消息的tool_calls属性会是一个非空列表包含模型请求的工具调用此时返回tool_node将流程导向工具执行。当模型认为不需要调用工具、直接给出了最终答案时tool_calls为空列表空列表在 Python 中是 falsy 的此时返回END终止图的执行。这个设计体现了 LangGraph 的一个重要理念路由逻辑与业务逻辑分离should_continue只做判断不做执行让图的控制流保持清晰可读。最后一步是构建和编译图。我们将两个节点添加进去用START → llm_call设置入口用条件边llm_call → should_continue → tool_node 或 END建立分支用tool_node → llm_call形成循环然后调用compile()编译为可执行应用# Build workflowagent_builderStateGraph(MessagesState)# Add nodesagent_builder.add_node(llm_call,llm_call)agent_builder.add_node(tool_node,tool_node)# Add edges to connect nodesagent_builder.add_edge(START,llm_call)agent_builder.add_conditional_edges(llm_call,should_continue,{tool_node:tool_node,END:END})agent_builder.add_edge(tool_node,llm_call)# Compile the agentagentagent_builder.compile()这段构建代码揭示了 LangGraph 的图组装的四个标准步骤。第一步StateGraph(MessagesState)创建一个以MessagesState为共享状态的空图这是整张图的画布后续所有操作都在这张画布上进行。第二步通过add_node注册两个处理节点分别负责 LLM 推理和工具执行节点注册后不会立即执行它们只是待命状态等待边来触发。第三步建立边的关系add_edge(START, llm_call)定义了图的入口意味着用户输入首先到达llm_call节点add_conditional_edges则在llm_call节点之后插入了一个分叉路口should_continue函数充当交通指挥第三个参数是路由映射表将函数的返回值tool_node和END分别映射到目标节点add_edge(tool_node, llm_call)形成了关键的循环回路工具执行完毕后一定会回到 LLM 节点重新推理这正是 ReAct 循环的引擎。第四步compile()看似简单实则关键它校验图的完整性例如是否有节点没有被任何边连接、绑定运行时配置、并将图编译为可执行的CompiledStateGraph对象。编译是必须的未编译的图无法执行。至此一个完整的 ReAct Agent 就构建好了。编译这一步做了两件重要的事一是对图结构进行基本检查确保没有孤立节点等二是绑定运行时参数如 checkpointer 和断点。你可以用以下命令可视化图结构fromIPython.displayimportImage,display display(Image(agent.get_graph(xrayTrue).draw_mermaid_png()))执行 Agent 同样简单传入初始消息invoke()会驱动完整的 ReAct 循环直到 LLM 不再调用工具fromlangchain.messagesimportHumanMessage messages[HumanMessage(contentAdd 3 and 4.)]messagesagent.invoke({messages:messages})forminmessages[messages]:m.pretty_print()可视化代码中agent.get_graph(xrayTrue).draw_mermaid_png()生成的是 Mermaid 格式的图结构渲染——xrayTrue参数会展开子图的内部结构让整个流程一览无余。这对于调试复杂工作流非常有用你可以直观地看到节点之间的连接关系和条件分支的走向。执行代码则展示了 LangGraph 最简洁的调用方式agent.invoke({messages: messages})传入初始消息列表LangGraph 自动驱动完整的 ReAct 循环LLM 收到Add 3 and 4后生成add(3, 4)的 tool_call工具节点执行加法返回结果 7LLM 再次收到工具结果后给出最终答案。整个过程对调用者完全透明invoke()返回的是所有消息累积后的最终状态。pretty_print()是 LangChain 消息对象的便捷方法会按角色Human / AI / Tool格式化输出每条消息的内容方便在终端中查看完整的对话历史。理解执行模型Super-Step 与消息传递LangGraph 的底层图算法采用消息传递模型灵感来源于 Google 的 Pregel 系统。图执行被划分为离散的 super-step每个 super-step 中所有被激活的节点并行执行。一个节点在收到入边上的新消息状态更新时变为 active 状态执行其函数然后通过出边将更新后的消息发送给下游节点。在每个 super-step 结束时没有收到新消息的节点投票 halt将自己标记为 inactive。当所有节点都处于 inactive 状态且没有消息在传输中时图执行终止。对于START → llm_call → tool_node → llm_call → ... → END这样的顺序图每个 super-step 只包含一个节点。但如果一个节点有多个出边目标比如条件边返回多个节点名称这些目标节点将在同一个 super-step 中并行执行。你可以通过recursion_limit参数从 v1.0.6 开始默认值为 1000 步来控制最大执行步数LangGraph 还提供了RemainingSteps托管值让你在节点内主动感知剩余步数并实现优雅降级。流式输出实时监控 Agent 的每一步invoke()适用于一次性获取最终结果但在交互式应用或调试场景中你需要实时观测 Agent 的每一步执行。LangGraph 提供了强大的stream()方法支持多种流模式。每个流模式的输出在 v2 格式下统一为StreamPart字典{type: ..., ns: ..., data: ...}通过chunk[type]即可区分。最常用的三种模式是updates流式输出每个节点返回的状态更新values输出每一步后的完整状态快照messages以 token 级别流式输出 LLM 的生成文本适合前端逐字展示。你还可以通过get_stream_writer()在节点内部发送自定义事件配合stream_modecustom来接收——这对于报告进度、发送日志等场景非常实用fromlanggraph.configimportget_stream_writerdefmy_node(state:State):writerget_stream_writer()writer({progress:thinking...})# ... do work ...writer({progress:done})return{result:completed}# 使用多个流模式forchunkingraph.stream(inputs,stream_mode[updates,custom],versionv2):ifchunk[type]updates:fornode_name,state_updateinchunk[data].items():print(fNode{node_name}updated:{state_update})elifchunk[type]custom:print(fCustom:{chunk[data]})get_stream_writer()是 LangGraph 流式体系中最灵活的机制它在节点内部返回一个可调用的 writer 对象你可以随时调用writer(some_dict)将任意数据推送到流中。这里的{progress: thinking...}只是示例实际上你可以推送任何 JSON 可序列化的数据比如中间计算结果、当前步骤的描述、甚至是前端 UI 需要渲染的组件信息。在消费端stream_mode[updates, custom]同时订阅了两种流模式chunk[type]用于区分每个 chunk 的来源updates类型的 chunk 的data字段是一个字典键是节点名称、值是该节点返回的状态更新custom类型的 chunk 的data字段就是你通过writer()推送的原始数据。这种设计让业务数据节点状态更新和 UI 数据进度通知、日志等可以在同一条流中传输但互不干扰。versionv2启用了统一的StreamPart格式无论你订阅了几种模式每个 chunk 的结构都是{type: ..., ns: ..., data: ...}这极大简化了消费端的类型判断逻辑。状态持久化与检查点LangGraph 的持久化层是它区别于简单 Chain 调用的关键特性之一。当你用compile(checkpointerInMemorySaver())编译图时LangGraph 会在每个 super-step 边界自动保存一个检查点checkpoint——这是图状态的完整快照。检查点按 thread由thread_id标识组织同一 thread 内的多次调用会沿着同一条时间线累积状态。这意味着你的 Agent 天然具备记忆能力下一轮对话中传入相同的thread_idAgent 就能访问之前的全部消息历史。fromlanggraph.checkpoint.memoryimportInMemorySaver checkpointerInMemorySaver()graphagent_builder.compile(checkpointercheckpointer)config{configurable:{thread_id:conversation-1}}graph.invoke({messages:[HumanMessage(content北京天气怎么样)]},config)# 第二轮对话自动继承之前的消息历史graph.invoke({messages:[HumanMessage(content我刚才问了什么)]},config)这段代码演示了 LangGraph 持久化最直接的应用跨轮次会话记忆。InMemorySaver()将所有检查点保存在内存中适合开发调试但重启后会丢失。生产环境应使用SqliteSaver或PostgresSaver。关键是config中的thread_id它相当于会话标识符LangGraph 的 checkpointer 以它为主键存储和检索检查点。当第一轮invoke执行完毕后完整的消息历史用户问题、LLM 的 tool_call、工具返回结果、LLM 最终回答都保存在了thread_idconversation-1的检查点链中。第二轮调用使用相同的thread_idLangGraph 会自动从上次的检查点恢复状态因此模型能看到北京天气怎么样这条历史消息并回答我刚才问了什么这类指代性问题。如果不传thread_id或每次使用新的thread_id每轮对话都会从零开始、彼此隔离。持久化还支撑了更高级的特性人工介入human-in-the-loop允许你在图的任意节点通过interrupt()暂停执行等待人工审核或修改状态后再通过Command(resume...)恢复。时间旅行让你可以回放到历史的任意检查点分叉出新的执行路径。故障恢复让你在节点失败后从上一个成功的检查点重试而不需要重新执行已完成的工作。通过graph.get_state(config)可以随时获取最新的状态快照通过graph.get_state_history(config)可以查看整个执行时间线的完整历史。对于本地开发InMemorySaver和SqliteSaver足够满足需求生产环境则可以使用PostgresSaver它支持异步操作和加密序列化。关键 API 速览API用途示例StateGraph(State)以给定的状态 Schema 创建图StateGraph(MessagesState)add_node(name, func)向图注册一个处理节点add_node(agent, call_model)add_edge(from, to)建立无条件固定边add_edge(tools, agent)add_conditional_edges(src, fn)以函数动态决定下一个节点add_conditional_edges(agent, should_continue)add_conditional_edges(src, fn, mapping)同上附带路由映射表add_conditional_edges(agent, fn, {yes: node_a, no: END})compile(checkpointer...)编译校验并绑定运行时参数graph.compile(checkpointerInMemorySaver())invoke(input, config)同步执行整个图agent.invoke({messages: [...]})stream(input, stream_mode...)逐步流式输出执行过程agent.stream(input, stream_modeupdates)add_edge与add_conditional_edges的核心区别在于add_edge建立的是静态、无条件路由——A 节点执行完毕后始终进入 B 节点add_conditional_edges建立的是动态、有条件路由——A 节点执行完毕后调用一个函数根据函数返回值决定进入哪个节点。使用场景上add_edge适合固定的管道式流程如工具执行完后务必返回 LLM 再思考而add_conditional_edges适合有分支决策的流程如LLM 输出包含工具调用则去执行工具否则结束。两者都是从一个源节点出发定义路由——不要混用普通边和条件边指向同一个源节点。AgentExecutor 与 LangGraph 的对比在学习 LangGraph 之前你已经在第 2 周接触了create_agent()和AgentExecutor。两者都实现了 ReAct 循环但在架构哲学上有本质区别。AgentExecutor是一个封装好的 Runnable内部通过一个while循环驱动LLM 思考 → 工具调用 → LLM 再思考的迭代循环逻辑固化在源码中、不可干预。它适合快速原型但当你需要在循环中插入人工审批步骤、根据中间结果动态切换策略、或者并行调用多个工具时AgentExecutor的灵活性就显得捉襟见肘。从状态管理的角度看AgentExecutor的状态散落在 Runnable 的内部闭包中外部几乎无法感知和修改。LangGraph 把同样的循环展开为一个显式的图结构——每一步都是独立可寻址的节点路由逻辑是纯粹的函数状态管理是完全透明的。这意味着你可以随时在图中插入新的节点来增强流程比如在工具执行后增加一个验证节点、在 LLM 调用前增加一个检索节点、或者在某个条件分支挂载一个完全不同的子图。从扩展性角度LangGraph 天然支持子图嵌套、多 Agent 协作和手把手交接handoff是构建复杂多智能体系统的基础设施。维度AgentExecutorLangGraph架构模式黑盒 while 循环显式有向图每步可寻址状态管理隐式闭包外部不可见显式 TypedDict/dataclass可读写流程控制固定的思考-工具循环自定义节点条件边任意拓扑流式输出基础 token 流多模式流状态/更新/token/自定义持久化不支持内置检查点机制支持断点恢复人工介入不支持原生interrupt()暂停-审批-恢复扩展性修改源码或包装 Runnable插入节点/嵌套子图/多 Agent 协作调试可见性低高——每一步状态可查询、可回放综合来看AgentExecutor是 LangChain 早期为简化 Agent 创建而设计的便捷工具适合简单、线性的 Agent 场景LangGraph 则是为生产级、可定制、有状态的 Agent 工作流而生的编排框架。随着 LangChain 生态的演进官方推荐使用create_agent()底层基于 LangGraph作为新的标准入口——这印证了 LangGraph 已经从可选扩展成长为框架的核心基础设施。练习任务参照本文中的代码示例绘制用户提问 → LLM → 工具 → LLM → 回答的 Mermaid 状态图参照本文中的代码实现完整的 ReAct 循环 LangGraph 版本至少包含 3 个不同功能的工具在上述代码基础上添加InMemorySaver持久化验证多轮对话的记忆能力撰写 AgentExecutor vs LangGraph 对比分析重点讨论架构、状态管理和扩展性三个维度考核点 ✅状态图绘制提交 LangGraph 状态图Mermaid/PlantUML 或手绘拍照代码实现提交完整可运行的 LangGraph ReAct 循环代码含持久化对比分析提交 AgentExecutor vs LangGraph 对比表架构/状态管理/扩展性API 掌握口头解释add_conditional_edges和add_edge的区别及使用场景