1
def _warn_invalid_state_schema(schema: type[Any] | Any) -> None:
2
    """
3
    警告无效的状态模式。
4

5
    如果模式不是类型，且没有泛型参数，则发出警告。
6
    这有助于用户发现状态模式定义错误。
7
    """
8
    if isinstance(schema, type):
9
        # 如果是类型，直接返回（有效）
10
        return
11
    if typing.get_args(schema):
12
        # 如果有泛型参数（如 Annotated），也认为是有效的
13
        return
14
    # 否则发出警告
15
    warnings.warn(
16
        f"Invalid state_schema: {schema}. Expected a type or Annotated[type, reducer]. "
17
        "Please provide a valid schema to ensure correct updates.\n"
18
        " See: https://langchain-ai.github.io/langgraph/reference/graphs/#stategraph"
19
    )

1
class _StateNode(Protocol[StateT_contra]):
2
    """最基本的节点协议：只接受状态参数"""
3
    def __call__(self, state: StateT_contra) -> Any: ...

1
class _NodeWithConfig(Protocol[StateT_contra]):
2
    """带配置的节点协议：接受状态和配置参数"""
3
    def __call__(self, state: StateT_contra, config: RunnableConfig) -> Any: ...

1
class _NodeWithWriter(Protocol[StateT_contra]):
2
    """带写入器的节点协议：接受状态和流写入器"""
3
    def __call__(self, state: StateT_contra, *, writer: StreamWriter) -> Any: ...

1
class _NodeWithStore(Protocol[StateT_contra]):
2
    """带存储的节点协议：接受状态和存储对象"""
3
    def __call__(self, state: StateT_contra, *, store: BaseStore) -> Any: ...

1
class _NodeWithWriterStore(Protocol[StateT_contra]):
2
    """带写入器和存储的节点协议"""
3
    def __call__(
4
        self, state: StateT_contra, *, writer: StreamWriter, store: BaseStore
5
    ) -> Any: ...

1
class _NodeWithConfigWriter(Protocol[StateT_contra]):
2
    """带配置和写入器的节点协议"""
3
    def __call__(
4
        self, state: StateT_contra, *, config: RunnableConfig, writer: StreamWriter
5
    ) -> Any: ...

1
class _NodeWithConfigStore(Protocol[StateT_contra]):
2
    """带配置和存储的节点协议"""
3
    def __call__(
4
        self, state: StateT_contra, *, config: RunnableConfig, store: BaseStore
5
    ) -> Any: ...

1
class _NodeWithConfigWriterStore(Protocol[StateT_contra]):
2
    """带配置、写入器和存储的节点协议（最完整的签名）"""
3
    def __call__(
4
        self,
5
        state: StateT_contra,
6
        *,
7
        config: RunnableConfig,
8
        writer: StreamWriter,
9
        store: BaseStore,
10
    ) -> Any: ...

1
StateNode: TypeAlias = Union[
2
    _StateNode[StateT_contra],              # 基础节点
3
    _NodeWithConfig[StateT_contra],         # 带配置
4
    _NodeWithWriter[StateT_contra],         # 带写入器
5
    _NodeWithStore[StateT_contra],          # 带存储
6
    _NodeWithWriterStore[StateT_contra],    # 带写入器和存储
7
    _NodeWithConfigWriter[StateT_contra],   # 带配置和写入器
8
    _NodeWithConfigStore[StateT_contra],    # 带配置和存储
9
    _NodeWithConfigWriterStore[StateT_contra], # 带所有参数
10
    Runnable[StateT_contra, Any],           # 可运行对象
11
]

1
def _get_node_name(node: StateNode) -> str:
2
    """
3
    获取节点名称。
4

5
    尝试从节点的 __name__ 属性获取，如果没有则使用类名。
6
    如果都获取不到，抛出类型错误。
7
    """
8
    try:
9
        return getattr(node, "__name__", node.__class__.__name__)
10
    except AttributeError:
11
        raise TypeError(f"Unsupported node type: {type(node)}")

1
edges: set[tuple[str, str]]
2
# 边的集合，每个边是 (起始节点, 结束节点) 的元组
3

4
nodes: dict[str, StateNodeSpec]
5
# 节点字典，键是节点名，值是节点规范
6

7
branches: defaultdict[str, dict[str, Branch]]
8
# 分支字典，键是源节点名，值是条件名到分支对象的映射
9

10
channels: dict[str, BaseChannel]
11
# 通道字典，键是状态键名，值是通道对象
12
# 💡 通俗理解：每个状态字段都有自己的"通道"（邮箱），节点通过通道读写数据
13

14
managed: dict[str, ManagedValueSpec]
15
# 托管值字典，键是状态键名，值是托管值规范
16

17
schemas: dict[type[Any], dict[str, BaseChannel | ManagedValueSpec]]
18
# 模式字典，键是模式类型，值是键到通道或托管值的映射
19

20
waiting_edges: set[tuple[tuple[str, ...], str]]
21
# 等待边集合，用于多源节点到单目标节点的边
22
# 💡 通俗理解：就像"等所有人到齐再出发"，多个节点必须都完成，才能执行下一个节点
23

24
compiled: bool
25
# 是否已编译的标志
26

27
state_schema: type[StateT]
28
# 状态模式类型
29

30
input_schema: type[InputT]
31
# 输入模式类型
32

33
output_schema: type[OutputT]
34
# 输出模式类型

1
def __init__(
2
    self,
3
    state_schema: type[StateT],           # 状态模式（必需）
4
    config_schema: type[Any] | None = None, # 配置模式（可选）
5
    *,
6
    input_schema: type[InputT] | None = None,  # 输入模式（可选）
7
    output_schema: type[OutputT] | None = None, # 输出模式（可选）
8
    **kwargs: Unpack[DeprecatedKwargs],   # 已弃用的关键字参数
9
) -> None:
10
    """
11
    初始化状态图。
12

13
    处理已弃用的参数（input/output），然后初始化所有数据结构。
14
    最后添加状态、输入和输出模式。
15
    """
16

17
    > 💡 **通俗理解**：初始化就像准备一个空的工作台，设置好工作台的规格（状态模式），准备好各种"邮箱"（通道）。之后你就可以往上面添加工作步骤（节点）了。
18

19
    # 处理已弃用的 input 参数
20
    if (input_ := kwargs.get("input", UNSET)) is not UNSET:
21
        warnings.warn(
22
            "`input` is deprecated and will be removed. Please use `input_schema` instead.",
23
            category=LangGraphDeprecatedSinceV05,
24
            stacklevel=2,
25
        )
26
        if input_schema is None:
27
            input_schema = cast(Union[type[InputT], None], input_)
28

29
    # 处理已弃用的 output 参数
30
    if (output := kwargs.get("output", UNSET)) is not UNSET:
31
        warnings.warn(
32
            "`output` is deprecated and will be removed. Please use `output_schema` instead.",
33
            category=LangGraphDeprecatedSinceV05,
34
            stacklevel=2,
35
        )
36
        if output_schema is None:
37
            output_schema = cast(Union[type[OutputT], None], output)
38

39
    # 初始化数据结构
40
    self.nodes = {}
41
    self.edges = set()
42
    self.branches = defaultdict(dict)
43
    self.schemas = {}
44
    self.channels = {}
45
    self.managed = {}
46
    self.compiled = False
47
    self.waiting_edges = set()
48

49
    # 设置模式
50
    self.state_schema = state_schema
51
    self.input_schema = cast(type[InputT], input_schema or state_schema)
52
    self.output_schema = cast(type[OutputT], output_schema or state_schema)
53
    self.config_schema = config_schema
54

55
    # 添加模式（这会解析模式并创建通道）
56
    self._add_schema(self.state_schema)
57
    self._add_schema(self.input_schema, allow_managed=False)
58
    self._add_schema(self.output_schema, allow_managed=False)

1
@property
2
def _all_edges(self) -> set[tuple[str, str]]:
3
    """
4
    获取所有边的集合（包括普通边和等待边）。
5

6
    等待边会被展开为多个普通边。
7
    """
8
    return self.edges | {
9
        (start, end) for starts, end in self.waiting_edges for start in starts
10
    }

1
def _add_schema(self, schema: type[Any], /, allow_managed: bool = True) -> None:
2
    """
3
    添加模式到图中。
4

5
    解析模式中的字段，创建相应的通道和托管值。
6
    如果模式已存在，则跳过。
7

8
    参数：
9
    - schema: 要添加的模式类型
10
    - allow_managed: 是否允许托管值（输入/输出模式不允许）
11
    """
12

13
    > 💡 **通俗理解**：模式就像是一个"表格模板"，定义了状态有哪些字段、每个字段是什么类型。这个方法会为每个字段创建一个"邮箱"（通道），节点可以通过这些邮箱传递数据。`allow_managed=False` 的意思是：输入/输出模式不允许使用"托管值"（一种特殊的通道类型，由系统自动管理）。
14

15
    if schema not in self.schemas:
16
        # 警告无效的模式
17
        _warn_invalid_state_schema(schema)
18
        # 获取通道、托管值和类型提示
19
        channels, managed, type_hints = _get_channels(schema)
20

21
        # 检查托管值是否允许
22
        if managed and not allow_managed:
23
            names = ", ".join(managed)
24
            schema_name = getattr(schema, "__name__", "")
25
            raise ValueError(
26
                f"Invalid managed channels detected in {schema_name}: {names}."
27
                " Managed channels are not permitted in Input/Output schema."
28
            )
29

30
        # 保存模式信息
31
        self.schemas[schema] = {**channels, **managed}
32

33
        # 添加通道
34
        for key, channel in channels.items():
35
            if key in self.channels:
36
                if self.channels[key] != channel:
37
                    # 如果通道类型不同，且不是 LastValue，则报错
38
                    if isinstance(channel, LastValue):
39
                        pass  # LastValue 可以兼容
40
                    else:
41
                        raise ValueError(
42
                            f"Channel '{key}' already exists with a different type"
43
                        )
44
            else:
45
                self.channels[key] = channel
46

47
        # 添加托管值
48
        for key, managed in managed.items():
49
            if key in self.managed:
50
                if self.managed[key] != managed:
51
                    raise ValueError(
52
                        f"Managed value '{key}' already exists with a different type"
53
                    )
54
            else:
55
                self.managed[key] = managed

1
def add_edge(self, start_key: str | list[str], end_key: str) -> Self:
2
    """
3
    添加有向边到图中。
4

5
    支持两种模式：
6
    1. 单源到单目标：start_key 是字符串
7
    2. 多源到单目标：start_key 是字符串列表（等待所有源节点完成）
8

9
    参数：
10
    - start_key: 起始节点（或节点列表）
11
    - end_key: 结束节点
12

13
    返回：
14
    - Self: 允许方法链式调用
15
    """
16

17
    > 💡 **通俗理解**：添加边就像是在工作流程中画箭头，表示"先做这个，再做那个"。
18
    > - 单源边：A → B（A 完成后执行 B）
19
    > - 多源边：[A, B, C] → D（A、B、C 都完成后才执行 D，就像等所有同事都到齐再开会）
20

21
    if self.compiled:
22
        logger.warning(
23
            "Adding an edge to a graph that has already been compiled. This will "
24
            "not be reflected in the compiled graph."
25
        )
26

27
    # 处理单源情况
28
    if isinstance(start_key, str):
29
        if start_key == END:
30
            raise ValueError("END cannot be a start node")
31
        if end_key == START:
32
            raise ValueError("START cannot be an end node")
33

34
        # 验证（仅对非 StateGraph 图）
35
        if not hasattr(self, "channels") and start_key in set(
36
            start for start, _ in self.edges
37
        ):
38
            raise ValueError(
39
                f"Already found path for node '{start_key}'.\n"
40
                "For multiple edges, use StateGraph with an Annotated state key."
41
            )
42

43
        self.edges.add((start_key, end_key))
44
        return self
45

46
    # 处理多源情况
47
    for start in start_key:
48
        if start == END:
49
            raise ValueError("END cannot be a start node")
50
        if start not in self.nodes:
51
            raise ValueError(f"Need to add_node `{start}` first")
52
    if end_key == START:
53
        raise ValueError("START cannot be an end node")
54
    if end_key != END and end_key not in self.nodes:
55
        raise ValueError(f"Need to add_node `{end_key}` first")
56

57
    # 添加到等待边集合
58
    self.waiting_edges.add((tuple(start_key), end_key))
59
    return self

1
def add_conditional_edges(
2
    self,
3
    source: str,  # 源节点
4
    path: Callable[..., Hashable | list[Hashable]]
5
    | Callable[..., Awaitable[Hashable | list[Hashable]]]
6
    | Runnable[Any, Hashable | list[Hashable]],  # 路径函数
7
    path_map: dict[Hashable, str] | list[str] | None = None,  # 路径映射
8
) -> Self:
9
    """
10
    添加条件边。
11

12
    条件边允许根据状态动态选择下一个节点。
13

14
    参数：
15
    - source: 源节点名
16
    - path: 路径函数，根据状态返回下一个节点名（或节点名列表）
17
    - path_map: 可选的路径映射，将路径函数的返回值映射到节点名
18

19
    返回：
20
    - Self: 允许方法链式调用
21
    """
22

23
    > 💡 **通俗理解**：条件边就像是一个"智能路口"，根据当前情况（状态）决定走哪条路。
24
    > - `path` 函数就像是一个"导航员"，查看当前状态后决定下一步去哪里
25
    > - `path_map` 就像一个"翻译表"，把导航员的决定（可能是数字、字符串）翻译成实际的节点名
26
    > - 例如：如果用户输入是问题，就去"回答节点"；如果是命令，就去"执行节点"
27

28
    if self.compiled:
29
        logger.warning(
30
            "Adding an edge to a graph that has already been compiled. This will "
31
            "not be reflected in the compiled graph."
32
        )
33

34
    # 将路径函数转换为可运行对象
35
    path = coerce_to_runnable(path, name=None, trace=True)
36
    name = path.name or "condition"
37

38
    # 验证条件名唯一性
39
    if name in self.branches[source]:
40
        raise ValueError(
41
            f"Branch with name `{path.name}` already exists for node `{source}`"
42
        )
43

44
    # 创建分支并保存
45
    self.branches[source][name] = Branch.from_path(path, path_map, True)
46

47
    # 如果分支有输入模式，添加它
48
    if schema := self.branches[source][name].input_schema:
49
        self._add_schema(schema)
50
    return self

1
def add_sequence(
2
    self,
3
    nodes: Sequence[StateNode[StateT] | tuple[str, StateNode[StateT]]],
4
) -> Self:
5
    """
6
    添加节点序列。
7

8
    按顺序添加多个节点，并在它们之间创建边。
9

10
    参数：
11
    - nodes: 节点序列，可以是 StateNode 或 (名称, StateNode) 元组
12

13
    返回：
14
    - Self: 允许方法链式调用
15
    """
16
    if len(nodes) < 1:
17
        raise ValueError("Sequence requires at least one node.")
18

19
    previous_name: str | None = None
20
    for node in nodes:
21
        # 确定节点名称
22
        if isinstance(node, tuple) and len(node) == 2:
23
            name, node = node
24
        else:
25
            name = _get_node_name(node)
26

27
        # 验证节点名唯一性
28
        if name in self.nodes:
29
            raise ValueError(
30
                f"Node names must be unique: node with the name '{name}' already exists. "
31
                "If you need to use two different runnables/callables with the same name (for example, using `lambda`), please provide them as tuples (name, runnable/callable)."
32
            )
33

34
        # 添加节点
35
        self.add_node(name, node)
36

37
        # 如果不是第一个节点，添加边
38
        if previous_name is not None:
39
            self.add_edge(previous_name, name)
40

41
        previous_name = name
42

43
    return self

1
def set_entry_point(self, key: str) -> Self:
2
    """
3
    设置图的入口点。
4

5
    等价于调用 add_edge(START, key)。
6

7
    参数：
8
    - key: 入口节点名
9

10
    返回：
11
    - Self: 允许方法链式调用
12
    """
13
    return self.add_edge(START, key)

1
def set_conditional_entry_point(
2
    self,
3
    path: Callable[..., Hashable | list[Hashable]]
4
    | Callable[..., Awaitable[Hashable | list[Hashable]]]
5
    | Runnable[Any, Hashable | list[Hashable]],
6
    path_map: dict[Hashable, str] | list[str] | None = None,
7
) -> Self:
8
    """
9
    设置条件入口点。
10

11
    允许根据输入动态选择第一个节点。
12

13
    参数：
14
    - path: 路径函数
15
    - path_map: 路径映射
16

17
    返回：
18
    - Self: 允许方法链式调用
19
    """
20
    return self.add_conditional_edges(START, path, path_map)

1
def set_finish_point(self, key: str) -> Self:
2
    """
3
    设置图的结束点。
4

5
    等价于调用 add_edge(key, END)。
6

7
    参数：
8
    - key: 结束节点名
9

10
    返回：
11
    - Self: 允许方法链式调用
12
    """
13
    return self.add_edge(key, END)

1
def validate(self, interrupt: Sequence[str] | None = None) -> Self:
2
    """
3
    验证图的结构。
4

5
    检查：
6
    1. 所有边的源节点都存在
7
    2. 图有入口点
8
    3. 所有边的目标节点都存在
9
    4. 中断节点存在
10

11
    参数：
12
    - interrupt: 中断节点列表（可选）
13

14
    返回：
15
    - Self: 允许方法链式调用
16

17
    副作用：
18
    - 设置 self.compiled = True
19
    """
20

21
    > 💡 **通俗理解**：验证就像是在"施工前检查图纸"，确保：
22
    > - 所有箭头都指向存在的节点（不能指向空）
23
    > - 有起点（不能没有入口）
24
    > - 所有目标节点都存在（不能指向不存在的节点）
25
    > - 如果设置了中断点，这些节点必须存在
26
    > 如果检查通过，就标记为"已验证"，可以编译了。
27

28
    # 收集所有源节点
29
    all_sources = {src for src, _ in self._all_edges}
30
    for start, branches in self.branches.items():
31
        all_sources.add(start)
32
    for name, spec in self.nodes.items():
33
        if spec.ends:
34
            all_sources.add(name)
35

36
    # 验证源节点
37
    for source in all_sources:
38
        if source not in self.nodes and source != START:
39
            raise ValueError(f"Found edge starting at unknown node '{source}'")
40

41
    # 验证有入口点
42
    if START not in all_sources:
43
        raise ValueError(
44
            "Graph must have an entrypoint: add at least one edge from START to another node"
45
        )
46

47
    # 收集所有目标节点
48
    all_targets = {end for _, end in self._all_edges}
49
    for start, branches in self.branches.items():
50
        for cond, branch in branches.items():
51
            if branch.ends is not None:
52
                for end in branch.ends.values():
53
                    if end not in self.nodes and end != END:
54
                        raise ValueError(
55
                            f"At '{start}' node, '{cond}' branch found unknown target '{end}'"
56
                        )
57
                    all_targets.add(end)
58
            else:
59
                # 如果没有指定目标，可能路由到任何节点
60
                all_targets.add(END)
61
                for node in self.nodes:
62
                    if node != start:
63
                        all_targets.add(node)
64
    for name, spec in self.nodes.items():
65
        if spec.ends:
66
            all_targets.update(spec.ends)
67

68
    # 验证目标节点
69
    for target in all_targets:
70
        if target not in self.nodes and target != END:
71
            raise ValueError(f"Found edge ending at unknown node `{target}`")
72

73
    # 验证中断节点
74
    if interrupt:
75
        for node in interrupt:
76
            if node not in self.nodes:
77
                raise ValueError(f"Interrupt node `{node}` not found")
78

79
    self.compiled = True
80
    return self

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[StateT, InputT, OutputT]:
12
    """
13
    编译状态图为 CompiledStateGraph。
14

15
    编译后的图实现了 Runnable 接口，可以调用、流式传输、批处理和异步运行。
16

17
    主要步骤：
18
    1. 验证图结构
19
    2. 准备输出通道和流通道
20
    3. 创建 CompiledStateGraph 实例
21
    4. 附加所有节点、边和分支
22

23
    返回：
24
    - CompiledStateGraph: 编译后的状态图
25
    """
26

27
    > 💡 **通俗理解**：编译就像把"设计图纸"变成"实际可运行的机器"。
28
    > - 先检查图纸有没有问题（validate）
29
    > - 准备好输出接口（output_channels）和流式接口（stream_channels）
30
    > - 创建一个"执行引擎"（CompiledStateGraph）
31
    > - 把所有节点、边、分支都"安装"到引擎上
32
    > 编译完成后，你就可以调用 `invoke()` 或 `stream()` 来运行这个图了！
33

34
    # 设置默认值
35
    interrupt_before = interrupt_before or []
36
    interrupt_after = interrupt_after or []
37

38
    # 验证图
39
    self.validate(
40
        interrupt=(
41
            (interrupt_before if interrupt_before != "*" else []) + interrupt_after
42
            if interrupt_after != "*"
43
            else []
44
        )
45
    )
46

47
    # 准备输出通道
48
    output_channels = (
49
        "__root__"
50
        if len(self.schemas[self.output_schema]) == 1
51
        and "__root__" in self.schemas[self.output_schema]
52
        else [
53
            key
54
            for key, val in self.schemas[self.output_schema].items()
55
            if not is_managed_value(val)
56
        ]
57
    )
58

59
    # 准备流通道
60
    stream_channels = (
61
        "__root__"
62
        if len(self.channels) == 1 and "__root__" in self.channels
63
        else [
64
            key for key, val in self.channels.items() if not is_managed_value(val)
65
        ]
66
    )
67

68
    # 创建编译后的图
69
    compiled = CompiledStateGraph[StateT, InputT, OutputT](
70
        builder=self,
71
        schema_to_mapper={},
72
        config_type=self.config_schema,
73
        nodes={},
74
        channels={
75
            **self.channels,
76
            **self.managed,
77
            START: EphemeralValue(self.input_schema),
78
        },
79
        input_channels=START,
80
        stream_mode="updates",
81
        output_channels=output_channels,
82
        stream_channels=stream_channels,
83
        checkpointer=checkpointer,
84
        interrupt_before_nodes=interrupt_before,
85
        interrupt_after_nodes=interrupt_after,
86
        auto_validate=False,
87
        debug=debug,
88
        store=store,
89
        cache=cache,
90
        name=name or "LangGraph",
91
    )
92

93
    # 附加 START 节点
94
    compiled.attach_node(START, None)
95

96
    # 附加所有节点
97
    for key, node in self.nodes.items():
98
        compiled.attach_node(key, node)
99

100
    # 附加所有边
101
    for start, end in self.edges:
102
        compiled.attach_edge(start, end)
103

104
    # 附加所有等待边
105
    for starts, end in self.waiting_edges:
106
        compiled.attach_edge(starts, end)
107

108
    # 附加所有分支
109
    for start, branches in self.branches.items():
110
        for name, branch in branches.items():
111
            compiled.attach_branch(start, name, branch)
112

113
    return compiled.validate()

1
class CompiledStateGraph(
2
    Pregel[StateT, InputT, OutputT], Generic[StateT, InputT, OutputT]
3
):
4
    """
5
    编译后的状态图。
6

7
    继承自 Pregel，实现了图的执行逻辑。
8
    """
9
    builder: StateGraph[StateT, InputT, OutputT]
10
    # 构建器引用，保存原始的状态图定义
11

12
    schema_to_mapper: dict[type[Any], Callable[[Any], Any] | None]
13
    # 模式到映射函数的字典，用于将状态字典转换为模式对象

1
def __init__(
2
    self,
3
    *,
4
    builder: StateGraph[StateT, InputT, OutputT],
5
    schema_to_mapper: dict[type[Any], Callable[[Any], Any] | None],
6
    **kwargs: Any,
7
) -> None:
8
    """
9
    初始化编译后的状态图。
10

11
    调用父类 Pregel 的初始化方法，然后保存构建器和模式映射器。
12
    """
13
    super().__init__(**kwargs)
14
    self.builder = builder
15
    self.schema_to_mapper = schema_to_mapper

1
def get_input_jsonschema(
2
    self, config: RunnableConfig | None = None
3
) -> dict[str, Any]:
4
    """
5
    获取输入的 JSON Schema。
6

7
    用于 API 文档生成和验证。
8
    """
9
    return _get_json_schema(
10
        typ=self.builder.input_schema,
11
        schemas=self.builder.schemas,
12
        channels=self.builder.channels,
13
        name=self.get_name("Input"),
14
    )

1
def get_output_jsonschema(
2
    self, config: RunnableConfig | None = None
3
) -> dict[str, Any]:
4
    """
5
    获取输出的 JSON Schema。
6

7
    用于 API 文档生成和验证。
8
    """
9
    return _get_json_schema(
10
        typ=self.builder.output_schema,
11
        schemas=self.builder.schemas,
12
        channels=self.builder.channels,
13
        name=self.get_name("Output"),
14
    )

1
def attach_node(self, key: str, node: StateNodeSpec | None) -> None:
2
    """
3
    将节点附加到编译后的图中。
4

5
    这是编译过程的关键步骤，将 StateGraph 的节点转换为 Pregel 节点。
6

7
    主要工作：
8
    1. 确定输出键
9
    2. 创建更新函数（处理节点返回值）
10
    3. 创建写入条目（状态更新和分支控制）
11
    4. 创建 PregelNode 并添加到图中
12

13
    参数：
14
    - key: 节点键（START 或节点名）
15
    - node: 节点规范（START 节点为 None）
16
    """
17

18
    > 💡 **通俗理解**：这个方法把"设计图纸上的节点"转换成"实际可执行的节点"。
19
    > - 确定节点可以更新哪些状态字段（output_keys）
20
    > - 创建一个"转换器"（_get_updates），把节点的返回值转换成状态更新
21
    > - 创建"写入器"（write_entries），负责把更新写入状态，并控制下一步去哪里
22
    > - 最后创建一个 PregelNode（实际执行单元）并添加到图中
23
    > 就像把"工作步骤说明"转换成"实际的工作指令"。
24

25
    # 确定输出键
26
    if key == START:
27
        # START 节点输出输入模式的所有键
28
        output_keys = [
29
            k
30
            for k, v in self.builder.schemas[self.builder.input_schema].items()
31
            if not is_managed_value(v)
32
        ]
33
    else:
34
        # 普通节点输出所有通道和托管值
35
        output_keys = list(self.builder.channels) + [
36
            k for k, v in self.builder.managed.items()
37
        ]
38

39
    def _get_updates(
40
        input: None | dict | Any,
41
    ) -> Sequence[tuple[str, Any]] | None:
42
        """
43
        将节点返回值转换为更新元组列表。
44

45
        处理多种返回类型：
46
        - None: 无更新
47
        - dict: 字典键值对
48
        - Command: 命令对象
49
        - list/tuple: 命令列表
50
        - 带注解的对象: 使用 get_update_as_tuples
51
        """
52
        if input is None:
53
            return None
54
        elif isinstance(input, dict):
55
            return [(k, v) for k, v in input.items() if k in output_keys]
56
        elif isinstance(input, Command):
57
            if input.graph == Command.PARENT:
58
                return None
59
            return [
60
                (k, v) for k, v in input._update_as_tuples() if k in output_keys
61
            ]
62
        elif (
63
            isinstance(input, (list, tuple))
64
            and input
65
            and any(isinstance(i, Command) for i in input)
66
        ):
67
            updates: list[tuple[str, Any]] = []
68
            for i in input:
69
                if isinstance(i, Command):
70
                    if i.graph == Command.PARENT:
71
                        continue
72
                    updates.extend(
73
                        (k, v) for k, v in i._update_as_tuples() if k in output_keys
74
                    )
75
                else:
76
                    updates.extend(_get_updates(i) or ())
77
            return updates
78
        elif (t := type(input)) and get_cached_annotated_keys(t):
79
            return get_update_as_tuples(input, output_keys)
80
        else:
81
            msg = create_error_message(
82
                message=f"Expected dict, got {input}",
83
                error_code=ErrorCode.INVALID_GRAPH_NODE_RETURN_VALUE,
84
            )
85
            raise InvalidUpdateError(msg)
86

87
    # 创建写入条目
88
    write_entries: tuple[ChannelWriteEntry | ChannelWriteTupleEntry, ...] = (
89
        ChannelWriteTupleEntry(
90
            mapper=_get_root if output_keys == ["__root__"] else _get_updates
91
        ),
92
        ChannelWriteTupleEntry(
93
            mapper=_control_branch,
94
            static=_control_static(node.ends)
95
            if node is not None and node.ends is not None
96
            else None,
97
        ),
98
    )
99

100
    # 添加节点
101
    if key == START:
102
        # START 节点
103
        self.nodes[key] = PregelNode(
104
            tags=[TAG_HIDDEN],
105
            triggers=[START],
106
            channels=START,
107
            writers=[ChannelWrite(write_entries)],
108
        )
109
    elif node is not None:
110
        # 普通节点
111
        input_schema = node.input if node else self.builder.state_schema
112
        input_channels = list(self.builder.schemas[input_schema])
113
        is_single_input = len(input_channels) == 1 and "__root__" in input_channels
114

115
        # 获取或创建映射器
116
        if input_schema in self.schema_to_mapper:
117
            mapper = self.schema_to_mapper[input_schema]
118
        else:
119
            mapper = _pick_mapper(input_channels, input_schema)
120
            self.schema_to_mapper[input_schema] = mapper
121

122
        # 创建分支通道
123
        branch_channel = CHANNEL_BRANCH_TO.format(key)
124
        self.channels[branch_channel] = (
125
            LastValueAfterFinish(Any)
126
            if node.defer
127
            else EphemeralValue(Any, guard=False)
128
        )
129

130
        # 创建 PregelNode
131
        self.nodes[key] = PregelNode(
132
            triggers=[branch_channel],
133
            channels=("__root__" if is_single_input else input_channels),
134
            mapper=mapper,
135
            writers=[ChannelWrite(write_entries)],
136
            metadata=node.metadata,
137
            retry_policy=node.retry_policy,
138
            cache_policy=node.cache_policy,
139
            bound=node.runnable,
140
        )
141
    else:
142
        raise RuntimeError