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 版）。本份源码可作为：

架构参考——Pregel BSP 三阶段循环、Channel 抽象、Reducer 合并、Checkpoint 时间旅行

边界 case 验证——apply_writes 的 Reducer 应用顺序、versions_seen 防重复触发、pending_writes 重放

演进借鉴参考——详见 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 种形式：

# 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) 是节点的编译期表示：

# 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 多入边）

# 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 行）：

# 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),))
            )

条件边（含动态分发）

# 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) 是条件边编译期表示，核心字段：

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`

# 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 映射：

# 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）：

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）：

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

典型用法：

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

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）：

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

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

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

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）

原生支持，三种方式：

多目标普通边：add_edge("A", "B") + add_edge("A", "C") -> A 完成后 B、C 并发执行

条件边返回列表：

def router(state) -> list[str]:
   return ["B", "C"]   # 同时路由到 B 和 C
graph.add_conditional_edges("A", router)

Send 对象（Map-Reduce 模式）：
```
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 动态分支

完全支持，三种形式：

静态条件边：add_conditional_edges("A", router_fn, path_map={...})
运行时 Command(goto=...)：节点内部根据状态决定下一步（state.py:1749 _control_branch）
运行时 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() 时指定节点列表，运行到该节点前后自动暂停：

graph.compile(checkpointer=memory, interrupt_before=["human_review"])

should_interrupt (_algo.py:155) 判断逻辑：

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() 函数（命令式，推荐）

节点内部主动抛出中断：

# 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=...) 恢复：

for chunk in graph.stream(Command(resume={"approved": True}), config):
    ...

恢复时：

checkpoint_pending_writes 中保存了中断前的写入
_reapply_writes_to_succeeded_nodes (_loop.py:736) 恢复已完成节点的写入（不重复执行）
中断的节点从头重新执行，但 interrupt() 会跳过（基于 interrupt id 匹配，返回 resume 值）

关键依赖：HITL 必须配置 checkpointer，因为中断状态需要持久化。

3. 状态管理与持久化

3.1 Checkpoint 数据结构

核心位置：libs/checkpoint/langgraph/checkpoint/base/__init__.py:92

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

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）：

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 三阶段循环（核心算法）

# 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 实现

# 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）

# 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

# 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

# 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 验证的依据

_algo.py 的 apply_writes — Reducer 应用顺序、版本号推进逻辑
_loop.py 的 tick/after_tick — superstep 边界的处理（interrupt、checkpoint、drain）
state.py 的 Schema -> Channel 映射 — 如何把声明式 state 定义转为运行时通道
_branch.py 的 _finish — 条件边 fan-out 与 Send 的统一处理
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 工作流节点系统

langgraph-source-reference.md 43 KB Riwayat Mentahan