# LangGraph 源码探索参考 > 探索对象：`D:\projects\agent-management\langgraph\` 目录，LangChain Inc. 开发的 **LangGraph 官方开源项目**（MIT 许可证，公网仓库 `github.com/langchain-ai/langgraph`）。 > > 探索目的：理解 LangGraph 的 DAG / 图计算核心模型（StateGraph + Pregel + Channel + Checkpoint），为本仓库（agent-management）的工作流引擎演进提供架构参考与边界 case 验证依据。 > > 应用方式：本仓库实现的"工作流引擎"在概念上是对 LangGraph Pregel 模型的重新实现（Java 版）。本份源码可作为： > 1. 架构参考——Pregel BSP 三阶段循环、Channel 抽象、Reducer 合并、Checkpoint 时间旅行 > 2. 边界 case 验证——`apply_writes` 的 Reducer 应用顺序、`versions_seen` 防重复触发、`pending_writes` 重放 > 3. 演进借鉴参考——详见 `workflow-langgraph-dify-comparison-reference.md` --- ## 1. 目录完整结构 ### 1.1 顶层结构 ``` D:\projects\agent-management\langgraph\ ├── .git/ # 独立 git 仓库（与主项目分离） ├── .github/ # CI 配置、Issue 模板 ├── .gitignore / .markdownlint.json ├── AGENTS.md / CLAUDE.md # 仓库级开发规范 ├── LICENSE # MIT ├── Makefile # 顶层格式化/lint/test 入口 ├── README.md # 项目总览 ├── docs/ # 文档目录（仅含 redirect 工具） │ ├── generate_redirects.py │ ├── llms.txt │ └── redirects.json ├── examples/ # Jupyter 示例（大部分已迁移到 docs） │ ├── human_in_the_loop/ multi_agent/ chatbots/ code_assistant/ │ ├── customer-support/ extraction/ lats/ llm-compiler/ │ ├── plan-and-execute/ rag/ reflection/ reflexion/ │ ├── rewoo/ self-discover/ tutorials/ usaco/ │ ├── web-navigation/ chatbot-simulation-evaluation/ delta-channel-dump/ │ ├── react-agent-from-scratch.ipynb │ ├── react-agent-structured-output.ipynb │ ├── subgraph.ipynb # 已迁移，仅留归档提示 │ ├── tool-calling.ipynb │ └── run-id-langsmith.ipynb └── libs/ # 核心：monorepo 子项目 ├── langgraph/ # 核心运行时框架 ├── prebuilt/ # 高层封装（ReAct Agent、ToolNode） ├── checkpoint/ # Checkpoint 基类/接口 ├── checkpoint-sqlite/ # SQLite Checkpoint 实现 ├── checkpoint-postgres/ # Postgres Checkpoint 实现 ├── checkpoint-conformance/ # Checkpoint 一致性测试套件 ├── cli/ # LangGraph CLI（部署/开发服务器） ├── sdk-py/ # Python SDK（连接 LangGraph Server） └── sdk-js/ # JS/TS SDK ``` ### 1.2 libs/langgraph/ 核心模块 ``` libs/langgraph/ ├── pyproject.toml / Makefile / README.md / uv.lock / LICENSE ├── bench/ # 性能基准 ├── tests/ # 测试套件 └── langgraph/ ├── version.py / typing.py / warnings.py / py.typed ├── constants.py # 公开常量：START="__start__", END="__end__" ├── errors.py # GraphInterrupt / GraphRecursionError 等 ├── types.py (35KB) # Send / Command / Interrupt / StateSnapshot / RetryPolicy / CachePolicy / TimeoutPolicy ├── runtime.py # Runtime / RunControl / ExecutionInfo ├── callbacks.py / config.py ├── _internal/ # 私有工具：序列化、pydantic 适配、超时、scratchpad │ ├── _config.py _constants.py _fields.py _future.py _pydantic.py │ ├── _queue.py _replay.py _retry.py _runnable.py _serde.py _timeout.py ├── graph/ # StateGraph 用户层 API │ ├── __init__.py │ ├── state.py (76KB) # StateGraph / CompiledStateGraph │ ├── _node.py # StateNode 协议 / StateNodeSpec │ ├── _branch.py # BranchSpec 条件边 │ ├── message.py (15KB) # MessagesState / MessageGraph / add_messages reducer │ └── ui.py ├── pregel/ # Pregel 执行引擎 │ ├── __init__.py │ ├── main.py (175KB) # Pregel 类（invoke/stream/astream） │ ├── _algo.py (52KB) # apply_writes / prepare_next_tasks / should_interrupt │ ├── _loop.py (82KB) # PregelLoop / SyncPregelLoop / AsyncPregelLoop │ ├── _runner.py (37KB) # PregelRunner 并发任务执行 │ ├── _read.py # ChannelRead / PregelNode │ ├── _write.py # ChannelWrite / ChannelWriteEntry │ ├── _call.py / _executor.py / _retry.py / _io.py │ ├── _tools.py / _messages.py / _draw.py / _validate.py / _utils.py │ ├── _checkpoint.py / _config.py / _log.py / debug.py │ ├── protocol.py # PregelProtocol 抽象 │ ├── remote.py (49KB) # RemotePregel（连接远程 LangGraph API） │ ├── _remote_run_stream.py │ └── types.py ├── channels/ # 状态通道（Channel）实现 │ ├── __init__.py # 公开导出 │ ├── base.py # BaseChannel 抽象基类 │ ├── last_value.py # LastValue / LastValueAfterFinish │ ├── binop.py # BinaryOperatorAggregate（用于 reducer） │ ├── topic.py # Topic（PubSub） │ ├── ephemeral_value.py # EphemeralValue（步内有效） │ ├── named_barrier_value.py # NamedBarrierValue（join 同步） │ ├── any_value.py / delta.py / untracked_value.py ├── managed/ # Managed Value（动态注入的非持久化值） │ ├── __init__.py / base.py / is_last_step.py ├── func/ # 函数式 API（task / entrypoint） │ └── __init__.py (23KB) ├── stream/ # 流式输出框架 │ ├── _types.py / _mux.py / run_stream.py / stream_channel.py / transformers.py / _convert.py └── utils/ # 公共工具（RunnableCallable 等） └── __init__.py (24KB) ``` ### 1.3 libs/checkpoint/ 结构 ``` libs/checkpoint/langgraph/checkpoint/ ├── base/__init__.py (30KB) # BaseCheckpointSaver / Checkpoint / CheckpointTuple / CheckpointMetadata ├── memory/ # InMemorySaver（开发用） └── serde/ # 序列化协议（JsonPlusSerializer / EncryptedSerializer） ``` ### 1.4 libs/prebuilt/ 结构 ``` libs/prebuilt/langgraph/prebuilt/ ├── __init__.py # 导出 create_react_agent / ToolNode / tools_condition 等 ├── chat_agent_executor.py (40KB) # create_react_agent 实现 ├── tool_node.py (79KB) # ToolNode / InjectedState / tools_condition ├── interrupt.py # HumanInterrupt / HumanResponse（已弃用，迁移到 langchain.agents） ├── tool_validator.py # ValidationNode └── _tool_call_transformer.py / _tool_call_stream.py ``` --- ## 2. 核心 DAG / StateGraph 实现 ### 2.1 节点定义方式 **核心位置**：`libs/langgraph/langgraph/graph/state.py:667` `add_node()` + `libs/langgraph/langgraph/graph/_node.py` 节点是一个 Python 可调用对象（函数、lambda 或 `Runnable`），签名支持 9 种形式： ```python # libs/langgraph/langgraph/graph/_node.py:22-87 class _Node(Protocol): # 最简形式 def __call__(self, state: NodeInputT) -> Any: ... class _NodeWithConfig(Protocol): # 注入 config def __call__(self, state, config: RunnableConfig) -> Any: ... class _NodeWithWriter(Protocol): # 注入流式写入器 def __call__(self, state, *, writer: StreamWriter) -> Any: ... class _NodeWithStore(Protocol): # 注入 KV store def __call__(self, state, *, store: BaseStore) -> Any: ... class _NodeWithRuntime(Protocol): # 注入 Runtime（v1.0 新增） def __call__(self, state, *, runtime: Runtime[ContextT]) -> Any: ... # 还有 4 种组合（Writer+Store / Config+Writer / Config+Store / Config+Writer+Store） ``` `add_node` 关键参数： - `node` / `action`: 节点逻辑（如果只传函数，则函数名作节点名） - `defer`: 是否延迟到运行结束前执行（用于优化顺序） - `input_schema`: 该节点的私有输入 schema（默认是图 state schema） - `retry_policy`: 重试策略（可叠加多个） - `cache_policy`: 结果缓存策略 - `error_handler`: 节点级错误处理器 - `destinations`: 显式声明可路由目标（仅用于绘图，不影响执行） - `timeout` / `trace_policy`: 超时与 trace 策略 `StateNodeSpec` (dataclass) 是节点的编译期表示： ```python # libs/langgraph/langgraph/graph/_node.py:91 @dataclass(slots=True) class StateNodeSpec(Generic[NodeInputT, ContextT]): runnable: StateNode[NodeInputT, ContextT] metadata: dict[str, Any] | None input_schema: type[NodeInputT] retry_policy: RetryPolicy | Sequence[RetryPolicy] | None cache_policy: CachePolicy | None is_error_handler: bool = False error_handler_node: str | None = None ends: tuple[str, ...] | dict[str, str] | None = EMPTY_SEQ defer: bool = False timeout: TimeoutPolicy | None = None trace_policy: TracePolicy | None = None ``` ### 2.2 边的定义方式（普通边、条件边、循环边） **核心位置**：`libs/langgraph/langgraph/graph/state.py:928` `add_edge`, `:982` `add_conditional_edges` #### 普通边（含 join 多入边） ```python # state.py:928 def add_edge(self, start_key: str | list[str], end_key: str) -> Self: ... if isinstance(start_key, str): self.edges.add((start_key, end_key)) # 单入边 else: self.waiting_edges.add((tuple(start_key), end_key)) # 多入边：等待全部完成 ``` **关键设计**：当 `start_key` 是 list 时，会进入 `waiting_edges`，编译时生成一个 `NamedBarrierValue` 通道实现 join 同步（见 `attach_edge` 第 1551-1575 行）： ```python # state.py:1551 attach_edge def attach_edge(self, starts: str | Sequence[str], end: str) -> None: if isinstance(starts, str): # 单入边：直接写入目标的 "branch:to:{end}" 触发通道 self.nodes[starts].writers.append( ChannelWrite((ChannelWriteEntry(_CHANNEL_BRANCH_TO.format(end), None),)) ) elif end != END: channel_name = f"join:{'+'.join(starts)}:{end}" # 多入边：使用 NamedBarrierValue 屏障，所有 starts 都写入后才触发 if self.builder.nodes[end].defer: self.channels[channel_name] = NamedBarrierValueAfterFinish(str, set(starts)) else: self.channels[channel_name] = NamedBarrierValue(str, set(starts)) self.nodes[end].triggers.append(channel_name) for start in starts: self.nodes[start].writers.append( ChannelWrite((ChannelWriteEntry(channel_name, start),)) ) ``` #### 条件边（含动态分发） ```python # state.py:982 def add_conditional_edges( self, source: str, path: Callable[..., Hashable | Sequence[Hashable]] | Runnable[...], path_map: dict[Hashable, str] | list[str] | None = None, ) -> Self: ... path = coerce_to_runnable(path, name=None, trace=True) name = path.name or "condition" self.branches[source][name] = BranchSpec.from_path(path, path_map, True) ``` `BranchSpec` (`libs/langgraph/langgraph/graph/_branch.py:83`) 是条件边编译期表示，核心字段： ```python class BranchSpec(NamedTuple): path: Runnable[Any, Hashable | list[Hashable]] # 路由函数（可同步/异步） ends: dict[Hashable, str] | None # path_map：返回值 -> 目标节点 input_schema: type[Any] | None = None # 推断的输入 schema ``` **运行期路由**（`_branch.py:192 _finish`）：分支函数返回 `Hashable | list[Hashable] | Send`，结果可以是： - 单值（路由到单节点） - 列表（**同时路由到多个节点**，实现 fan-out 并行） - `Send` 对象（动态 push 任务到指定节点，带自定义 state，见 2.5） **循环边**：LangGraph 不区分有向边和循环边——`add_edge("B", "A")` 即可创建从 B 回到 A 的循环。这是 LangGraph 的核心特性（参见 README 第 82 行：受 Pregel/Beam/NetworkX 启发）。循环终止条件： - 节点路由到 `END` - 步数超过 `recursion_limit`（默认 25，抛 `GraphRecursionError`） - 节点路由到当前没有任何更新的通道（自然终止） #### `set_entry_point` / `set_conditional_entry_point` / `set_finish_point` ```python # state.py:1079, 1092, 1116 def set_entry_point(self, key): return self.add_edge(START, key) def set_conditional_entry_point(self, path, path_map=None): return self.add_conditional_edges(START, path, path_map) def set_finish_point(self, key): return self.add_edge(key, END) ``` `START` (`__start__`) 和 `END` (`__end__`) 是预定义的虚拟节点（`libs/langgraph/langgraph/constants.py:28-31`）。 #### `add_sequence` (v1.0 新增) `state.py:1032` 一次性把多个节点串成线性链。 --- ### 2.3 State 状态传递机制 #### 2.3.1 Schema -> Channel 映射 **核心位置**：`libs/langgraph/langgraph/graph/state.py:1815` `_get_channels` 用户用 `TypedDict` + `Annotated[type, reducer]` 定义 state schema，编译时被转换为 channel 映射： ```python # state.py:1815 def _get_channels(schema): type_hints = get_type_hints(schema, include_extras=True) all_keys = {name: _get_channel(name, typ) for name, typ in type_hints.items()} return ( {k: v for k, v in all_keys.items() if isinstance(v, BaseChannel)}, {k: v for k, v in all_keys.items() if is_managed_value(v)}, type_hints, ) # state.py:1850 _get_channel def _get_channel(name, annotation, *, allow_managed=True): if manager := _is_field_managed_value(name, annotation): return manager elif channel := _is_field_channel(annotation): ... # 显式指定 channel 实例 elif channel := _is_field_binop(annotation): ... # Annotated[list, reducer_func] fallback: LastValue = LastValue(annotation) # 默认：LastValue（覆盖） return fallback ``` **State 字段的三种语义**： | 注解形式 | 推导出的 Channel | 行为 | |---|---|---| | `x: int` | `LastValue(int)` | 每步只能有一个写入，覆盖 | | `x: Annotated[list, operator.add]` | `BinaryOperatorAggregate(list, add)` | **Reducer**：多节点写入通过二元运算合并 | | `x: Annotated[int, SomeChannel()]` | 用户指定的 Channel 实例 | 自定义合并语义 | #### 2.3.2 节点读写 state **核心位置**：`libs/langgraph/langgraph/graph/state.py:1444` `attach_node` 每个节点编译为一个 `PregelNode`（actor），具备： - `triggers`: 触发该节点的 channel 列表 - `channels`: 读取的 state keys - `writers`: 一组 `ChannelWrite`，把节点返回值写回对应 channel 节点返回值通过 `_get_updates` (`state.py:1456`) 提取为 `(key, value)` 元组序列，支持： - `dict` 直接返回 - `Command` 对象（带 `update` / `goto` / `resume`，见 2.6） - `list[Command]` 多命令组合 - 带 `Annotated` 注解的 dataclass / pydantic model **节点输出写入使用 `ChannelWriteTupleEntry` 动态映射**（`state.py:1495-1505`）： ```python write_entries = ( ChannelWriteTupleEntry(mapper=_get_root if output_keys == ["__root__"] else _get_updates), ChannelWriteTupleEntry(mapper=_control_branch, static=_control_static(node.ends) if node.ends else None), ) ``` `_control_branch` (`state.py:1749`) 处理节点返回的 `Command(goto=...)`，把控制信号写到 `branch:to:{target}` 通道或 `TASKS`（用于 `Send`）。 --- ### 2.4 Reducer 机制 Reducer 通过 `BinaryOperatorAggregate` channel 实现（`libs/langgraph/langgraph/channels/binop.py:65`）： ```python class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]): def __init__(self, typ, operator: Callable[[Value, Value], Value]): ... self.operator = operator def update(self, values: Sequence[Value]) -> bool: if not values: return False if self.value is MISSING: self.value = values[0]; values = values[1:] seen_overwrite = False for value in values: is_overwrite, overwrite_value = _get_overwrite(value) # 支持 Overwrite 语义 if is_overwrite: self.value = overwrite_value # 强制覆盖（绕过 reducer） else: self.value = self.operator(self.value, value) # 应用 reducer return True ``` **典型用法**： ```python from typing import Annotated import operator class State(TypedDict): messages: Annotated[list, operator.add] # 多节点 append，自动合并 count: Annotated[int, lambda a, b: a + b] ``` `Overwrite` (`binop.py:31`) 是特殊值，可让某次更新绕过 reducer 直接覆盖。 --- ### 2.5 Pregel 调度与执行模型（核心） LangGraph 灵感来自 Google **Pregel 论文**（`README.md:82`）和 **Apache Beam** 的 **BSP（Bulk Synchronous Parallel）** 模型。 #### 2.5.1 Pregel 类 **核心位置**：`libs/langgraph/langgraph/pregel/main.py:450` ```python class Pregel(PregelProtocol[StateT, ContextT, InputT, OutputT], Generic[...]): """Pregel combines actors and channels into a single application. Each step consists of three phases: - Plan: 决定本步执行哪些 actor - Execution: 并行执行所有选中的 actor，直到全部完成或失败 - Update: 用 actor 写入的值更新 channel Repeat until no actors are selected or max steps reached. """ ``` `__init__` (`main.py:758`) 关键字段： - `nodes: dict[str, PregelNode]` — 所有 actor - `channels: dict[str, BaseChannel]` — 所有 channel（包含自动注入的 `TASKS` 通道） - `input_channels` / `output_channels` / `stream_channels` - `checkpointer` / `store` / `cache` - `interrupt_before_nodes` / `interrupt_after_nodes` - `trigger_to_nodes: Mapping[str, Sequence[str]]` — channel -> 受影响节点的反向索引（优化） **关键约束**：`TASKS` 通道被强制为 `Topic(Send, accumulate=False)`（`main.py:804-809`），用于承载动态 push 任务。 #### 2.5.2 PregelLoop — 主循环 **核心位置**：`libs/langgraph/langgraph/pregel/_loop.py:158` `PregelLoop` 基类（同步版 `SyncPregelLoop:1469`，异步版 `AsyncPregelLoop:1722`） 每个 step 由 `tick()` + `after_tick()` 两个方法完成（`_loop.py:599, 683`）： ```python def tick(self) -> bool: """Execute a single iteration of the Pregel loop.""" # 1. 检查步数限制 if self.step > self.stop: self.status = "out_of_steps"; return False # 2. Plan 阶段：决定本步要执行哪些任务 self.tasks = prepare_next_tasks( self.checkpoint, self.checkpoint_pending_writes, self.nodes, self.channels, self.managed, self.config, self.step, self.stop, for_execution=True, ... ) # 3. 没有任务 -> 结束 if not self.tasks: self.status = "done"; return False # 4. drain 请求（优雅关闭） if self.control is not None and self.control.drain_requested: self.status = "draining"; return False # 5. 恢复历史写入（resume 场景） if not self.is_replaying and self.checkpoint_pending_writes: self._reapply_writes_to_succeeded_nodes(self.tasks) self._resume_error_handlers_if_applicable() # 6. interrupt_before 检查（Human-in-the-loop） if self.interrupt_before and should_interrupt(...): self.status = "interrupt_before"; raise GraphInterrupt() return True def after_tick(self) -> None: # Update 阶段：应用本步所有任务的写入 writes = [w for t in self.tasks.values() for w in t.writes] self.updated_channels = apply_writes( self.checkpoint, self.channels, self.tasks.values(), self.checkpointer_get_next_version, self.trigger_to_nodes, ) # 发出 values 流 self._emit("values", map_output_values, self.output_keys, writes, self.channels) # 清理 pending writes self.checkpoint_pending_writes.clear() # 持久化 checkpoint self._put_checkpoint({"source": "loop"}) # interrupt_after 检查 if self.interrupt_after and should_interrupt(...): self.status = "interrupt_after"; raise GraphInterrupt() ``` #### 2.5.3 Plan 阶段 — `prepare_next_tasks` **核心位置**：`libs/langgraph/langgraph/pregel/_algo.py:392` ```python def prepare_next_tasks(checkpoint, pending_writes, processes, channels, managed, config, step, stop, *, for_execution, ...): tasks: list = [] # 1. 消费 PUSH 任务（来自 Send 对象） tasks_channel = cast(Topic[Send] | None, channels.get(TASKS)) if tasks_channel and tasks_channel.is_available(): for idx, _ in enumerate(tasks_channel.get()): if task := prepare_single_task((PUSH, idx), None, ...): tasks.append(task) # 2. 优化：根据 updated_channels 反向索引找出候选节点 if updated_channels and trigger_to_nodes: triggered_nodes = set() for channel in updated_channels: if node_ids := trigger_to_nodes.get(channel): triggered_nodes.update(node_ids) candidate_nodes = sorted(triggered_nodes) else: candidate_nodes = processes.keys() # 3. PULL 任务：遍历候选节点，检查触发条件 for name in candidate_nodes: if task := prepare_single_task((PULL, name), None, ...): tasks.append(task) return {t.id: t for t in tasks} ``` **两种触发模式**： - **PULL（订阅式）**：节点声明 `triggers=[channel1, channel2, ...]`，任何 channel 更新且版本号 > 已见版本，触发节点 - **PUSH（推送式）**：通过 `Send(node, arg)` 把任务直接 push 到 `TASKS` Topic，下一 step 立即执行 #### 2.5.4 Execution 阶段 — `PregelRunner` **核心位置**：`libs/langgraph/langgraph/pregel/_runner.py:135` ```python class PregelRunner: """Responsible for executing a set of Pregel tasks concurrently, committing their writes, yielding control to caller when there is output to emit, and interrupting other tasks if appropriate.""" ``` `tick` 方法（同步版 `_runner.py:176`）负责： - 用 `FuturesDict`（Future + Event 组合）追踪并发任务 - 通过 `submit`（线程池或 asyncio）并发执行所有 `PregelExecutableTask` - 提供 fast path：单任务无超时直接同步执行 - 实现错误传播：一个任务失败时是否停止其他任务（`_should_stop_others`） - 集成 `arun_with_retry` / `run_with_retry`（带 retry policy） - 集成 `error_handler`：失败时调度 error_handler 节点（`schedule_error_handler`） **并行执行模型**： - 同步图：使用 `concurrent.futures.ThreadPoolExecutor`（默认） - 异步图：使用 `asyncio`（推荐，支持 timeout 取消） - 每个 superstep 内的所有任务**并发执行**，写入在 step 边界同步（BSP） #### 2.5.5 Update 阶段 — `apply_writes` **核心位置**：`libs/langgraph/langgraph/pregel/_algo.py:232` ```python def apply_writes(checkpoint, channels, tasks, get_next_version, trigger_to_nodes) -> set[str]: # 按路径排序，确保确定性 tasks = sorted(tasks, key=lambda t: task_path_str(t.path[:3])) bump_step = any(t.triggers for t in tasks) # 更新 versions_seen：每个任务记录"已消费的 channel 版本" for task in tasks: checkpoint["versions_seen"].setdefault(task.name, {}).update({ chan: checkpoint["channel_versions"][chan] for chan in task.triggers if chan in checkpoint["channel_versions"] }) # 消费触发 channel（topic 类会被清空） for chan in {... tasks.triggers ...}: if channels[chan].consume() and next_version is not None: checkpoint["channel_versions"][chan] = next_version # 按 channel 分组所有写入 pending_writes_by_channel = defaultdict(list) for task in tasks: for chan, val in task.writes: pending_writes_by_channel[chan].append(val) # 应用写入到 channel（这里调用 reducer） updated_channels = set() for chan, vals in pending_writes_by_channel.items(): if channels[chan].update(vals) and next_version is not None: checkpoint["channel_versions"][chan] = next_version if channels[chan].is_available(): updated_channels.add(chan) # 通知未更新的 channel 进入新 step（用于触发"无变化即终止"） if bump_step: for chan in channels: if channels[chan].is_available() and chan not in updated_channels: if channels[chan].update(EMPTY_SEQ) and next_version is not None: ... # 最后一步：调用所有 channel 的 finish() if bump_step and updated_channels.isdisjoint(trigger_to_nodes): for chan in channels: if channels[chan].finish() and next_version is not None: ... return updated_channels ``` --- ### 2.6 是否支持并行、子图、动态分支 #### 2.6.1 并行（fan-out） **原生支持**，三种方式： 1. **多目标普通边**：`add_edge("A", "B")` + `add_edge("A", "C")` -> A 完成后 B、C 并发执行 2. **条件边返回列表**： ```python def router(state) -> list[str]: return ["B", "C"] # 同时路由到 B 和 C graph.add_conditional_edges("A", router) ``` 3. **`Send` 对象（Map-Reduce 模式）**： ```python def continue_to_jokes(state): return [Send("generate_joke", {"subject": s}) for s in state["subjects"]] ``` 每个 `Send` 创建一个独立的 push 任务，**可携带独立 state**，并发执行后通过 reducer 合并回主 state。 #### 2.6.2 子图（Subgraph） **完全支持**。子图本身是一个编译好的 `CompiledStateGraph`（即 `Pregel` 实例），作为父图的某个节点的 `bound` 运行。 **关键 API**： - `Pregel.get_subgraphs(namespace=None, recurse=False)` (`main.py:1076`) — 通过 namespace 路径访问嵌套子图 - 子图通过 `node.subgraphs` 字段挂载（见 `_read.py` 的 `PregelNode.subgraphs`） - 子图复用父图的 checkpointer / store，namespace 自动加前缀（`NS_SEP`） **状态隔离**：每个子图有独立的 channel namespace 和 checkpoint namespace，父图通过 `Command(PARENT, update=...)` 向上通信（见 `_control_branch` 中的 `ParentCommand` 异常）。 #### 2.6.3 动态分支 **完全支持**，三种形式： 1. **静态条件边**：`add_conditional_edges("A", router_fn, path_map={...})` 2. **运行时 `Command(goto=...)`**：节点内部根据状态决定下一步（`state.py:1749 _control_branch`） 3. **运行时 `Send`**：节点动态生成任意数量的并行子任务（map-reduce） **与 Dify 等静态 DAG 的关键区别**：LangGraph 的边不是编译期固定的拓扑序——节点可以在运行时通过 `Command` 跳到任意节点，甚至回到前序节点（实现循环、回退、重试）。 --- ### 2.7 Human-in-the-loop（人机交互） **核心位置**：`libs/langgraph/langgraph/types.py:575` `Interrupt` + `:851` `interrupt()` 函数 #### 2.7.1 两种机制 **机制 1：`interrupt_before` / `interrupt_after`（声明式）** 在 `compile()` 时指定节点列表，运行到该节点前后自动暂停： ```python graph.compile(checkpointer=memory, interrupt_before=["human_review"]) ``` `should_interrupt` (`_algo.py:155`) 判断逻辑： ```python def should_interrupt(checkpoint, interrupt_nodes, tasks) -> list[PregelExecutableTask]: # 任意 channel 自上次 interrupt 后被更新过 any_updates_since_prev_interrupt = any( version > seen.get(chan, null_version) for chan, version in checkpoint["channel_versions"].items() ) # 且有任务在 interrupt_nodes 列表中 return [task for task in tasks if task.name in interrupt_nodes] if any_updates_since_prev_interrupt else [] ``` **机制 2：`interrupt()` 函数（命令式，推荐）** 节点内部主动抛出中断： ```python # types.py:851 def interrupt(value: Any) -> Any: """Interrupt the graph with a resumable exception from within a node. In a given node, the first invocation raises a GraphInterrupt exception. The provided value is included with the exception and sent to the client. A client resuming the graph must use Command(resume=...) to continue. The graph resumes from the START of the node, RE-EXECUTING all logic. """ ``` #### 2.7.2 恢复机制 通过 `Command(resume=...)` 恢复： ```python for chunk in graph.stream(Command(resume={"approved": True}), config): ... ``` 恢复时： 1. `checkpoint_pending_writes` 中保存了中断前的写入 2. `_reapply_writes_to_succeeded_nodes` (`_loop.py:736`) 恢复已完成节点的写入（不重复执行） 3. 中断的节点从头重新执行，但 `interrupt()` 会跳过（基于 interrupt id 匹配，返回 resume 值） **关键依赖**：HITL **必须**配置 checkpointer，因为中断状态需要持久化。 --- ## 3. 状态管理与持久化 ### 3.1 Checkpoint 数据结构 **核心位置**：`libs/checkpoint/langgraph/checkpoint/base/__init__.py:92` ```python class Checkpoint(TypedDict): v: int # 版本号（当前 1） id: str # 单调递增的唯一 ID（UUID6） ts: str # ISO8601 时间戳 channel_values: dict[str, Any] # 所有 channel 的当前值（序列化） channel_versions: ChannelVersions # channel -> 版本号映射 versions_seen: dict[str, ChannelVersions] # 节点 -> 已见 channel 版本（防止重复触发） updated_channels: list[str] | None # 本步更新的 channel 列表 class CheckpointTuple(NamedTuple): config: RunnableConfig # 指向此 checkpoint 的配置 checkpoint: Checkpoint metadata: CheckpointMetadata # source / step / parents / run_id parent_config: RunnableConfig | None # 父 checkpoint（形成历史链） pending_writes: list[PendingWrite] # 当前 step 未完成的写入 ``` **关键设计**： - **版本号驱动**：每个 channel 有递增版本号，节点通过对比"已见版本"和"当前版本"判断是否应被触发 - **时间旅行**：`parent_config` 形成链表，可回溯任意历史状态（`get_state_history`） - **fork**：从历史 checkpoint 分叉创建新分支（metadata `source="fork"`） ### 3.2 BaseCheckpointSaver 抽象 **核心位置**：`libs/checkpoint/langgraph/checkpoint/base/__init__.py:176` ```python class BaseCheckpointSaver(Generic[V]): def get(self, config) -> Checkpoint | None: ... def get_tuple(self, config) -> CheckpointTuple | None: ... def list(self, config, *, filter, before, limit) -> Iterator[CheckpointTuple]: ... def put(self, config, checkpoint, metadata, new_versions) -> RunnableConfig: ... def put_writes(self, config, writes, task_id) -> None: ... def delete_thread(self, thread_id) -> None: ... # 删除整个 thread def delete_for_runs(self, thread_id, run_ids) -> None: ... def copy_thread(self, from_thread, to_thread) -> None: ... def prune(self, keep) -> None: ... def get_next_version(self, current, channel) -> V: ... ``` **三种官方实现**： - `libs/checkpoint/langgraph/checkpoint/memory/` — `InMemorySaver`（开发测试用） - `libs/checkpoint-sqlite/` — SQLite 持久化 - `libs/checkpoint-postgres/` — Postgres 生产级持久化 - `libs/checkpoint-conformance/` — 一致性测试套件（验证第三方实现是否符合规范） ### 3.3 持久化与恢复流程 **在 Pregel loop 中的位置**（`_loop.py:718`）： ```python def after_tick(self) -> None: self.updated_channels = apply_writes(...) # 应用写入 self._emit("values", ...) # 发出流式事件 self.checkpoint_pending_writes.clear() self._put_checkpoint({"source": "loop"}) # 持久化 checkpoint ... ``` **Durability 模式**（`Pregel.invoke` 的 `durability` 参数，v1.0+）： - `"sync"`: 同步持久化，下一步开始前完成写入（最安全） - `"async"`: 异步持久化，与下一步并发（默认，性能优） - `"exit"`: 仅在图退出时持久化（最快，但中途崩溃无法恢复） **断线重连**：客户端可通过 `thread_id` 重新加载最新 checkpoint 恢复执行；中断的 pending_writes 会被 `_reapply_writes_to_succeeded_nodes` 重放。 --- ## 4. 流式输出（Streaming） ### 4.1 多种 StreamMode **核心位置**：`libs/langgraph/langgraph/pregel/main.py:2616` `stream()` + `:3024` `astream()` + `libs/langgraph/langgraph/stream/` `stream_mode` 支持以下值（可多选）： - `"values"`: 每个 step 后的完整 state 快照 - `"updates"`: 每个 task 的增量写入（默认） - `"messages"`: LLM token 级流（用于 chat） - `"custom"`: 节点内通过 `writer: StreamWriter` 自定义事件 - `"debug"`: 调试事件（task / checkpoint） - `"tasks"`: 任务调度事件 - `"checkpoints"`: checkpoint 保存事件 ### 4.2 Stream 框架 `libs/langgraph/langgraph/stream/` 包含： - `_types.py`: `ProtocolEvent` / `StreamTransformer` 抽象（流事件转换器） - `_mux.py`: `StreamMux` 多路复用器（合并子图流） - `transformers.py`: 内置 transformer（`MessagesTransformer` 等） - `run_stream.py`: 流式运行核心 - `stream_channel.py`: 单流通道抽象 ### 4.3 v2 协议（v1.0+） 新增 `version="v2"` 选项，返回强类型 `StreamPart` TypedDict（`types.py:264-355`）： - `ValuesStreamPart` / `UpdatesStreamPart` / `MessagesStreamPart` - `CustomStreamPart` / `CheckpointStreamPart` / `TasksStreamPart` / `DebugStreamPart` --- ## 5. 关键代码片段要点总结 ### 5.1 Pregel BSP 三阶段循环（核心算法） ```python # libs/langgraph/langgraph/pregel/_loop.py:599,683 def tick(self): # Plan self.tasks = prepare_next_tasks(...) if not self.tasks: self.status = "done"; return False # interrupt_before 检查 if self.interrupt_before and should_interrupt(...): raise GraphInterrupt() return True # (runner 在 tick 返回后并发执行所有 tasks) def after_tick(self): # Update self.updated_channels = apply_writes(...) # 应用 reducer self._put_checkpoint(...) # 持久化 # interrupt_after 检查 if self.interrupt_after and should_interrupt(...): raise GraphInterrupt() ``` ### 5.2 Reducer 通过 Channel 实现 ```python # libs/langgraph/langgraph/channels/binop.py:123 def update(self, values): for value in values: is_overwrite, overwrite_value = _get_overwrite(value) if is_overwrite: self.value = overwrite_value else: self.value = self.operator(self.value, value) # reducer return True ``` ### 5.3 条件边路由（含 Send fan-out） ```python # libs/langgraph/langgraph/graph/_branch.py:192 def _finish(self, writer, input, result, config): if not isinstance(result, (list, tuple)): result = [result] if self.ends: destinations = [r if isinstance(r, Send) else self.ends[r] for r in result] else: destinations = cast(Sequence[Send | str], result) # destinations 可包含多个 Send -> 实现 map-reduce 并行 entries = writer(destinations, False) ... ``` ### 5.4 Send 支持独立 state ```python # libs/langgraph/langgraph/types.py:704 class Send: """A message or packet to send to a specific node in the graph. The sent state can differ from the core graph's state, allowing for flexible and dynamic workflow management. One such example is a 'map-reduce' workflow.""" __slots__ = ("node", "arg", "timeout") ``` ### 5.5 HITL 必须配合 checkpointer ```python # libs/langgraph/langgraph/types.py:851 def interrupt(value: Any) -> Any: """... To use an interrupt, you must enable a checkpointer, as the feature relies on persisting the graph state.""" ``` --- ## 6. 与本项目（agent-management）的关联 根据主项目 `AGENTS.md`，本项目实现了"工作流引擎 + Agentic 模式"，关键概念与 LangGraph 高度对应： | 本项目概念 | LangGraph 对应 | 关键差异 | |---|---|---| | 工作流 DAG 节点 | `StateGraph.add_node` | 本项目节点签名固定（`NodeExecutor.execute(nodeId, data, context, workspace)`），LangGraph 支持 9 种注入组合 | | 条件节点 | `add_conditional_edges` / `Command(goto=...)` | 本项目 condition 节点互斥分支，LangGraph 路径函数返回列表可 fan-out | | Agentic 模式（全局 Planner 决策） | `Command(goto=...)` + 循环 | 本项目由引擎层全局 Planner 决策，LangGraph 由节点内部 `Command(goto=...)` 决策 | | 重访决策 | 循环边 + recursion_limit | 本项目通过 `MAX_VISITS=50` 兜底，LangGraph 通过 `recursion_limit=25` + `versions_seen` 防重复触发 | | AskFor（暂停-恢复） | `interrupt()` + `Command(resume=...)` | 完全对应的 HITL 模型，但本项目 AskFor 仅限 Hermes Agent 内部 | | 运行记录回放 | `get_state_history` + checkpoint | LangGraph 支持时间旅行，本项目仅当前状态可查 | | 断线重连 | `thread_id` + checkpoint 恢复 | 本项目通过 `SseEventBus` 双轨 + `WorkflowPause` 表实现，机制不同但目标一致 | | Hermes 节点输出 | Reducer（`BinaryOperatorAggregate`） | 本项目按 nodeId 隔离不合并，LangGraph 自动合并 | | 死循环保护 | `GraphRecursionError` + recursion_limit | 本项目 AGENTIC 模式 `MAX_VISITS=50`，LangGraph 默认 25 步 | ### 6.1 可作为本仓库架构参考与边界 case 验证的依据 1. **`_algo.py` 的 `apply_writes`** — Reducer 应用顺序、版本号推进逻辑 2. **`_loop.py` 的 `tick`/`after_tick`** — superstep 边界的处理（interrupt、checkpoint、drain） 3. **`state.py` 的 Schema -> Channel 映射** — 如何把声明式 state 定义转为运行时通道 4. **`_branch.py` 的 `_finish`** — 条件边 fan-out 与 Send 的统一处理 5. **`binop.py` 的 Overwrite 语义** — Reducer 中"强制覆盖"的边界情况 ### 6.2 借鉴建议（详见 `workflow-langgraph-dify-comparison-reference.md §15`） 短期低成本借鉴： - HITL 节点级 `interrupt_before/after`（`should_interrupt` 机制） - 节点级 `error_handler_node`（失败路由到错误处理节点） - 节点级 `TimeoutPolicy`（`Pregel.invoke(timeout=...)`） 中期中等成本借鉴： - 子图嵌套（`CompiledStateGraph` 作为节点 bound） - 迭代节点（参考 `Send` + Reducer 的 map-reduce 模式） 长期慎选借鉴： - Checkpoint 时间旅行（`get_state_history` + fork） - 完整 Channel + Reducer 体系（与现有 nodeId 隔离冲突） - Send map-reduce（与现有 condition 互斥分支冲突） --- ## 7. 关键文件索引（绝对路径） ### 核心 StateGraph API - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\graph\state.py` — StateGraph/CompiledStateGraph 主体（76KB） - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\graph\_node.py` — StateNode 协议定义 - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\graph\_branch.py` — BranchSpec 条件边 - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\graph\message.py` — MessagesState / add_messages reducer ### Pregel 执行引擎（最核心） - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\pregel\main.py` — Pregel 类（175KB，invoke/stream/astream） - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\pregel\_algo.py` — apply_writes / prepare_next_tasks / should_interrupt - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\pregel\_loop.py` — PregelLoop / SyncPregelLoop / AsyncPregelLoop - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\pregel\_runner.py` — PregelRunner（并发执行） - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\pregel\_read.py` — PregelNode / ChannelRead - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\pregel\_write.py` — ChannelWrite ### Channel（状态通道） - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\channels\base.py` — BaseChannel 抽象 - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\channels\last_value.py` — LastValue（默认覆盖） - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\channels\binop.py` — BinaryOperatorAggregate（Reducer） - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\channels\topic.py` — Topic（PubSub） - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\channels\named_barrier_value.py` — NamedBarrierValue（join 同步） ### Checkpoint（持久化） - `D:\projects\agent-management\langgraph\libs\checkpoint\langgraph\checkpoint\base\__init__.py` — BaseCheckpointSaver / Checkpoint / CheckpointTuple - `D:\projects\agent-management\langgraph\libs\checkpoint\langgraph\checkpoint\memory\` — InMemorySaver - `D:\projects\agent-management\langgraph\libs\checkpoint-sqlite\` — SQLite 实现 - `D:\projects\agent-management\langgraph\libs\checkpoint-postgres\` — Postgres 实现 ### 类型与控制原语 - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\types.py` — Send / Command / Interrupt / RetryPolicy / CachePolicy / TimeoutPolicy / StateSnapshot - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\errors.py` — GraphInterrupt / GraphRecursionError / GraphDrained / ParentCommand - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\constants.py` — START / END / TAG_HIDDEN ### 函数式 API - `D:\projects\agent-management\langgraph\libs\langgraph\langgraph\func\__init__.py` — `@task` / `@entrypoint` 装饰器（隐式 Pregel） ### 高层封装 - `D:\projects\agent-management\langgraph\libs\prebuilt\langgraph\prebuilt\chat_agent_executor.py` — `create_react_agent` - `D:\projects\agent-management\langgraph\libs\prebuilt\langgraph\prebuilt\tool_node.py` — `ToolNode` / `tools_condition` ### 文档与示例 - `D:\projects\agent-management\langgraph\README.md` — 项目总览（提及 Pregel/Beam/NetworkX 启发） - `D:\projects\agent-management\langgraph\AGENTS.md` / `CLAUDE.md` — 仓库开发规范 - `D:\projects\agent-management\langgraph\examples\` — Jupyter 示例（多已迁移到 docs） - `D:\projects\agent-management\langgraph\libs\langgraph\tests\` — 测试套件（理解行为的最佳资料） --- ## 8. 关联文档 - `workflow-langgraph-dify-comparison-reference.md` — LangGraph / Dify / 本仓库三方对比，含演进建议 - `workflow-vs-dify-comparison.md` — 本仓库 vs Dify 详细对比 - `dify-reference.md` — Dify 平台总览 - `dify-workflow-nodes-reference.md` — Dify 工作流节点系统