LangGraph 源码深潜：Plan-and-Execute 与 Subgraph 运行机制解剖

1
class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
2
    def __init__(
3
        self,
4
        state_schema: type[StateT],
5
        context_schema: type[ContextT] | None = None,
6
        *,
7
        input_schema: type[InputT] | None = None,
8
        output_schema: type[OutputT] | None = None,
9
        **kwargs,
10
    ) -> None:
11
        ...

1
builder = StateGraph(TravelState)
2
  ↓
3
StateGraph.__init__
4
  ↓
5
_add_schema(state_schema)
6
  ↓
7
_get_channels(schema)

1
def __init__(state_schema, context_schema=None, input_schema=None, output_schema=None):
2
    # 1. 初始化图结构容器
3
    self.nodes = {}
4
    self.edges = set()
5
    self.branches = defaultdict(dict)
6
    self.waiting_edges = set()
7

8
    # 2. 初始化状态相关容器
9
    self.schemas = {}
10
    self.channels = {}
11
    self.managed = {}
12

13
    # 3. 保存 schema 边界
14
    self.state_schema = state_schema
15
    self.input_schema = input_schema or state_schema
16
    self.output_schema = output_schema or state_schema
17
    self.context_schema = context_schema
18

19
    # 4. 解析主状态 schema
20
    self._add_schema(self.state_schema)
21

22
    # 5. 解析输入输出 schema，禁止 managed value 进入 I/O schema
23
    self._add_schema(self.input_schema, allow_managed=False)
24
    self._add_schema(self.output_schema, allow_managed=False)
25

26
    # 6. 标记尚未编译
27
    self.compiled = False

1
TravelState TypedDict
2
  ↓
3
_get_channels
4
  ↓
5
plan_steps / current_step_index / final_answer 等字段变成 channels
6
  ↓
7
StateGraph builder 可注册节点和边

1
def add_node(
2
    self,
3
    node: str | StateNode[NodeInputT, ContextT],
4
    action: StateNode[NodeInputT, ContextT] | None = None,
5
    *,
6
    defer: bool = False,
7
    metadata: dict[str, Any] | None = None,
8
    input_schema: type[NodeInputT] | None = None,
9
    retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
10
    cache_policy: CachePolicy | None = None,
11
    error_handler: StateNode[Any, ContextT] | None = None,
12
    destinations: dict[str, str] | tuple[str, ...] | None = None,
13
    timeout: float | timedelta | TimeoutPolicy | None = None,
14
    **kwargs,
15
) -> Self:
16
    ...

1
builder.add_node("planner_node", planner_node)
2
  ↓
3
StateGraph.add_node
4
  ↓
5
coerce_to_runnable / StateNodeSpec
6
  ↓
7
self.nodes[name] = spec

1
def add_node(node, action=None, *, input_schema=None, retry_policy=None, cache_policy=None, error_handler=None, destinations=None, timeout=None):
2
    # 1. 如果 node 不是字符串，说明用户传的是函数或 Runnable
3
    if not isinstance(node, str):
4
        action = node
5
        node_name = infer_name_from_callable_or_runnable(action)
6
    else:
7
        node_name = node
8

9
    # 2. 校验节点名
10
    if node_name is None:
11
        raise ValueError("Node name must be provided")
12
    if node_name in self.nodes:
13
        raise ValueError("Node already exists")
14
    if node_name in {START, END}:
15
        raise ValueError("Reserved node name")
16

17
    # 3. 推断输入 schema
18
    node_input_schema = input_schema or self.state_schema
19
    self._add_schema(node_input_schema, allow_managed=False)
20

21
    # 4. 将 action 归一化为 runnable-like node
22
    runnable = coerce_to_runnable(action, name=node_name)
23

24
    # 5. 保存节点规格
25
    self.nodes[node_name] = StateNodeSpec(
26
        runnable=runnable,
27
        metadata=metadata,
28
        input_schema=node_input_schema,
29
        retry_policy=retry_policy,
30
        cache_policy=cache_policy,
31
        error_handler=error_handler,
32
        destinations=destinations,
33
        timeout=timeout,
34
    )
35

36
    return self

1
planner_node function
2
  ↓
3
add_node
4
  ↓
5
StateNodeSpec
6
  ↓
7
compile 时 attach 为 PregelNode

1
def add_conditional_edges(
2
    self,
3
    source: str,
4
    path: Callable[..., Hashable | Sequence[Hashable]] | Runnable[Any, Hashable | Sequence[Hashable]],
5
    path_map: dict[Hashable, str] | list[str] | None = None,
6
) -> Self:
7
    ...

1
builder.add_conditional_edges("execute_step", route_next_step)
2
  ↓
3
StateGraph.add_conditional_edges
4
  ↓
5
BranchSpec.from_path(path, path_map, infer_schema=True)
6
  ↓
7
self.branches[source][branch_name] = branch_spec

1
def add_conditional_edges(source, path, path_map=None):
2
    # 1. 校验 source 是否可作为边起点
3
    if source == END:
4
        raise ValueError("END cannot have outgoing edges")
5

6
    # 2. 将 path 函数转换为 branch spec
7
    branch = BranchSpec.from_path(
8
        path=path,
9
        path_map=path_map,
10
        infer_schema=True,
11
    )
12

13
    # 3. 生成 branch 名称，避免同一 source 下重复
14
    branch_name = branch.name or "condition"
15
    if branch_name in self.branches[source]:
16
        raise ValueError("Branch already exists for source")
17

18
    # 4. 保存到 builder
19
    self.branches[source][branch_name] = branch
20

21
    return self

1
execute_step 执行完成
2
  ↓
3
BranchSpec.run(route_next_step)
4
  ↓
5
返回 execute_step / replan / final_response
6
  ↓
7
Pregel 写入对应 branch channel

1
subgraph = subgraph_builder.compile()
2
parent_builder.add_node("execute_step", subgraph)

1
executor_builder = StateGraph(ExecutorState)
2
  ↓
3
executor_builder.add_node(...)
4
  ↓
5
executor_graph = executor_builder.compile()
6
  ↓
7
parent_builder.add_node("execute_step", executor_graph)

1
def build_executor_subgraph():
2
    # 1. 定义 executor 私有状态
3
    class ExecutorState(TypedDict):
4
        current_step: PlanStep
5
        tool_results: list[str]
6
        step_status: str
7
        step_error: str | None
8

9
    # 2. 构建子图
10
    executor = StateGraph(ExecutorState)
11
    executor.add_node("prepare_step", prepare_step)
12
    executor.add_node("call_tools", call_tools)
13
    executor.add_node("verify_step", verify_step)
14

15
    # 3. 添加内部边
16
    executor.add_edge(START, "prepare_step")
17
    executor.add_edge("prepare_step", "call_tools")
18
    executor.add_edge("call_tools", "verify_step")
19
    executor.add_edge("verify_step", END)
20

21
    # 4. 编译为可执行子图
22
    return executor.compile()
23

24

25
def parent_execute_step_node(parent_state):
26
    # 5. 将父图 state 映射为子图输入
27
    sub_input = {
28
        "current_step": parent_state["plan_steps"][parent_state["current_step_index"]],
29
        "tool_results": [],
30
        "step_status": "running",
31
        "step_error": None,
32
    }
33

34
    # 6. 调用子图
35
    sub_output = executor_graph.invoke(sub_input)
36

37
    # 7. 将子图输出映射回父图 partial update
38
    return convert_executor_output_to_parent_update(parent_state, sub_output)

1
parent TravelState
2
  ↓
3
wrapper 提取 current_step
4
  ↓
5
executor subgraph.invoke(ExecutorState)
6
  ↓
7
subgraph 执行内部工具/校验
8
  ↓
9
wrapper 转成 TravelState partial update

1
def compile(
2
    self,
3
    checkpointer: Checkpointer = None,
4
    *,
5
    cache: BaseCache | None = None,
6
    store: BaseStore | None = None,
7
    interrupt_before: All | list[str] | None = None,
8
    interrupt_after: All | list[str] | None = None,
9
    debug: bool = False,
10
    name: str | None = None,
11
) -> CompiledStateGraph:
12
    ...

1
graph = builder.compile()
2
  ↓
3
StateGraph.compile
4
  ↓
5
validate graph structure
6
  ↓
7
CompiledStateGraph(...)
8
  ↓
9
attach_node / attach_edge / attach_branch
10
  ↓
11
validate compiled graph

1
def compile(checkpointer=None, cache=None, store=None, interrupt_before=None, interrupt_after=None, debug=False, name=None):
2
    # 1. 校验 builder 图结构
3
    self.validate(
4
        interrupt_before=interrupt_before,
5
        interrupt_after=interrupt_after,
6
    )
7

8
    # 2. 准备输出和 stream channels
9
    output_channels = infer_output_channels(self.output_schema)
10
    stream_channels = infer_stream_channels(self.channels)
11

12
    # 3. 构造 CompiledStateGraph
13
    compiled = CompiledStateGraph(
14
        builder=self,
15
        channels=self.channels,
16
        nodes={},
17
        input_channels=START,
18
        output_channels=output_channels,
19
        stream_channels=stream_channels,
20
        checkpointer=checkpointer,
21
        cache=cache,
22
        store=store,
23
        interrupt_before_nodes=interrupt_before,
24
        interrupt_after_nodes=interrupt_after,
25
        debug=debug,
26
        name=name,
27
    )
28

29
    # 4. 挂载所有普通节点
30
    for node_name, node_spec in self.nodes.items():
31
        compiled.attach_node(node_name, node_spec)
32

33
    # 5. 挂载固定边
34
    for start, end in self.edges:
35
        compiled.attach_edge(start, end)
36

37
    # 6. 挂载等待边和条件分支
38
    for starts, end in self.waiting_edges:
39
        compiled.attach_edge(starts, end)
40

41
    for source, branches in self.branches.items():
42
        for branch_name, branch_spec in branches.items():
43
            compiled.attach_branch(source, branch_name, branch_spec)
44

45
    # 7. 校验 compiled graph
46
    compiled.validate()
47

48
    self.compiled = True
49
    return compiled

1
builder.nodes / edges / branches
2
  ↓
3
compile
4
  ↓
5
compiled.nodes / channels / triggers / writes
6
  ↓
7
invoke 时可调度

1
def invoke(self, input: InputT, config: RunnableConfig | None = None, **kwargs) -> OutputT:
2
    ...

1
graph.invoke(initial_state)
2
  ↓
3
Pregel.invoke
4
  ↓
5
Pregel.stream
6
  ↓
7
loop.input / input channel writes

1
def invoke(input_state, config=None):
2
    # 1. 归一化 RunnableConfig
3
    config = ensure_config(config)
4

5
    # 2. 创建运行时 loop
6
    loop = create_pregel_loop(
7
        input=input_state,
8
        config=config,
9
        channels=compiled.channels,
10
        nodes=compiled.nodes,
11
    )
12

13
    # 3. 执行 stream，但只收集最终输出
14
    final_output = None
15
    for event in loop.stream():
16
        if event_is_output(event):
17
            final_output = event_to_output(event)
18

19
    # 4. 返回最终 state 或 output schema 投影
20
    return final_output

1
initial_state dict
2
  ↓
3
input channels
4
  ↓
5
START message
6
  ↓
7
planner_node active

1
def run_node(node, current_state, config, runtime):
2
    # 1. 从 channels 读取当前 state 快照
3
    state_snapshot = read_state_from_channels(current_state)
4

5
    # 2. 准备节点参数
6
    if node_accepts_runtime(node):
7
        args = (state_snapshot, runtime)
8
    else:
9
        args = (state_snapshot,)
10

11
    # 3. 调用节点函数或 runnable
12
    update = node.invoke_or_call(*args, config=config)
13

14
    # 4. 将节点返回值转换成 channel writes
15
    writes = map_update_to_channel_writes(update)
16

17
    return writes

1
def planner_node(state: TravelState) -> dict:
2
    ...

1
Pregel runtime
2
  ↓
3
PregelNode for planner_node
4
  ↓
5
planner_node(state)
6
  ↓
7
ChannelWrite(plan_steps, current_step_index)

1
def planner_node(state):
2
    # 1. 读取用户请求
3
    user_request = state["user_request"]
4

5
    # 2. 调用 LLM / 规则 planner 生成结构化计划
6
    plan_steps = build_plan_steps(user_request)
7

8
    # 3. 校验计划基本结构
9
    if not plan_steps:
10
        return {
11
            "plan_error": "empty_plan",
12
            "plan_status": "invalid",
13
        }
14

15
    # 4. 标准化 step_id 与 status
16
    normalized_steps = []
17
    for i, step in enumerate(plan_steps):
18
        normalized_steps.append({
19
            "step_id": i + 1,
20
            "task": step["task"],
21
            "status": "pending",
22
        })
23

24
    # 5. 返回 partial state update
25
    return {
26
        "plan_steps": normalized_steps,
27
        "current_step_index": 0,
28
        "step_results": [],
29
        "plan_status": "ready",
30
    }

1
user_request
2
  ↓
3
planner_node
4
  ↓
5
{plan_steps, current_step_index=0}
6
  ↓
7
state merge
8
  ↓
9
execute_step

1
def execute_step(state: TravelState) -> dict:
2
    ...

1
Pregel runtime
2
  ↓
3
execute_step node
4
  ↓
5
read state["plan_steps"][current_step_index]
6
  ↓
7
call tool / subgraph / LLM
8
  ↓
9
return partial update

1
def execute_step(state):
2
    # 1. 读取执行游标
3
    index = state["current_step_index"]
4
    steps = state["plan_steps"]
5

6
    # 2. 边界检查
7
    if index < 0 or index >= len(steps):
8
        return {
9
            "step_error": "index_out_of_range",
10
            "execution_status": "failed",
11
        }
12

13
    # 3. 取出当前步骤
14
    current_step = steps[index]
15

16
    # 4. 标记 running，避免重复执行时状态不清
17
    updated_steps = copy_steps(steps)
18
    updated_steps[index]["status"] = "running"
19

20
    # 5. 执行当前步骤
21
    try:
22
        step_result = run_step_task(current_step, state)
23
        updated_steps[index]["status"] = "done"
24
        next_index = index + 1
25
        execution_status = "ok"
26
        step_error = None
27
    except RecoverableStepError as exc:
28
        step_result = None
29
        updated_steps[index]["status"] = "failed"
30
        next_index = index
31
        execution_status = "recoverable_failed"
32
        step_error = str(exc)
33
    except Exception as exc:
34
        raise
35

36
    # 6. 返回 partial update
37
    update = {
38
        "plan_steps": updated_steps,
39
        "current_step_index": next_index,
40
        "execution_status": execution_status,
41
        "step_error": step_error,
42
    }
43

44
    if step_result is not None:
45
        update["step_results"] = state["step_results"] + [step_result]
46

47
    return update

1
current_step_index = 0
2
  ↓
3
execute step 1
4
  ↓
5
step_results += result_1
6
  ↓
7
current_step_index = 1

1
def route_next_step(state: TravelState) -> Literal[
2
    "execute_step",
3
    "replan",
4
    "final_response",
5
]:
6
    ...

1
builder.add_conditional_edges("check_step_result", route_next_step)

1
check_step_result / execute_step 完成
2
  ↓
3
BranchSpec.run
4
  ↓
5
route_next_step(state)
6
  ↓
7
BranchSpec._finish
8
  ↓
9
write branch:to:<target>

1
def route_next_step(state):
2
    # 1. 如果执行出现可恢复失败，进入 replan
3
    if state.get("execution_status") == "recoverable_failed":
4
        return "replan"
5

6
    # 2. 如果计划本身无效，进入 replan
7
    if state.get("plan_status") == "invalid":
8
        return "replan"
9

10
    # 3. 如果还有未执行步骤，继续 executor
11
    if state["current_step_index"] < len(state["plan_steps"]):
12
        return "execute_step"
13

14
    # 4. 所有步骤完成，进入最终响应
15
    return "final_response"

1
def branch_route(input_state):
2
    # 1. 调用用户 path function
3
    result = route_next_step(input_state)
4

5
    # 2. 归一化为目标列表
6
    if not isinstance(result, list):
7
        destinations = [result]
8
    else:
9
        destinations = result
10

11
    # 3. path_map 映射业务标签到节点名
12
    destinations = [path_map.get(d, d) for d in destinations]
13

14
    # 4. 校验目标
15
    for dest in destinations:
16
        if dest is None or dest == START:
17
            raise ValueError("Invalid branch destination")
18

19
    # 5. 写入 branch channel，激活后继节点
20
    return write_to_branch_channels(destinations)