☰
LangGraph 子图与状态传递笔记
2026/10/8 6:46:37 网站建设 项目流程

1. 什么是子图

子图本质上也是一个StateGraph。

先创建并编译:

sub_builder = StateGraph(SubState)# 添加节点、边...subgraph = sub_builder.compile()

然后可以直接把这个已经compile()的图,当成主图里的一个节点:

builder.add_node("reflection", subgraph)

所以可以记成:

普通节点 = Python 函数 子图节点 = compile() 后的 StateGraph

主图看来:

A ↓ reflection ↓ B

实际上执行到reflection时:

A ↓ ┌──────── 子图 ────────┐ │ START │ │ ↓ │ │ node1 │ │ ↓ │ │ node2 │ │ ↓ │ │ END │ └──────────────────────┘ ↓ B

2. 子图中的状态如何传递

可以把子图执行过程近似理解成:

result = subgraph.invoke(current_state)

虽然实际上这是 LangGraph 内部自动调度的,但这样理解最直观。

整个过程:

主图当前 State ↓ 进入子图 ↓ 子图获得输入状态 ↓ 子图内部节点不断更新 State ↓ END ↓ 子图产生输出 ↓ 更新/合并到主图 State ↓ 主图继续运行

例如主图当前:

{ "user_message": "帮我请假", "task_result": { "isSuccess": True }}

进入反思子图。

子图执行:

return { "reflection": "任务执行正确"}

回来以后主图就可以继续使用:

{ "user_message": "帮我请假", "task_result": { "isSuccess": True }, "reflection": "任务执行正确"}

需要注意,状态不是简单粗暴地全部覆盖,而是仍然遵守 LangGraph 的 reducer 规则。

普通字段通常类似覆盖:

旧值 → 新值

如果定义:

messages: Annotated[list, add]

则会:

旧 messages + 新 messages

3. 三种 Schema

LangGraph 中最重要的是:

state_schemainput_schemaoutput_schema

可以直接记成:

input_schema ↓ 进入图 ↓ state_schema ↓ 图内部运行 ↓ output_schema ↓ 离开图

4. state_schema:内部完整状态空间

例如:

class State(TypedDict): user_message: str intent: str task_result: dict reflection: str response: str

然后:

builder = StateGraph(State)

state_schema表示:

这张图运行过程中有哪些正式状态字段。

比如开始可能只有:

{ "user_message": "帮我请假"}

运行识别节点之后:

{ "user_message": "帮我请假", "intent": "submit_leave"}

执行任务后:

{ "user_message": "帮我请假", "intent": "submit_leave", "task_result": {...}}

所以:

state_schema是图运行过程中完整的工作空间,不代表一开始所有字段都有值。


5. input_schema:图的输入接口

例如:

class InputState(TypedDict): user_message: str

完整状态:

class State(TypedDict): user_message: str intent: str task_result: dict response: str

定义:

builder = StateGraph( State, input_schema=InputState)

那么调用图时:

graph.invoke({ "user_message": "帮我请假"})

即可。

不需要:

graph.invoke({ "user_message": "...", "intent": None, "task_result": None, "response": None})

因为:

input_schema决定图入口需要接受什么。

而其余字段:

intent task_result response

由图运行过程中逐步产生。

可以类比:

def graph(user_message): # intent 是内部变量 # task_result 是内部变量 # response 是内部变量

所以:

input_schema = 对外参数

6. 子图设置 input_schema 时

这一点对理解子图非常重要。

假设主图当前状态:

{ "user_message": "帮我请假", "intent": "submit_leave", "task_result": {"isSuccess": True}, "debug_info": {...}}

子图定义:

class ReflectionInput(TypedDict): user_message: str task_result: dict

那么进入子图时,可以理解为从主图当前状态中取:

{ "user_message": "帮我请假", "task_result": {"isSuccess": True}}

作为子图入口。

所以:

主图当前 State ↓ 子图 input_schema ↓ 挑出子图需要的信息 ↓ 进入子图

因此:

子图的input_schema就是子图和外层图之间的输入接口。


7. output_schema:图的输出接口

例如内部状态:

class State(TypedDict): user_message: str intent: str task_result: dict reflection: str response: str

但定义:

class OutputState(TypedDict): reflection: str response: str

那么图结束后,对外主要暴露的是:

{ "reflection": "...", "response": "..."}

而:

intent task_result

可以只作为内部工作状态。

所以:

output_schema决定这个图结束以后向外暴露什么。

对于子图:

主图 State ↓ input_schema ↓ 子图内部 state_schema ↓ END ↓ output_schema ↓ 返回主图

8. 三种 Schema 的核心区别

Schema作用类比
input_schema图允许接收什么函数参数
state_schema图内部完整状态空间内部变量/成员变量
output_schema图最终暴露什么函数返回值

最简单的记忆方式:

input_schema = 怎么进 state_schema = 里面有什么 output_schema = 怎么出

或者:

def graph(input_schema) -> output_schema: # 内部工作空间 = state_schema

9. 什么是“私有状态”

私有状态最好理解成:

属于某一张 Graph 自己内部使用,而不需要暴露给其他 Graph 的状态。

例如主图:

class MainState(TypedDict): user_message: str task_result: dict response: str

反思子图:

class ReflectionState(TypedDict): user_message: str task_result: dict reflection_score: float critique: str reflection: str

其中:

reflection_score critique

只需要 ReflectionGraph 内部使用。

那么它们就是:

ReflectionGraph 的私有状态

而:

user_message task_result reflection

可以作为和主图共享的信息。


10. 私有状态是“图级”的

不是:

某个节点的私有状态

而主要是:

某张 Graph 的内部状态

同一张图中的节点,只要字段属于这张图的状态空间,就都可以使用。

所以可以类比 Java:

Graph ≈ class 节点 ≈ class 中的方法 State ≈ class 中的数据 图私有状态 ≈ private 成员 共享状态 ≈ 对外接口

例如:

ReflectionGraph ├── evaluate_node ├── critique_node └── decision_node

它们都可以共同使用:

state["reflection_score"]

但 MainGraph 不一定需要看到它。


11. 最推荐的子图设计方式

例如你的 Reflection 子图:

class ReflectionInput(TypedDict): user_message: str task_result: dict

内部完整状态:

class ReflectionState(TypedDict): user_message: str task_result: dict score: float critique: str passed: bool

输出:

class ReflectionOutput(TypedDict): passed: bool critique: str

子图:

builder = StateGraph( ReflectionState, input_schema=ReflectionInput, output_schema=ReflectionOutput,)

整个过程就特别清楚:

MainGraph │ user_message task_result │ ↓ ReflectionInput ↓ ┌──── ReflectionGraph ────┐ │ │ │ user_message │ │ task_result │ │ score ← 私有 │ │ critique │ │ passed │ │ │ └─────────────────────────┘ ↓ ReflectionOutput ↓ passed + critique ↓ MainGraph

这实际上就是一个完整的封装。


需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询