LangGraph基础入门6(小白必看)- 惊觉

LangGraph 子图实战：父图嵌套、持久化策略、流式输出与动态路由

当 LangGraph 项目只有三四个节点时，一个 StateGraph 就够用了；但一旦开始做 RAG、工具调用、多轮对话、任务规划，所有逻辑都写在同一张图里会很快失控。

这时就需要子图（Subgraph）。

子图的核心价值是把复杂 Agent 拆成多个可独立运行、测试和复用的小模块：

父图：负责整体编排
子图：负责一个明确的局部任务

本文集中讲清楚 LangGraph 子图最容易混淆的几个问题：怎么嵌入、状态如何保存、流式 chunk 怎么看、怎么动态选择子图，以及它和 LangChain Agent 有什么关系。

一、什么是父图和子图？

先看一个学习助手的结构：

用户请求
   ↓
父图：识别任务
   ├── 检索子图：查知识库
   ├── 写作子图：生成内容
   └── 审核子图：检查质量
   ↓
父图：汇总最终回答

这里：

父图负责决定整体流程和模块之间的顺序；
子图负责完成一个相对独立的局部流程；
子图本身也是一个完整的 StateGraph，同样需要 compile()。

可以把子图理解为“图中的一个大节点”，只是这个节点内部不是一行函数，而是另一张图。

二、两种把子图放进父图的方式

1. 在父图节点函数中调用子图

这是最灵活的一种方式：父图先进入一个普通节点，节点内部再调用 subgraph.invoke()。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class SubgraphState(TypedDict):
    raw_text: str
    clean_text: str

def clean_node(state: SubgraphState):
    return {"clean_text": state["raw_text"].strip()}

sub_builder = StateGraph(SubgraphState)
sub_builder.add_node("clean_node", clean_node)
sub_builder.add_edge(START, "clean_node")
sub_builder.add_edge("clean_node", END)
subgraph = sub_builder.compile()

class ParentState(TypedDict):
    input_text: str
    output_text: str

def call_subgraph(state: ParentState):
    sub_res = subgraph.invoke({"raw_text": state["input_text"]})
    return {"output_text": sub_res["clean_text"]}

这个例子的执行过程是：

父图拿到 input_text
  ↓
call_subgraph 节点手动调用子图
  ↓
父图把 input_text 改名为 raw_text 传给子图
  ↓
子图返回 clean_text
  ↓
父图把 clean_text 改名为 output_text

它的优点是父图、子图可以使用不同的状态字段。特别适合做输入转换、字段筛选、结果再加工。

2. 直接把子图注册为父图节点

如果父图与子图的状态字段能够直接对接，可以把编译后的子图直接传给 add_node()：

subgraph = sub_builder.compile()

parent_builder = StateGraph(SharedState)
parent_builder.add_node("clean_subgraph", subgraph)
parent_builder.add_edge(START, "clean_subgraph")
parent_builder.add_edge("clean_subgraph", END)

这种写法更短，但要求更高：子图读写的字段必须能在父图的共享状态中找到。

简单记忆：

字段需要转换或想加额外逻辑 -> 在节点函数中 invoke 子图
字段已经共享、能直接衔接 -> 子图直接作为父图节点

三、子图的三种状态策略

子图使用检查点时，真正关键的不是“是否要保存”，而是“子图的历史要不要跨调用延续”。

模式	编译方式	子图是否保存自身历史	适合什么
Per-invocation	`subgraph.compile()`	每次调用独立	一次性处理任务
Per-thread	`subgraph.compile(checkpointer=True)`	同一线程持续保留	多轮对话、中断恢复
Stateless	`subgraph.compile(checkpointer=False)`	不保存	纯函数式清洗、转换

1. Per-invocation：默认按“本次调用”隔离

subgraph = sub_builder.compile()

这表示每次父图调用子图时，子图都视为一次独立任务。

它常用于：

单次文本清洗
单次文档解析
一次性的格式转换
不需要聊天上下文的工具流程

虽然可以观察本次子图的执行，但不同调用之间的子图历史不会像对话一样连续。

2. Per-thread：让子图拥有线程级记忆

subgraph = sub_builder.compile(checkpointer=True)

True 的含义是：子图使用父图提供的检查点机制，并在相同 thread_id 下延续自己的状态。

这适合：

子图内部做多轮对话
子图内部触发 interrupt()，后续需要恢复
子图需要记住上一轮处理结果

但它也有代价：如果父图在同一个线程里反复调用同一个有状态子图，前一次输入与输出可能留在子图历史里，从而影响下一次结果。

例如子图状态使用了：

messages: Annotated[list, add]

那么新消息会追加到旧消息后面，而不是自动清空。多次调用后，历史会越来越长。

3. Stateless：明确让子图完全无状态

subgraph = sub_builder.compile(checkpointer=False)

无状态子图每次都只处理当前输入，不保存检查点、不支持基于检查点恢复，也不保留多轮对话。

它适合“输入确定，输出确定”的局部处理，例如：

去除首尾空格
给文本补句号
格式规范化
字段映射

如果你发现同一个子图被多次调用后出现结果重复、旧消息串进新任务，优先检查自己是不是无意中使用了 Per-thread。

四、子图检查点怎么看？

父图开启检查点后，子图的状态通常也会以命名空间的形式出现在父图快照中。

你可以先获取父图的历史：

history = list(parent_graph.get_state_history(config))

每个检查点快照里通常会包含：

values：当时的状态值；
next：从这个快照继续时，要执行的下一个节点；
config：包括 thread_id、checkpoint_id、命名空间等信息；
tasks：该步骤的任务与可能发生的中断。

对子图来说，命名空间用于区分“这是父图状态，还是哪一个子图调用的状态”。

父图节点名 + 子图调用信息 -> 子图检查点命名空间

因此，父图不仅能知道“调用过子图”，还可以继续定位和查看子图内部执行到了哪里。

五、子图流式输出：chunk 为什么多了一层？

父图默认 stream() 时，通常只会返回父图的流数据。

如果希望连同子图内部的节点更新或消息增量一起拿到，需要加：

for chunk in parent_graph.stream(
    {"input_text": "LangGraph 真有意思"},
    stream_mode=["updates"],
    subgraphs=True,
):
    print(chunk)

开启 subgraphs=True 后，子图并不会替代父图输出，而是父图和子图的流数据一起返回。

这时 chunk 常见结构是：

(namespace, stream_mode, data)

例如：

(
    ("call_subgraph:任务ID",),
    "updates",
    {"strip_node": {"stripped_text": "LangGraph 真有意思"}}
)

三个部分含义如下：

部分	含义
`namespace`	数据来自父图还是哪一个子图
`stream_mode`	本段数据属于 `updates`、`messages` 等哪种模式
`data`	真实的状态更新或消息片段

通常：

()                       -> 父图自身输出
("call_subgraph:...",)  -> 来自普通子图调用
("call_subgraph",)      -> 来自 Per-thread 子图调用

所以前端拿到 chunk 后，不能只看数据本身，还应先看 namespace，否则无法判断消息到底来自主流程还是某个子任务。

多轮对话中的 messages 流

如果子图内部是聊天模型，流数据里会出现：

AIMessageChunk(...)

常见处理方式：

for namespace, mode, data in parent_graph.stream(
    input_data,
    stream_mode=["messages"],
    subgraphs=True,
):
    message_chunk, metadata = data

    if namespace == ("call_subgraph",):
        print(message_chunk.content, end="", flush=True)

这段代码的意思是：只把 call_subgraph 这个子图产生的模型文本，实时显示到页面上。

六、动态路由：运行时选择调用哪个子图

父图不一定只能固定调用一个子图。它可以先识别用户意图，再动态决定进入哪个模块：

用户输入
  ↓
意图识别节点
  ├── 水果介绍子图
  ├── 蔬菜介绍子图
  └── 结构化信息抽取子图

实现思路还是条件边：

def route_subgraph(state) -> str:
    if state["task_type"] == "fruit":
        return "fruit_subgraph"
    if state["task_type"] == "vegetable":
        return "vegetable_subgraph"
    return "extract_subgraph"

builder.add_conditional_edges(
    "router_node",
    route_subgraph,
    path_map=["fruit_subgraph", "vegetable_subgraph", "extract_subgraph"],
)

这里“动态”的意思不是在运行时创建新的图，而是：所有子图先注册好，运行时再决定这一次走哪一个。

如果路由判断依赖 LLM，建议让模型输出结构化字段，例如：

task_type: Literal["fruit", "vegetable", "extract"]

这样比让模型自由生成“我觉得应该去水果模块”更稳定。

七、LangChain Agent 和 LangGraph 是什么关系？

这两个不是互相替代，而是不同层级的工具。

名称	更擅长解决什么
LangChain Agent	快速创建模型 + 工具调用 Agent
LangGraph	明确控制状态、分支、循环、中断、检查点和多 Agent 协作

可以这样理解：

LangChain：提供模型、消息、工具、提示词等组件
LangGraph：把这些组件组织成有状态、可恢复、可观测的运行图

简单工具调用可以直接使用 LangChain Agent；当你需要下面这些能力时，更适合用 LangGraph 承载：

工具调用前人工审批；
多轮中断与恢复；
复杂条件分支；
多个子图协作；
检查点、回放与状态修改；
对执行过程做流式展示与调试。

因此，一个常见工程组合是：

LangChain 提供 ChatModel、Tools、Prompt
LangGraph 把它们封装进节点、边、子图和状态管理

八、子图设计建议

写子图时可以先问自己四个问题：

这个模块能否单独定义输入和输出？
它是否需要跨调用保留历史？
它会不会在同一线程内被重复调用？
前端是否需要看到它的中间流式输出？

对应选择：

单次独立任务 -> Per-invocation 或 Stateless
多轮对话 / 中断恢复 -> Per-thread
同一子图反复调用 -> 小心历史串扰，必要时隔离线程或改无状态
需要显示内部进度 -> 父图 stream(..., subgraphs=True)

九、总结

子图让 LangGraph 从“能画流程图”，真正走向“能组织复杂系统”。

本文最重要的结论可以压缩成下面几句：

子图是可复用的已编译图。
要转字段就在节点里调用子图；状态共享时可直接把子图作为节点。
Per-thread 有记忆，但要注意历史串扰；Stateless 最干净，但不支持恢复和多轮记忆。
开启 subgraphs=True 后，要通过 namespace 区分父图和子图的流数据。
动态路由是在运行时选择已注册的子图，不是临时创建新节点。
LangChain 提供组件，LangGraph 负责把组件组织成可控的 Agent 系统。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/agood_day_/article/details/163733280

LangGraph基础入门6(小白必看)