LangGraph 高级特性实战：流式执行、子图与六大 Agent 工作流设计模式

前面学习 LangGraph 时，我们已经能用 StateGraph 定义状态、节点和边，也能处理条件分支、记忆和中断。

继续往真实项目走，还需要掌握三个能力：

流式执行：图还没跑完，就把中间结果实时交给前端
子图：把复杂流程拆成可以复用的小流程
设计模式：根据任务特点选择合适的图结构

这篇文章围绕这三部分展开，目标不是背 API，而是知道在什么场景下该怎样组织 LangGraph。

一、流式执行：为什么不只用 invoke()？

平时最常见的写法是：

res = graph.invoke({"topic": "布偶猫"})
print(res)

invoke() 会一直等待，直到整个图执行结束，再一次性返回最终状态。

这对短任务没有问题，但如果图中包含多个 LLM 调用、工具调用或耗时节点，用户会长时间看不到任何反馈。

流式执行（Streaming Execution）就是：图一边执行，一边把状态变化、消息增量或自定义进度事件交给调用方。

invoke：图全部结束 -> 返回最终结果
stream：图执行过程中 -> 持续产出 chunk

1. stream() 与 astream()

LangGraph 提供两套流式 API：

API	使用方式	适合场景
stream()	for chunk in ...	普通 Python 脚本、命令行程序
astream()	async for chunk in ...	FastAPI、WebSocket、异步服务

两者关注的内容和 stream_mode 基本一致，主要区别是调用方式：

# 同步流式输出
for chunk in graph.stream(input_data, stream_mode="updates"):
    print(chunk)

# 异步流式输出
async for chunk in graph.astream(input_data, stream_mode="updates"):
    print(chunk)

可以先这样理解：stream 是同步版，astream 是异步版。

2. stream_mode：决定想看到什么

stream_mode 决定每一个 chunk 里输出哪一种数据。

模式	输出内容	常见用途
values	每个超步之后的完整状态	看完整状态快照
updates	每个节点刚写入的增量更新	观察节点执行结果
messages	消息增量，通常是模型逐字输出	聊天页面打字机效果
checkpoints	检查点事件	监控持久化和恢复
tasks	任务的开始、结束、错误	运行观测与排错
debug	检查点与任务调试信息	开发阶段排查问题
custom	节点主动写出的自定义数据	进度条、阶段通知

3. values：每一步的完整状态

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

class OverAllState(TypedDict):
    initial_state: str
    node_a_output: str
    node_b_output: str

def node_a(state: OverAllState):
    return {"node_a_output": "节点 A 的输出"}

def node_b(state: OverAllState):
    return {"node_b_output": "节点 B 的输出"}

builder = StateGraph(OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
graph = builder.compile()

for mode, chunk in graph.stream(
    {"initial_state": "初始状态"},
    stream_mode=["values"],
):
    print(mode, chunk)

输出状态会逐渐变完整：

初始状态
初始状态 + node_a_output
初始状态 + node_a_output + node_b_output

所以 values 很适合观察“此刻整个图的状态是什么”。缺点是状态很大时，每次传完整状态会比较冗余。

4. updates：只看谁改了什么

把上例改成：

for mode, chunk in graph.stream(
    {"initial_state": "初始状态"},
    stream_mode=["updates"],
):
    print(mode, chunk)

你会得到类似：

("updates", {"node_a": {"node_a_output": "节点 A 的输出"}})
("updates", {"node_b": {"node_b_output": "节点 B 的输出"}})

这里最关键的是：

values：给你当前完整状态
updates：只给你本次节点产生的变化

实际项目中，如果前端只需刷新局部内容，通常 updates 更省数据，也更容易定位“哪个节点做了更新”。

5. messages：实现模型逐字输出

对于聊天机器人，最常用的是 messages。

from langgraph.graph import StateGraph, START, END, MessagesState
from langchain.messages import HumanMessage

def llm_node(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("llm_node", llm_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", END)
graph = builder.compile()

for mode, chunk in graph.stream(
    {"messages": [HumanMessage(content="你好")]},
    stream_mode=["messages"],
):
    message_chunk, metadata = chunk
    print(message_chunk.content, end="", flush=True)

message_chunk.content 是本次新增的文本片段；metadata 里常有当前节点名、图步骤、模型信息等运行元数据。

注意：想真正获得模型的逐 token 输出，底层模型集成也必须支持流式响应。

6. custom：节点主动报告进度

有些信息不是状态，也不是模型消息，例如：

正在读取 PDF...
已完成向量检索...
正在生成报告第 2 部分...

这类内容适合 custom。节点通过 stream_writer 写出自定义事件，前端可以据此展示进度。

from langgraph.config import get_stream_writer

def process_node(state):
    writer = get_stream_writer()
    writer({"stage": "retrieval", "message": "正在检索知识库"})

    # 执行真正业务逻辑
    result = "检索完成"

    writer({"stage": "retrieval", "message": "检索完成"})
    return {"result": result}

调用方监听：

for mode, chunk in graph.stream(input_data, stream_mode=["custom"]):
    print(chunk)

7. astream_events：看 Runnable 级别事件

stream/astream 主要关注图的执行结果和状态变化。

astream_events() 则更偏底层运行追踪：它可以输出 Runnable 的启动、结束、输入、输出以及父子调用关系。做复杂 Agent 调试、链路追踪或接入可观测平台时更有价值。

async for event in graph.astream_events(input_data, version="v2"):
    print(event["event"], event.get("name"))

日常页面流式回复优先用 astream；需要分析内部调用链时，再看 astream_events。

二、子图：把大图拆成可复用模块

当一个 Agent 的流程越来越长，所有节点都堆在一个 StateGraph 中会很难维护。

例如一个“智能学习助手”可能包含：

主图
├── 用户意图识别
├── RAG 检索子图
├── 练习题生成子图
├── 答案审核子图
└── 汇总输出

子图（Subgraph）本质上就是：一个已经 compile() 的 LangGraph 图，被父图当成流程的一部分使用。

1. 两种嵌入方式

方式一：在父图节点函数里调用子图

subgraph = sub_builder.compile()

def call_subgraph_node(state: ParentState):
    sub_res = subgraph.invoke({"topic": state["topic"]})
    return {"summary": sub_res["summary"]}

特点：

• 父节点可以自由转换输入输出
• 父图与子图不必使用完全相同的状态结构
• 适合需要做参数适配、数据清洗、结果二次处理的场景

方式二：直接把子图注册为父图节点

subgraph = sub_builder.compile()

parent_builder.add_node("research_subgraph", subgraph)
parent_builder.add_edge(START, "research_subgraph")

特点：

• 写法更短
• 父图和子图需要有能对接的共享状态字段
• 适合“子图就是一个完整节点”的情况

一句话区分：

需要转换数据：在节点函数里 invoke 子图
状态能够直接对接：子图直接作为父图节点

2. 为什么子图很重要

子图解决的不是“能不能跑”，而是复杂系统的组织问题：

• 复用：同一个检索、审核、工具调用流程可被多个父图使用
• 隔离：每个模块只关注自己的状态和节点
• 协作：不同人维护不同子图，降低相互影响
• 测试：可以单独对一个子图调用 invoke() 和调试

3. 子图持久化：Per-invocation 与 Per-thread

子图和检查点结合时，容易遇到一个关键选择。

策略	含义	适合场景
Per-invocation	每次调用子图都像新任务，不保留子图自身历史	独立、无状态的小任务
Per-thread	同一个 thread 内的子图调用保留历史	多轮对话、可恢复流程

# 默认思路：一次调用对应一次独立的子图执行
subgraph = sub_builder.compile()

# 让子图跟随线程保存自己的检查点与历史
subgraph = sub_builder.compile(checkpointer=True)

如果子图负责多轮对话、interrupt() 后恢复，通常需要 Per-thread。

如果一个父图里多次调用同一个有状态子图，也要警惕历史互相影响：前一次子图的上下文可能影响后一次调用。此时需要设计不同的命名空间、不同线程，或者直接选无状态调用。

4. 子图流式输出

父图流式运行时，也可以开启子图输出。此时 chunk 通常会带有命名空间信息，用来标识数据来自哪一层图：

(命名空间, mode, data)

例如命名空间 ("call_subgraph",) 表示：这一段流数据来自父图中名为 call_subgraph 的子图调用。

这对前端很重要，因为它可以区分：

主图的最终回答
子图中的检索进度
子图中的 LLM 消息片段

三、六种常用运行图设计模式

LangGraph 的节点和边可以自由组合，但工程上有一些反复出现的成熟结构。它们不是固定模板，而是帮助你先判断“这个任务本质上是哪一类”。

1. Prompt Chaining：提示词链

核心思路：把一个复杂任务拆成多个顺序清晰的小步骤，前一步输出作为后一步输入。

用户主题 -> 生成提纲 -> 撰写初稿 -> 润色文章 -> 输出

适合：

• 长文写作
• 报告生成
• 数据提取后再总结
• 先分类、后处理的任务

优点是步骤稳定、容易调试；缺点是路径固定，不能根据中间结果自由调整。

2. Parallelization：并行化

核心思路：多个互不依赖的任务同时执行，最后再汇总。

               -> 生成诗
输入主题 ->    -> 生成笑话  -> 汇总
               -> 生成文案

LangGraph 中可以从同一个上游节点连出多条边：

builder.add_edge(START, "node_poem")
builder.add_edge(START, "node_joke")
builder.add_edge(["node_poem", "node_joke"], "node_output")

这里 node_output 会等待 node_poem 和 node_joke 都完成后再执行，这就是扇入（Fan-in）。

适合：独立检索、多份内容生成、多路数据校验。

3. Routing：路由

核心思路：先判断任务类型，再走对应节点。

输入 -> 路由判断 -> 写诗节点 / 讲笑话节点 / 兜底节点

常见写法是让模型生成结构化结果，再由条件边决定去向：

def router(state) -> Literal["poem_node", "joke_node", "default_node"]:
    if state["mode"] == "poem":
        return "poem_node"
    if state["mode"] == "joke":
        return "joke_node"
    return "default_node"

builder.add_conditional_edges("router_node", router)

重点：路由判断最好使用 Literal、Pydantic Schema 等结构化输出，不要完全依赖模型生成的自然语言句子。

4. Orchestrator-worker：编排器与工作节点

这个模式由“规划”和“执行”两部分组成：

输入任务 -> Orchestrator 制定计划 -> 动态分发多个 Worker -> 汇总

例如用户要求“写一篇 RAG 教程”，编排器可以先规划：

1. 什么是 RAG
2. 文档加载与切分
3. 向量化与检索
4. 生成回答与评估

接着通过 Send 为每一个章节创建一个 Worker 任务。

from langgraph.types import Send

def router(state):
    return [
        Send("worker_node", {"section": section})
        for section in state["plan"]
    ]

它和普通并行化的区别是：

普通并行：任务数量在写图时就确定
编排器-工作节点：任务数量在运行时由编排器决定

因此它非常适合：自动报告、复杂研究任务、批量处理不定长列表。

5. Evaluator-optimizer：评估器与优化器

核心思路是：生成结果后先评估；不合格就携带反馈重新生成。

Generator -> Evaluator
                 | 合格 -> END
                 | 不合格 + 反馈 -> Generator

例如先生成一个笑话，再让评估器输出结构化结果：

class Feedback(BaseModel):
    grade: Literal["好笑", "不好笑"]
    feedback: str

路由函数根据 grade 决定结束还是重试：

def route_joke(state):
    if state["funny_or_not"] == "好笑":
        return "accept"
    return "reject_and_feedback"

适合：

• 文案生成与质量检查
• SQL 生成与语法验证
• 代码生成与测试
• 翻译生成与审校

这个模式一定要有最大重试次数或递归限制，否则可能无限循环，持续消耗模型 token。

6. Agent：智能体循环

前面五种模式大多由开发者事先确定主要步骤。Agent 的不同在于：模型会根据当前消息和工具结果，动态决定下一步是否调用工具、调用哪个工具。

最基础的 ReAct 结构是：

用户问题 -> LLM 决策 -> 是否需要工具？
                         | 是 -> Tool -> 回到 LLM
                         | 否 -> END

示例骨架：

from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt.tool_node import ToolNode

def model_node(state: MessagesState):
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def router(state: MessagesState):
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tool_node"
    return END

builder = StateGraph(MessagesState)
builder.add_node("model_node", model_node)
builder.add_node("tool_node", ToolNode(tools=tools))
builder.add_edge(START, "model_node")
builder.add_conditional_edges("model_node", router)
builder.add_edge("tool_node", "model_node")

注意：Agent 不是“比工作流更高级”，而是“把更多执行决策交给模型”。

当步骤明确、风险高、需要稳定执行时，固定工作流通常更合适；当工具选择和执行顺序无法提前确定时，才更适合 Agent。

四、如何选择设计模式？

你的任务特点	优先选择
步骤固定，前一步依赖后一步	Prompt Chaining
多个任务互不依赖	Parallelization
不同意图走不同流程	Routing
运行时才知道要创建多少任务	Orchestrator-worker
需要生成、检查、重试	Evaluator-optimizer
工具和执行顺序无法预先确定	Agent

真实项目经常是组合使用：

Routing 判断用户意图
  -> RAG 子图检索资料
  -> Orchestrator-worker 并行生成报告章节
  -> Evaluator-optimizer 检查质量
  -> stream 把生成过程实时发送给前端

五、总结

LangGraph 高级能力的价值，不在于把图画得多复杂，而在于把复杂任务拆得更可控。

最后用几句话记住本篇重点：

stream：图执行时实时拿结果
subgraph：把复杂图拆成可复用模块
Prompt Chaining：固定顺序的步骤链
Parallelization：独立任务同时跑
Routing：根据状态选择路径
Orchestrator-worker：运行时动态分发任务
Evaluator-optimizer：不合格就带反馈重试
Agent：让模型决定是否调用工具和下一步行动

掌握这些模式后，再写 RAG、工具调用、多轮对话或数字人智能体时，就不再只是“把代码接起来”，而是能先设计出合理、可维护、可观察的执行流程。