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 │ └──────────────────────┘ ↓ B2. 子图中的状态如何传递
可以把子图执行过程近似理解成:
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 + 新 messages3. 三种 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_schema9. 什么是“私有状态”
私有状态最好理解成:
属于某一张 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这实际上就是一个完整的封装。