Week 3 -- Day 1：LangGraph 入门

为什么需要 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=..., goto="target_node")将状态更新和控制流合并在一步中完成。Send则支持 map-reduce 模式，让你从一个节点向多个下游节点发送不同的状态片段，当节点数量在运行前不可知时尤其有用。

构建第一个 LangGraph 工作流

让我们按照最新官方文档的实践方式，从头构建一个基于 ReAct 循环的 Agent。首先确保安装了 LangGraph：

pipinstall-Ulanggraph

第一步是定义工具和模型。我们沿用前两周的模式，定义几个简单的@tool工具，然后使用model.bind_tools(tools)将工具绑定到模型上，让模型在需要时自动生成 tool_call：

fromlangchain.toolsimporttoolfromlangchain.chat_modelsimportinit_chat_modelfromdotenvimportload_dotenv load_dotenv()model=init_chat_model("Qwen/Qwen2.5-7B-Instruct",model_provider="openai")@tooldefmultiply(a:int,b:int)->int:"""Multiply a and b."""returna*b@tooldefadd(a:int,b:int)->int:"""Add a and b."""returna+b@tooldefdivide(a:int,b:int)->float:"""Divide a and b."""returna/b tools=[add,multiply,divide]tools_by_name={tool.name:toolfortoolintools}model_with_tools=model.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(content="You 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:tool=tools_by_name[tool_call["name"]]observation=tool.invoke(tool_call["args"])result.append(ToolMessage(content=observation,tool_call_id=tool_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."""messages=state["messages"]last_message=messages[-1]iflast_message.tool_calls:return"tool_node"returnEND

这个函数虽然只有几行代码，但它承载了 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_builder=StateGraph(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 agentagent=agent_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(xray=True).draw_mermaid_png()))

执行 Agent 同样简单，传入初始消息，invoke()会驱动完整的 ReAct 循环直到 LLM 不再调用工具：

fromlangchain.messagesimportHumanMessage messages=[HumanMessage(content="Add 3 and 4.")]messages=agent.invoke({"messages":messages})forminmessages["messages"]:m.pretty_print()

可视化代码中agent.get_graph(xray=True).draw_mermaid_png()生成的是 Mermaid 格式的图结构渲染——xray=True参数会展开子图的内部结构，让整个流程一览无余。这对于调试复杂工作流非常有用，你可以直观地看到节点之间的连接关系和条件分支的走向。执行代码则展示了 LangGraph 最简洁的调用方式：agent.invoke({"messages": messages})传入初始消息列表，LangGraph 自动驱动完整的 ReAct 循环，LLM 收到"Add 3 and 4"后生成add(3, 4)的 tool_call，工具节点执行加法返回结果 7，LLM 再次收到工具结果后给出最终答案。整个过程对调用者完全透明，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_mode="custom"来接收——这对于报告进度、发送日志等场景非常实用：

fromlanggraph.configimportget_stream_writerdefmy_node(state:State):writer=get_stream_writer()writer({"progress":"thinking..."})# ... do work ...writer({"progress":"done"})return{"result":"completed"}# 使用多个流模式forchunkingraph.stream(inputs,stream_mode=["updates","custom"],version="v2"):ifchunk["type"]=="updates":fornode_name,state_updateinchunk["data"].items():print(f"Node{node_name}updated:{state_update}")elifchunk["type"]=="custom":print(f"Custom:{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 数据（进度通知、日志等）可以在同一条流中传输但互不干扰。version="v2"启用了统一的StreamPart格式，无论你订阅了几种模式，每个 chunk 的结构都是{"type": ..., "ns": ..., "data": ...}，这极大简化了消费端的类型判断逻辑。

状态持久化与检查点

LangGraph 的持久化层是它区别于简单 Chain 调用的关键特性之一。当你用compile(checkpointer=InMemorySaver())编译图时，LangGraph 会在每个 super-step 边界自动保存一个检查点（checkpoint）——这是图状态的完整快照。检查点按 thread（由thread_id标识）组织，同一 thread 内的多次调用会沿着同一条时间线累积状态。这意味着你的 Agent 天然具备"记忆"能力：下一轮对话中传入相同的thread_id，Agent 就能访问之前的全部消息历史。

fromlanggraph.checkpoint.memoryimportInMemorySaver checkpointer=InMemorySaver()graph=agent_builder.compile(checkpointer=checkpointer)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_id="conversation-1"的检查点链中。第二轮调用使用相同的thread_id，LangGraph 会自动从上次的检查点恢复状态，因此模型能看到"北京天气怎么样？"这条历史消息并回答"我刚才问了什么？"这类指代性问题。如果不传thread_id或每次使用新的thread_id，每轮对话都会从零开始、彼此隔离。

持久化还支撑了更高级的特性，人工介入（human-in-the-loop）允许你在图的任意节点通过interrupt()暂停执行，等待人工审核或修改状态后再通过Command(resume=...)恢复。时间旅行让你可以回放到历史的任意检查点，分叉出新的执行路径。故障恢复让你在节点失败后从上一个成功的检查点重试，而不需要重新执行已完成的工作。通过graph.get_state(config)可以随时获取最新的状态快照，通过graph.get_state_history(config)可以查看整个执行时间线的完整历史。对于本地开发，InMemorySaver和SqliteSaver足够满足需求，生产环境则可以使用PostgresSaver，它支持异步操作和加密序列化。

关键 API 速览

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），是构建复杂多智能体系统的基础设施。

综合来看，AgentExecutor是 LangChain 早期为简化 Agent 创建而设计的便捷工具，适合简单、线性的 Agent 场景；LangGraph 则是为生产级、可定制、有状态的 Agent 工作流而生的编排框架。随着 LangChain 生态的演进，官方推荐使用create_agent()（底层基于 LangGraph）作为新的标准入口——这印证了 LangGraph 已经从可选扩展成长为框架的核心基础设施。

练习任务

1. 参照本文中的代码示例，绘制"用户提问 → LLM → 工具 → LLM → 回答"的 Mermaid 状态图
2. 参照本文中的代码实现完整的 ReAct 循环 LangGraph 版本，至少包含 3 个不同功能的工具
3. 在上述代码基础上添加InMemorySaver持久化，验证多轮对话的记忆能力
4. 撰写 AgentExecutor vs LangGraph 对比分析，重点讨论架构、状态管理和扩展性三个维度

考核点 ✅

1. 状态图绘制：提交 LangGraph 状态图（Mermaid/PlantUML 或手绘拍照）
2. 代码实现：提交完整可运行的 LangGraph ReAct 循环代码（含持久化）
3. 对比分析：提交 AgentExecutor vs LangGraph 对比表（架构/状态管理/扩展性）
4. API 掌握：口头解释add_conditional_edges和add_edge的区别及使用场景