LangGraph 源码深潜：Reflection 与评价-修改循环机制解剖

1
class StateGraph(Graph):
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: Any,
10
    ) -> None:
11
        ...

1
用户代码 StateGraph(TravelState)
2
  ↓
3
StateGraph.__init__
4
  ↓
5
_add_schema / _get_channels / Graph.__init__

1
class StateGraph(Graph):
2
    def __init__(self, state_schema, context_schema=None,
3
                 input_schema=None, output_schema=None):
4
        # 1. 初始化 Graph 基类中的节点、边、分支容器
5
        super().__init__()
6

7
        # 2. 保存 schema 形态
8
        self.schema = state_schema
9
        self.input_schema = input_schema or state_schema
10
        self.output_schema = output_schema or state_schema
11
        self.context_schema = context_schema
12

13
        # 3. 初始化状态相关容器
14
        self.schemas = {}
15
        self.channels = {}
16
        self.managed = {}
17

18
        # 4. 解析 state schema
19
        self._add_schema(state_schema)
20

21
        # 5. 如果 input / output schema 不同，也单独解析
22
        if self.input_schema is not state_schema:
23
            self._add_schema(self.input_schema)
24
        if self.output_schema is not state_schema:
25
            self._add_schema(self.output_schema)

1
TravelState
2
  ↓
3
_add_schema
4
  ↓
5
channels: draft_plan / critique / quality_score / revision_count / final_plan
6
  ↓
7
StateGraph builder ready

1
def add_node(
2
    self,
3
    node: str | StateNode[NodeInputT] | None = None,
4
    action: StateNode[NodeInputT] | 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
    destinations: dict[str, str] | tuple[str, ...] | None = None,
12
    **kwargs: Any,
13
) -> Self:
14
    ...

1
builder.add_node("evaluate_draft", evaluate_draft)
2
  ↓
3
StateGraph.add_node
4
  ↓
5
coerce_to_runnable / node spec 保存

1
def add_node(self, node=None, action=None, **options):
2
    # 1. 解析调用形式
3
    if isinstance(node, str):
4
        node_name = node
5
        node_action = action
6
    else:
7
        node_action = node
8
        node_name = infer_name_from_callable(node_action)
9

10
    # 2. 校验节点名
11
    if node_name in RESERVED_NAMES:
12
        raise ValueError("Node name conflicts with reserved names")
13
    if node_name in self.nodes:
14
        raise ValueError("Node already exists")
15

16
    # 3. 把 callable 转成 runnable-like 节点
17
    runnable = coerce_to_runnable(
18
        node_action,
19
        name=node_name,
20
    )
21

22
    # 4. 保存节点 spec
23
    self.nodes[node_name] = StateNodeSpec(
24
        runnable=runnable,
25
        metadata=options.get("metadata"),
26
        input_schema=options.get("input_schema"),
27
        retry_policy=options.get("retry_policy"),
28
        cache_policy=options.get("cache_policy"),
29
        destinations=options.get("destinations"),
30
        defer=options.get("defer", False),
31
    )
32

33
    return self

1
evaluate_draft function
2
  ↓
3
add_node
4
  ↓
5
StateNodeSpec
6
  ↓
7
self.nodes["evaluate_draft"]

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

1
builder.add_conditional_edges("evaluate_draft", route_by_score)
2
  ↓
3
StateGraph.add_conditional_edges
4
  ↓
5
BranchSpec.from_path
6
  ↓
7
self.branches[source][branch_name] = branch

1
def add_conditional_edges(self, source, path, path_map=None):
2
    # 1. 校验 source
3
    if source not in self.nodes:
4
        # 允许稍后校验，也可能立即报错，具体以源码为准
5
        record_pending_source(source)
6

7
    # 2. 把 route function 转成 runnable
8
    path_runnable = coerce_to_runnable(path, name=None)
9

10
    # 3. 推断 branch 名称
11
    branch_name = path_runnable.name or "condition"
12

13
    # 4. 防止同一个 source 下 branch 名冲突
14
    if branch_name in self.branches[source]:
15
        raise ValueError("Branch already exists")
16

17
    # 5. 从 path / path_map / Literal 返回类型构造 BranchSpec
18
    branch = BranchSpec.from_path(
19
        path=path_runnable,
20
        path_map=path_map,
21
        infer_schema=True,
22
    )
23

24
    # 6. 保存到 source 的 branch 表
25
    self.branches[source][branch_name] = branch
26

27
    return self

1
evaluate_draft 执行完成
2
  ↓
3
state 包含 quality_score / revision_count
4
  ↓
5
route_by_score(state)
6
  ↓
7
revise_draft 或 final_response

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
builder.compile()
2
  ↓
3
StateGraph.compile
4
  ↓
5
validate graph
6
  ↓
7
CompiledStateGraph(...)
8
  ↓
9
attach_node / attach_edge / attach_branch

1
def compile(self, checkpointer=None, **options):
2
    # 1. 校验构建期图结构
3
    self.validate(
4
        interrupt_before=options.get("interrupt_before"),
5
        interrupt_after=options.get("interrupt_after"),
6
    )
7

8
    # 2. 创建 CompiledStateGraph
9
    compiled = CompiledStateGraph(
10
        builder=self,
11
        channels=self.channels,
12
        nodes={},
13
        input_channels=START,
14
        output_channels=self.output_channels,
15
        checkpointer=checkpointer,
16
        cache=options.get("cache"),
17
        store=options.get("store"),
18
        debug=options.get("debug", False),
19
        name=options.get("name"),
20
    )
21

22
    # 3. attach START、普通节点
23
    compiled.attach_node(START, None)
24
    for node_name, node_spec in self.nodes.items():
25
        compiled.attach_node(node_name, node_spec)
26

27
    # 4. attach 普通边
28
    for start, end in self.edges:
29
        compiled.attach_edge(start, end)
30

31
    # 5. attach 条件分支
32
    for source, branches in self.branches.items():
33
        for branch_name, branch in branches.items():
34
            compiled.attach_branch(source, branch_name, branch)
35

36
    # 6. 校验 compiled graph
37
    return compiled.validate()

1
StateGraph builder
2
  ↓
3
compile
4
  ↓
5
CompiledStateGraph
6
  ↓
7
可 invoke / stream / ainvoke

1
def invoke(
2
    self,
3
    input: InputT,
4
    config: RunnableConfig | None = None,
5
    *,
6
    context: ContextT | None = None,
7
    stream_mode: StreamMode | None = None,
8
    **kwargs: Any,
9
) -> OutputT:
10
    ...

1
graph.invoke(initial_state)
2
  ↓
3
CompiledStateGraph / Pregel.invoke
4
  ↓
5
stream loop
6
  ↓
7
return latest output

1
def invoke(self, input_value, config=None, context=None, **kwargs):
2
    # 1. 归一化 RunnableConfig
3
    config = ensure_config(config)
4

5
    # 2. 把 context 放入 runtime，上下文不是 state
6
    runtime = create_runtime_context(
7
        config=config,
8
        context=context,
9
    )
10

11
    # 3. 调用 stream 主链，但只保留最终结果
12
    latest_output = None
13
    for chunk in self.stream(
14
        input_value,
15
        config=config,
16
        context=context,
17
        **kwargs,
18
    ):
19
        latest_output = extract_output(chunk)
20

21
    # 4. 返回最终 state 或 output schema 对应对象
22
    return latest_output

1
initial TravelState
2
  ↓
3
state channels 初始化
4
  ↓
5
Pregel 执行
6
  ↓
7
final TravelState

1
def execute_superstep(loop, config, runtime):
2
    # 1. 从 channels 读取当前 state 快照
3
    current_state = loop.read_state()
4

5
    # 2. 根据当前触发条件选择待执行 tasks
6
    tasks = loop.prepare_tasks(current_state)
7

8
    # 3. 给每个 task 注入 config / runtime / state
9
    for task in tasks:
10
        task_input = build_node_input(
11
            state=current_state,
12
            config=config,
13
            runtime=runtime,
14
        )
15
        task.schedule(task_input)
16

17
    # 4. task 返回 partial update
18
    writes = collect_task_writes(tasks)
19

20
    # 5. 应用 writes 到 channels
21
    loop.apply_writes(writes)

1
def generate_draft(state: TravelState) -> dict[str, Any]:
2
    ...

1
Pregel runner
2
  ↓
3
generate_draft runnable
4
  ↓
5
node function
6
  ↓
7
state channel writes

1
def run_generate_draft_task(state, config, runtime):
2
    # 1. Pregel 构造节点输入
3
    node_input = state
4

5
    # 2. 调用用户函数
6
    update = generate_draft(node_input)
7

8
    # 3. 校验返回值是 mapping / Command 等合法写入
9
    writes = normalize_node_update(update)
10

11
    # 4. 将每个 key 转成 channel write
12
    for key, value in writes.items():
13
        emit_write(channel=key, value=value)
14

15
    # 5. 当前 super-step 结束时统一应用 write
16
    return writes

1
user_request
2
  ↓
3
generate_draft
4
  ↓
5
{"draft_plan": ..., "revision_count": 0}
6
  ↓
7
state 更新

1
def evaluate_draft(state: TravelState) -> dict[str, Any]:
2
    ...

1
Pregel runner
2
  ↓
3
evaluate_draft runnable
4
  ↓
5
LLM / rule-based evaluator
6
  ↓
7
state channel writes

1
def run_evaluate_draft_task(state, config, runtime):
2
    # 1. 读取当前草案
3
    draft = state.get("draft_plan")
4
    if not draft:
5
        return {
6
            "quality_score": 0.0,
7
            "critique": "没有可评价的草案。",
8
        }
9

10
    # 2. 构造评价请求
11
    evaluation_input = {
12
        "draft_plan": draft,
13
        "criteria": runtime.context.criteria,
14
    }
15

16
    # 3. 调用 evaluator，推荐结构化输出
17
    evaluation = evaluator_model.with_structured_output(
18
        EvaluationResult
19
    ).invoke(evaluation_input)
20

21
    # 4. 返回 partial update
22
    return {
23
        "quality_score": evaluation.quality_score,
24
        "critique": evaluation.critique,
25
    }

1
draft_plan
2
  ↓
3
evaluator
4
  ↓
5
quality_score + critique
6
  ↓
7
route_by_score

1
def route_by_score(state: TravelState) -> Literal["revise_draft", "final_response"]:
2
    ...

1
class BranchSpec:
2
    def run(self, writer, reader=None) -> Runnable:
3
        ...

1
evaluate_draft 完成
2
  ↓
3
BranchSpec.run
4
  ↓
5
BranchSpec._route
6
  ↓
7
route_by_score(state)
8
  ↓
9
BranchSpec._finish
10
  ↓
11
写入目标节点 channel

1
def branch_route_after_evaluate(state, config, runtime):
2
    # 1. 读取 state 快照
3
    branch_input = state
4

5
    # 2. 调用用户路由函数
6
    route_result = route_by_score(branch_input)
7

8
    # 3. 归一化返回值
9
    if isinstance(route_result, (list, tuple)):
10
        destinations = list(route_result)
11
    else:
12
        destinations = [route_result]
13

14
    # 4. 处理 END
15
    normalized = []
16
    for dest in destinations:
17
        if dest == END:
18
            normalized.append(END)
19
        else:
20
            normalized.append(resolve_path_map(dest))
21

22
    # 5. 将目的地写入 Pregel channel
23
    for dest in normalized:
24
        if dest is END:
25
            emit_end_signal()
26
        else:
27
            emit_branch_write(target_node=dest)
28

29
    return normalized

1
quality_score >= 0.8
2
  ↓
3
final_response
4

5
quality_score < 0.8 and revision_count < MAX_REVISION
6
  ↓
7
revise_draft

1
def revise_draft(state: TravelState) -> dict[str, Any]:
2
    ...

1
route_by_score → revise_draft
2
  ↓
3
Pregel runner
4
  ↓
5
revise_draft callable
6
  ↓
7
state update
8
  ↓
9
evaluate_draft

1
def run_revise_draft_task(state, config, runtime):
2
    # 1. 读取旧草案与评价意见
3
    old_draft = state["draft_plan"]
4
    critique = state["critique"]
5
    old_count = state["revision_count"]
6

7
    # 2. 调用 revise 模型或规则修改器
8
    revised = revise_model.invoke({
9
        "draft_plan": old_draft,
10
        "critique": critique,
11
        "user_request": state["user_request"],
12
    })
13

14
    # 3. 递增 revision_count
15
    new_count = old_count + 1
16

17
    # 4. 返回 partial update
18
    return {
19
        "draft_plan": revised.content,
20
        "revision_count": new_count,
21
    }

1
critique
2
  ↓
3
revise_draft
4
  ↓
5
new draft_plan + revision_count + 1
6
  ↓
7
evaluate_draft

1
draft_history: Annotated[list[DraftVersion], add]
2
best_draft: str | None
3
best_score: float | None

1
def final_response(state: TravelState) -> dict[str, Any]:
2
    ...

1
route_by_score → final_response
2
  ↓
3
Pregel runner
4
  ↓
5
final_response callable
6
  ↓
7
state update
8
  ↓
9
END

1
def run_final_response_task(state, config, runtime):
2
    # 1. 优先选择 best_draft，如果没有则选择当前 draft
3
    if "best_draft" in state and state["best_draft"]:
4
        selected = state["best_draft"]
5
    else:
6
        selected = state["draft_plan"]
7

8
    # 2. 附加质量说明
9
    final_text = format_final_plan(
10
        plan=selected,
11
        score=state.get("quality_score"),
12
        critique=state.get("critique"),
13
        revision_count=state.get("revision_count"),
14
    )
15

16
    # 3. 写入 final_plan
17
    return {
18
        "final_plan": final_text
19
    }

1
final_response
2
  ↓
3
{"final_plan": ...}
4
  ↓
5
END