用 LangChain 的AgentExecutor写过复杂 Agent 的人大概都遇过这种情况Agent 在第7步突然报错你想从第5步重试一下但做不到只能从头跑白白烧了一大串 LLM 调用费。或者你想在某个关键决策节点插入人工审核发现 AgentExecutor 根本没提供 hook 点只能 hack 进去。这些不是 bug是架构决策的代价。而 LangGraph 就是 LangChain 团队对这套决策的全面推翻。有意思的是LangGraph 官方现在已经把 AgentExecutor 标为遗留组件所有新 Agent 推荐用 LangGraph 构建。这不是小修小补是直接推了一套新框架。翻源码之后发现它的设计决策远比带状态的 DAG这个描述有意思——它把 Actor 模型、事件驱动和检查点持久化在架构层强耦合在了一起不是后来打补丁加进去的。本文顺着源码捋 LangGraph 的内部实现重点讲三件事图执行模型怎么跑、状态怎么管、Checkpoint 机制怎么做到开箱即用。最后给出一个直接的使用建议。AgentExecutor 的根本缺陷在哪LangChain 早期的AgentExecutor是个很直接的 while 循环# 简化版 AgentExecutor 核心逻辑langchain/agents/agent.py while True: action agent.plan(intermediate_steps, **kwargs) if isinstance(action, AgentFinish): return action.return_values # 执行工具 observation tool.run(action.tool_input) intermediate_steps.append((action, observation)) # 检查迭代上限 if len(intermediate_steps) max_iterations: return {output: Agent stopped due to iteration limit}三个硬伤•状态是个无结构列表没有 schema中途做分支判断很费劲•没有持久化一旦中断就从头来human-in-the-loop 要自己实现•执行是同步串行的工具调用没法并发LangGraph 的答案是引入图。但这里说的图不是普通的 DAG——更接近有限状态机FSM和数据流图的混合体每条边可以是条件边conditional edge每个节点是一个状态变换函数。StateGraph 的核心Channel 是什么先看StateGraph的定义langgraph/graph/state.pyclass StateGraph(Graph): def __init__(self, state_schema: Type[Any], ...): super().__init__() self.schema state_schema self.nodes: dict[str, RunnableCallable] {} self.edges: set[tuple[str, str]] set() self._conditional_edges: dict {} # 关键每个状态字段都会变成一个 Channel 对象 self.channels: dict[str, BaseChannel] {} self._schema_to_channels(state_schema)注意最后两行_schema_to_channels会把你定义的 State 类的每个字段都转成一个 Channel 对象。Channel 是什么把它想象成 Google Docs 的协同编辑合并策略——两个人同时修改同一段文字系统得决定怎么合并是后者覆盖前者还是追加到末尾还是取最大值。Channel 就是那套合并规则的容器专门处理多个节点并发写同一状态字段时的冲突# langgraph/channels/base.py class BaseChannel(ABC): abstractmethod def update(self, values: Sequence[Any]) - bool: 接收一批并发写入决定怎么合并返回是否有变化 ... abstractmethod def get(self) - Any: 拿当前值 ... # Append 语义messages 字段默认用这个 # 对应 TypedDict 里写 Annotated[list, operator.add] class BinaryOperatorAggregate(BaseChannel): def __init__(self, typ, operator): self.operator operator # operator.add 就是列表追加 self.value typ() def update(self, values): if values: new_val reduce(self.operator, values, self.value) if new_val ! self.value: self.value new_val return True return False这就是为什么你在 TypedDict State 里写Annotated[list, operator.add]messages 就会自动追加而不是覆盖——Channel 在幕后帮你做了归约不需要手动 append。执行引擎Pregel 超步是接力跑LangGraph 的执行引擎叫Pregel直接取自 Google 那篇图计算论文。核心文件是langgraph/pregel/__init__.py主循环逻辑是这样的# 简化的 Pregel 执行主循环 async def astream(self, input, config, ...): # 1. 初始化 channels从 checkpoint 恢复或全新初始化 channels self._prepare_channels(config, checkpoint) # 2. 写入初始输入 apply_writes(checkpoint, channels, input_writes) # 3. Pregel 超步循环 for step in range(max_steps): # 找出所有就绪的节点输入 channel 有新数据的节点 next_tasks self._prepare_next_tasks( checkpoint, channels, config, step ) if not next_tasks: break # 没有可执行任务结束 # 并发执行当前 superstep 的所有就绪任务 async with TaskGroup() as tg: for task in next_tasks: tg.create_task( execute_task_and_write_results(task, channels) ) # 保存 checkpoint checkpoint create_checkpoint(channels, next_tasks) await checkpointer.aput(config, checkpoint) # yield 中间状态.stream() 能看到中间步骤的原因 yield channels_to_output(channels)Pregel 的超步superstep可以类比接力跑的交接棒——每一棒只能接上一棒传来的棒不能跨棒传递这保证了数据不会乱。具体来说节点只能读上一个 superstep 写入的 Channel 值当前 superstep 里多个节点并发写同一 Channel写入的结果要等到本轮结束才对外可见下一个 superstep 才能读到。这个设计的关键副产品TaskGroup会把当前 step 里所有就绪的节点并发跑工具并行调用是原生支持的不需要你写任何并发代码。条件边路由逻辑怎么变成图结构LangGraph 最常用的模式是LLM 决定下一步跳哪个节点这靠的是条件边。源码里的实现分两层# 添加条件边时内部存储为 Branch 对象 def add_conditional_edges(self, source, path, path_mapNone): self._add_edge(source, Branch(path, path_map, thenNone))# Branch 在执行时调用路由函数把目标节点名写进 Send channel class Branch: def run(self, writer, reader, config): value reader(self.path) # 调用路由函数得到目标节点名 destinations self._route(value, self.path_map) for dest in destinations: writer([Send(dest, reader())]) # 往目标节点的输入 channel 写消息关键在Send这个对象——它不是直接跳转而是往目标节点的输入 channel 里写了一条消息。下一个 superstep 开始时_prepare_next_tasks扫描所有 channel发现某节点的输入 channel 有数据就把它加入就绪队列。这个设计的好处路由决策和执行是解耦的。路由函数可以返回多个节点名LangGraph 会并发执行它们——这是实现 map-reduce 模式一个任务拆成多个子任务并发处理再聚合的直接基础。Checkpoint把从第5步重试变成现实Checkpoint 是我认为 LangGraph 里工程做得最扎实的部分没有之一。它解决的是 AgentExecutor 最让人头疼的两个问题中途暂停等人审核、出错后回滚到历史状态重试。核心数据结构dataclass class Checkpoint: v: int # schema 版本号 id: str # UUID唯一标识这个检查点 ts: str # 时间戳 channel_values: dict[str, Any] # 所有 channel 的序列化值快照 channel_versions: dict[str, int] # 每个 channel 当前的版本号 versions_seen: dict[str, dict] # 各节点上次执行时看到的版本号 pending_sends: list[SendProtocol] # 等待发送的消息用于恢复 dataclass class CheckpointMetadata: source: Literal[input, loop, update, fork] step: int # 第几个 superstep writes: dict # 本次写入了什么 parents: dict # 父 checkpoint 的 id支持分支回溯versions_seen是防重放的关键。把它想象成每个节点自己的进度条——节点记录自己上次处理时各 Channel 的版本号下次执行前检查有没有比我进度条更新的版本有才执行没有就跳过。这保证了即使 checkpoint 被恢复多次节点也不会重复执行不会重复调 LLM。Human-in-the-loop 的实现方式出乎意料地简单# 编译时声明在 human_review 节点之前暂停 graph builder.compile( checkpointerMemorySaver(), interrupt_before[human_review] ) # 执行到 interrupt_before 节点时Pregel 抛出 GraphInterrupt 异常 # 当前 checkpoint 已保存调用方 catch 到后可以做任何事展示给人看、等待输入等 # 人工审核完用同一个 thread_id 重新 invoke # Pregel 从最新 checkpoint 恢复从暂停位置继续不会重跑之前的步骤 result graph.invoke( Command(resumehuman_feedback), # 把人的反馈注入 config{configurable: {thread_id: session_1}} )源码里_prepare_next_tasks在发现即将执行的任务名在interrupt_before列表里时往任务上打INTERRUPT标记主循环 raiseGraphInterrupt。恢复时 Pregel 读最新 checkpoint通过versions_seen判断哪些节点已经跑过、哪些没跑只执行没跑过的。SubGraph图里跑图模块化 AgentLangGraph 支持在节点里嵌套另一个图SubGraph这是构建复杂多 Agent 系统的基础# 子图编译后直接作为节点加入父图 parent_graph StateGraph(ParentState) parent_graph.add_node(subagent, child_graph.compile()) # Pregel 执行子图节点时 # 1. 父图 State 中对应字段映射为子图的输入 # 2. 子图内部走完整的 Pregel 超步循环嵌套执行 # 3. 子图的最终输出映射回父图的 State有一个细节值得注意子图有独立的 checkpoint namespace。源码里通过在 config 的configurable[checkpoint_ns]字段加上子图名来区分父子图的历史记录是隔离的time-travel 时可以分别回滚。这意味着你可以单独重跑某个子 Agent而不影响父图的状态。四种 stream_mode 的实现差异LangGraph 的.stream()支持四种模式搞清楚差异能省很多调试时间stream_modeyield 的内容适用场景values每个 superstep 后的完整 State 快照需要完整状态做 UI 渲染updates每个节点输出的增量 dict监控每个节点的输出日志debug任务开始/结束事件 完整数据调试、接 LangSmith tracemessagesLLM token 级别的流式输出前端实时展示打字效果messages模式是最复杂的源码里用了一个StreamMessagesHandler回调在 LLM 调用时 hook 进去把 token 流实时转出来同时维护 message 的 id 映射防止子图里的 message 和父图冲突。这块在langgraph/pregel/io.py里有自定义流式处理需求可以着重看这里。一个最小 ReAct Agent30 行背后做了什么from typing import Annotated, TypedDict import operator from langgraph.graph import StateGraph, END from langchain_core.messages import BaseMessage, HumanMessage # 1. 定义状态messages 用 add 语义追加而非覆盖 class AgentState(TypedDict): messages: Annotated[list[BaseMessage], operator.add] # 2. 定义节点 def call_model(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} def call_tools(state: AgentState): last_msg state[messages][-1] # 并行执行所有工具调用Pregel TaskGroup 原生支持 results tool_executor.batch(last_msg.tool_calls) return {messages: results} # 3. 路由函数 def should_continue(state: AgentState): last_msg state[messages][-1] if last_msg.tool_calls: return tools return END # 4. 组装图 builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, call_tools) builder.set_entry_point(agent) builder.add_conditional_edges(agent, should_continue) builder.add_edge(tools, agent) # 5. 编译加 checkpointer 支持持久化 human-in-the-loop graph builder.compile(checkpointerMemorySaver()) result graph.invoke( {messages: [HumanMessage(帮我查一下今天天气)]}, config{configurable: {thread_id: session_1}} )这 30 行代码背后LangGraph 帮你处理了Pregel 超步循环、Channel 合并归约、Checkpoint 保存、并发工具调用、流式输出、会话隔离。你只定义节点逻辑和路由规则。LangGraph 的局限性不做无脑安利说清楚几个需要注意的地方•序列化耦合Checkpoint 需要把所有 Channel 的值序列化State 里不能放不可序列化的对象数据库连接、文件句柄。实践中要放 ID在节点里按需重建资源•调试体验有限图执行的中间状态虽然可以通过 stream 获取但原生没有可视化。接 LangSmith 能解决但不想引入 observability 服务的话纯靠日志调试还是费劲•图结构静态目前图在 compile 时确定不支持运行时动态增减节点。要做根据情况动态扩展图这类需求只能通过 SubGraph 条件路由模拟不够直接•版本迭代快LangGraph API 变化比较频繁0.2 到 0.3 有几处 breaking changeCommand对象的引入就是一个生产环境要锁好版本到底什么时候用 LangGraph什么时候不用给一个直接的判断省掉调研时间直接用 LangGraph• 需要持久化对话状态、支持中途暂停/恢复的 Agenthuman-in-the-loop 场景• 需要多工具并发调用或者实现 map-reduce 模式一个任务拆成多个子任务并发处理• 复杂多 Agent 协作系统需要模块化和状态隔离• 需要 time-travel 调试回到历史某个状态重放LCEL 够了不需要 LangGraph• 无状态的单轮 LLM 调用或者简单的 chain提示词 → LLM → 解析• 不需要持久化、不需要 human-in-the-loop 的简单 ReAct Agent• 快速原型不想引入图结构的概念复杂度应该考虑更重的调度框架Temporal/Prefect• 任务执行时间超长小时级需要分布式队列和可靠的重试机制• 不是 AI 工作流而是通用的业务流程编排AI 只是其中一步• 需要细粒度的资源控制、并发限流、跨服务事务一句话总结我的判断做 Agent 系统LangGraph 是目前最务实的选择没有之一——但它是一个代码级抽象不是拖拽式平台需要你真的理解它的执行模型才能用好。 Dify/Coze 的 workflow 更易上手但灵活性有天花板LangGraph 上手成本高但一旦搞懂了能做到的事情边界远大于前者。下一步值得看的方向LangGraph Server 的 HTTP 接口层实现——它是怎么把 Pregel 引擎包成对齐 OpenAI Assistants API 的服务的以及 Multi-Agent Supervisor 模式在源码层面的具体实现这两块是从懂 LangGraph到用好 LangGraph的关键跨越。如果觉得有帮助欢迎分享给同样在研究 AI 工具链的朋友。