第二阶段：LangGraph 高级控制流与状态管理

2.1 人工介入模式（Human-in-the-Loop）

为什么需要人工介入？

在真实业务中，AI 不应该"自作主张"做所有事。比如：

• 发送邮件前，需要人工确认内容
• 执行支付操作前，需要人工审批
• AI 生成的代码，需要人工 review 后才能合并

LangGraph 提供了三种人工介入机制。

2.1.1 interrupt_before —— 暂停等待批准

核心思想： 在某个节点执行之前暂停整个图，等你确认后再继续。

# ———- 前置条件：Checkpointer ———-
# Checkpointer 是人工介入和时间旅行的前提
# – MemorySaver：内存存储（演示用，重启丢失）
# – 生产环境可用 SqliteSaver / PostgresSaver 持久化
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

# ———- 构建图并设置中断点 ———-
app = builder.compile(
checkpointer=checkpointer,
interrupt_before=["approval"] # ← 在 approval 节点执行前暂停
)

运行：图在 approval 前自动暂停

thread = {"configurable": {"thread_id": "quote_real"}}

# 启动图，会在 approval 节点前暂停
print("=== LLM 正在生成报价草案… ===")
for event in app.stream({"requirement": requirement}, thread):
for node_name, node_output in event.items():
if node_name == "generate":
print(f"\\n{node_output['draft']}")

# interrupt_before 生效 —— 此时 approval 节点尚未执行
state = app.get_state(thread)
print(f"下一步节点: {state.next}") # 输出 ('approval',)，证明已暂停

执行时序图：

START → [generate] → ⏸️ 暂停！(interrupt_before=["approval"])
│
│ 人工审批（y/n）
▼
[approval] → [parallel] → …

2.1.2 update_state —— 动态修改状态后继续

核心思想： 图暂停后，你不仅可以"继续"，还可以修改状态再继续。这就像在流水线上把工件拿下来改一改再放回去。

# ———- 人工审批（update_state） ———-
print(f"\\n草案内容：\\n{state.values['draft']}")
user_input = input("是否批准？(y/n): ").strip().lower()
approved = user_input == 'y'

if approved:
new_draft = input("请输入修改意见（直接回车则保持不变）：").strip()
if new_draft:
# ★ 修改草案并批准
# as_node="approval"：告诉图"这次更新等同于 approval 节点的输出"
# 可以同时修改多个字段（draft + approved）
app.update_state(thread, {"draft": new_draft, "approved": True}, as_node="approval")
print(f"草案已修改为：{new_draft}")
else:
# 保持原草案，仅批准
app.update_state(thread, {"approved": True}, as_node="approval")
else:
print("草案未批准，流程结束。")
app.update_state(thread, {"approved": False}, as_node="approval")
exit()

update_state 的参数：

条件分支：根据 approved 决定走向

# approved=True → 进入 parallel 节点继续询价
# approved=False → 直接结束
builder.add_conditional_edges(
"approval",
lambda state: "parallel" if state.approved else END
)

2.1.3 Command —— “编辑后重放”

核心思想： Command 是一种更优雅的方式，在恢复执行时同时提供新输入和流程控制。

本示例使用 update_state 等价演示。Command 的写法如下：

from langgraph.types import Command

# 用 Command 恢复执行，同时修改状态
result = app.invoke(
Command(
update={ # 要更新的状态
"draft": "修改后的报价草案",
"approved": True
},
resume="approved" # 恢复执行的信号值（可选）
),
config=thread
)

Command vs update_state + invoke(None)：

2.2 时间旅行与分支回溯

为什么需要时间旅行？

• 调试： 对话出 bug 了，想回到第 3 步看看当时的状态
• 复现： 从某个历史状态重新开始，走不同的路
• 对比： 从同一起点分叉，尝试不同的策略

Checkpointer 保存了每一步的状态快照，这就是时间旅行的基础。

2.2.1 查看历史状态

sys_demo.py 中的实现：

# ———- 时间旅行：回退到审批节点前重新决策 ———-

# ★ 列出所有历史检查点
checkpoints = list(app.get_state_history(thread))
print(f"历史检查点数量：{len(checkpoints)}")
for i, cp in enumerate(checkpoints):
print(f" [{i}] 下一步节点: {cp.next}")

输出示例：

历史检查点数量：8
[0] 下一步节点: ('generate',) ← START
[1] 下一步节点: ('approval',) ← generate 完成，等待审批
[2] 下一步节点: ('approval',) ← update_state 注入审批
[3] 下一步节点: ('parallel',) ← approval 完成
…
[-1] 下一步节点: () ← END（最终状态）

2.2.2 回溯到历史状态（replay）

核心思想： 从某个历史检查点重新执行，就像存档读档。

2.2.3 创建新分支（fork）

核心思想： 从某个历史状态分叉出一条新的对话，不影响原来的对话。就像 Git 的分支一样。

# 回退到 generate 之后、approval 之前的检查点（索引 1）
# 这相当于"fork"出一条新分支，从审批前重新开始
if len(checkpoints) >= 2:
rollback_checkpoint = checkpoints[1].config # generate 完成后的检查点
print("回退到检查点 [1]，修改 approved=False（模拟拒绝）…")

# ★ 在历史检查点上修改状态 → 创建新分支
app.update_state(rollback_checkpoint, {"approved": False}, as_node="approval")

# ★ 从回退点继续执行（走新分支）
print("从回退点继续执行：")
for event in app.stream(None, config=rollback_checkpoint):
print(f" {event}")

# 验证新分支的最终状态
new_state = app.get_state(thread)
print(f"时间旅行后的最终状态 – draft: {new_state.values.get('draft', 'N/A')}")
print(f"approved: {new_state.values.get('approved', 'N/A')}")
print("✅ 时间旅行完成！原始分支的结果不受影响。")

fork 的思维模型：

时间轴:
[START] → [generate] → [approval(批准)] → [parallel] → [fetch×3] → [aggregate] → [END]
│
└── fork! → [approval(拒绝)] → [END]
（新分支，独立发展）

2.3 Send API 实现并行分叉

为什么需要并行？

• 同时查 3 个供应商的报价，串行太慢
• 并行审核 3 份文档，互不依赖
• MapReduce 模式：分拆任务 → 并行处理 → 汇总

Send 就是 LangGraph 的"并行分发器"。

状态定义

from pydantic import BaseModel, Field
from typing import Annotated
from operator import add
from langgraph.graph.message import add_messages

class QuoteState(BaseModel):
messages: Annotated[list, add_messages]
requirement: str = "" # 用户原始采购需求
draft: str = "" # LLM 生成的报价草案
approved: bool = False # 审批标志，控制条件分支走向
# ★ Annotated[list, add]：reducer 模式，并发写入自动合并
supplier_results: Annotated[list, add] = Field(default_factory=list)
final_quote: str = ""
suppliers_list: list = Field(default_factory=list)

model_config = {"arbitrary_types_allowed": True}

分发节点：准备并行数据

def parallel_fetch(state: QuoteState):
"""LLM 分析草案，决定需要询价的供应商列表"""
response = llm.invoke([
SystemMessage(content="你是一位采购专家。根据以下报价草案，列出3个适合的供应商名称（中文，用逗号分隔，只输出名称，不要其他内容）。"),
HumanMessage(content=state.draft)
])
# 解析供应商列表
raw = response.content.strip().replace("，", ",").replace(" ", "")
suppliers = [s.strip() for s in raw.split(",") if s.strip()]
# 兜底：如果 LLM 输出格式不规范，补齐到 3 个
while len(suppliers) < 3:
suppliers.append(f"备选供应商{len(suppliers)+1}")
suppliers = suppliers[:3]
# ★ 节点函数必须返回 dict，不能直接返回 Send 列表
return {"suppliers_list": suppliers}

路由函数：生成 Send 列表

def route_to_suppliers(state: QuoteState):
"""路由函数：返回 Send 列表，触发并行分叉"""
suppliers = state.suppliers_list if hasattr(state, 'suppliers_list') else state.get('suppliers_list', [])
# ★ 每个 Send 携带独立子状态，并发执行 fetch_supplier 节点
return [Send("fetch_supplier", {"supplier": s, "base_draft": state.draft}) for s in suppliers]

Send 的参数：

子节点：并发执行

def fetch_supplier(state: Dict[str, Any]):
"""LLM 模拟单个供应商根据草案给出报价（并发执行）"""
supplier = state["supplier"]
base_draft = state["base_draft"]
response = llm.invoke([
SystemMessage(content="你是一家供应商的销售代表。请根据客户的采购草案，给出你们公司的报价。格式：供应商名称 报价：XXX 元，交货期：XX天，备注：XXX。只输出一条报价，不要多余解释。"),
HumanMessage(content=f"你是【{supplier}】。客户草案如下：\\n{base_draft}")
])
result = response.content.strip()
# ★ supplier_results 使用 Annotated[list, add] reducer，并发写入会自动合并
return {"supplier_results": [result]}

汇总节点

def aggregate(state: QuoteState):
"""汇总节点：所有并行任务完成后，LLM 分析选出最优方案"""
all_quotes = "\\n".join([f" {i+1}. {q}" for i, q in enumerate(state.supplier_results)])
response = llm.invoke([
SystemMessage(content="你是一位采购决策专家。请分析以下多家供应商的报价，选出最优的一家，并给出推荐理由。格式：\\n最优选择：XXX\\n理由：XXX\\n最终报价：XXX元。用中文回答。"),
HumanMessage(content=f"采购草案：{state.draft}\\n\\n各供应商报价：\\n{all_quotes}")
])
final = response.content.strip()
return {
"final_quote": final,
"messages": [AIMessage(content=final)]
}

构建图：绑定 Send 路由

builder = StateGraph(QuoteState)
builder.add_node("generate", generate_draft)
builder.add_node("approval", human_approval)
builder.add_node("parallel", parallel_fetch)
builder.add_node("fetch_supplier", fetch_supplier)
builder.add_node("aggregate", aggregate)

builder.add_edge(START, "generate")
builder.add_edge("generate", "approval")
builder.add_conditional_edges("approval", lambda state: "parallel" if state.approved else END)

# ★ 关键：条件边绑定路由函数，路由函数返回 Send 列表触发并行
builder.add_conditional_edges("parallel", route_to_suppliers)
builder.add_edge("fetch_supplier", "aggregate")
builder.add_edge("aggregate", END)

观察并行输出

print("=== Send 并行分叉 – 并发查询供应商 ===")
for event in app.stream(None, thread):
for node_name, node_output in event.items():
if node_name == "parallel":
print(f"询价供应商：{node_output.get('suppliers_list', [])}")
elif node_name == "fetch_supplier":
for r in node_output.get("supplier_results", []):
print(f" -> {r}")

输出示例：

询价供应商：['中鼎物流', '备选供应商2', '备选供应商3']
-> 供应商名称 中鼎物流 报价：9580元 交货期：20天 备注：无
-> 供应商名称 备选供应商3 报价：9580元 交货期：20天 备注：无
-> 备选供应商2 报价：9580元，交货期：10天，备注：无特殊要求。

Send 的关键理解

小结

完整流程图

[用户输入采购需求]
│
▼
┌─────────────┐
│ generate │ LLM 生成报价草案
└─────────────┘
│
▼
⏸️ interrupt_before=["approval"] 暂停！
│
▼
┌─────────────┐
│ approval │ 人工审批（update_state 注入决策）
└─────────────┘
│
├── approved=True → 继续
└── approved=False → END
│
▼
┌─────────────┐
│ parallel │ LLM 推荐 3 家供应商
└─────────────┘
│ Send × 3（并行分发）
├──→ [fetch_supplier(供应商A)]
├──→ [fetch_supplier(供应商B)]
└──→ [fetch_supplier(供应商C)]
│（全部完成后）
▼
┌─────────────┐
│ aggregate │ LLM 分析选出最优方案
└─────────────┘
│
▼
时间旅行（可选）
回退到历史检查点，fork 新分支

代码示例demo：

from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langgraph.checkpoint.memory import MemorySaver
from pydantic import BaseModel, Field
from typing import Annotated, Dict, Any
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langchain_ollama import ChatOllama

# ———- LLM 初始化 ———-
llm = ChatOllama(
model="qwen2.5:1.5b",
temperature=0,
base_url="http://localhost:11434"
)

# ==================================================================
# Checkpointer 是人工介入和时间旅行的前提
# – MemorySaver：内存存储（演示用，重启丢失）
# – 生产环境可用 SqliteSaver / PostgresSaver 持久化
# ==================================================================

# ———- 状态定义 ———-
class QuoteState(BaseModel):
messages: Annotated[list, add_messages]
requirement: str = "" # 用户原始采购需求
draft: str = "" # LLM 生成的报价草案
approved: bool = False # 审批标志，控制条件分支走向
supplier_results: Annotated[list, add] = Field(default_factory=list)
final_quote: str = ""
suppliers_list: list = Field(default_factory=list)

model_config = {"arbitrary_types_allowed": True}

# ==================================================================
# Send API 并行分叉 – 节点定义
# ==================================================================

# ———- 节点函数 ———-

def generate_draft(state: QuoteState):
"""LLM 根据采购需求生成报价草案"""
requirement = state.requirement
response = llm.invoke([
SystemMessage(content="你是一位专业的采购报价分析师。请根据用户的采购需求，生成一份简洁的报价草案，包含：产品名称、建议单价、建议数量、预估总价、是否含税。格式清晰，用中文回答。"),
HumanMessage(content=f"采购需求：{requirement}")
])
draft = response.content
return {
"messages": [AIMessage(content=f"草案已生成：{draft}")],
"draft": draft
}

# ==================================================================
# 人工介入 – 占位节点
# interrupt_before=["approval"] 会使图在执行到此节点前暂停
# 节点本身不需要逻辑，真正的审批逻辑在外部通过 update_state 注入
# ==================================================================
def human_approval(state: QuoteState):
"""占位节点：interrupt_before 在此暂停，等待人工审批"""
return {}

# ==================================================================
# Send API 并行分叉 – 分发节点
# 该节点将供应商列表存入状态，由路由函数读取并生成 Send 任务
# ⚠️ 注意：节点函数必须返回 dict，不能直接返回 Send 列表
# ==================================================================
def parallel_fetch(state: QuoteState):
"""LLM 分析草案，决定需要询价的供应商列表"""
response = llm.invoke([
SystemMessage(content="你是一位采购专家。根据以下报价草案，列出3个适合的供应商名称（中文，用逗号分隔，只输出名称，不要其他内容）。"),
HumanMessage(content=state.draft)
])
# 解析供应商列表
raw = response.content.strip().replace("，", ",").replace(" ", "")
suppliers = [s.strip() for s in raw.split(",") if s.strip()]
# 兜底：如果 LLM 输出格式不规范，补齐到 3 个
while len(suppliers) < 3:
suppliers.append(f"备选供应商{len(suppliers)+1}")
suppliers = suppliers[:3]
return {"suppliers_list": suppliers}

# ==================================================================
# Send API 并行分叉 – 路由函数
# 路由函数从状态中读取供应商列表，为每个供应商创建一个 Send 对象
# 每个 Send 会触发 fetch_supplier 节点的独立并发执行
# ==================================================================
def route_to_suppliers(state: QuoteState):
"""路由函数：返回 Send 列表，触发并行分叉"""
suppliers = state.suppliers_list if hasattr(state, 'suppliers_list') else state.get('suppliers_list', [])
# 每个 Send 携带独立子状态，并发执行 fetch_supplier 节点
return [Send("fetch_supplier", {"supplier": s, "base_draft": state.draft}) for s in suppliers]

def fetch_supplier(state: Dict[str, Any]):
"""LLM 模拟单个供应商根据草案给出报价（并发执行）"""
supplier = state["supplier"]
base_draft = state["base_draft"]
response = llm.invoke([
SystemMessage(content="你是一家供应商的销售代表。请根据客户的采购草案，给出你们公司的报价。格式：供应商名称 报价：XXX 元，交货期：XX天，备注：XXX。只输出一条报价，不要多余解释。"),
HumanMessage(content=f"你是【{supplier}】。客户草案如下：\\n{base_draft}")
])
result = response.content.strip()
# supplier_results 使用 Annotated[list, add] reducer，并发写入会自动合并
return {"supplier_results": [result]}

def aggregate(state: QuoteState):
"""汇总节点：所有并行任务完成后，LLM 分析选出最优方案"""
all_quotes = "\\n".join([f" {i+1}. {q}" for i, q in enumerate(state.supplier_results)])
response = llm.invoke([
SystemMessage(content="你是一位采购决策专家。请分析以下多家供应商的报价，选出最优的一家，并给出推荐理由。格式：\\n最优选择：XXX\\n理由：XXX\\n最终报价：XXX元。用中文回答。"),
HumanMessage(content=f"采购草案：{state.draft}\\n\\n各供应商报价：\\n{all_quotes}")
])
final = response.content.strip()
return {
"final_quote": final,
"messages": [AIMessage(content=final)]
}

# ———- 构建图 ———-
builder = StateGraph(QuoteState)
builder.add_node("generate", generate_draft)
builder.add_node("approval", human_approval)
builder.add_node("parallel", parallel_fetch)
builder.add_node("fetch_supplier", fetch_supplier)
builder.add_node("aggregate", aggregate)

builder.add_edge(START, "generate")
builder.add_edge("generate", "approval")

# ==================================================================
# 条件分支：根据 approved 状态决定走向
# – approved=True → 进入 parallel 节点继续询价
# – approved=False → 直接结束
# ==================================================================
builder.add_conditional_edges("approval", lambda state: "parallel" if state.approved else END)

# ==================================================================
# Send 并行分叉：条件边绑定路由函数
# parallel 节点执行完毕后，由 route_to_suppliers 返回 Send 列表
# 每个 Send 并发触发 fetch_supplier，全部完成后汇入 aggregate
# ==================================================================
builder.add_conditional_edges("parallel", route_to_suppliers)
builder.add_edge("fetch_supplier", "aggregate")
builder.add_edge("aggregate", END)

# ==================================================================
# Checkpointer 是核心基础设施
# – interrupt_before=["approval"]：在 approval 节点前暂停
# – MemorySaver()：保存每一步状态快照，支持后续回溯
# ==================================================================
checkpointer = MemorySaver()
app = builder.compile(checkpointer=checkpointer, interrupt_before=["approval"])

# ==================================================================
# 运行流程：演示全部三个知识点
# ==================================================================
thread = {"configurable": {"thread_id": "quote_real"}}

# ——————– 第一步：用户输入 + LLM 生成草案 ——————–
print("=" * 50)
print(" 智能采购报价系统")
print("=" * 50)
requirement = input("\\n请输入您的采购需求：").strip()
if not requirement:
requirement = "采购办公用笔记本电脑 10 台，要求 i7 处理器、16G 内存"

print("\\n=== LLM 正在生成报价草案… ===")
for event in app.stream({"requirement": requirement}, thread):
for node_name, node_output in event.items():
if node_name == "generate":
print(f"\\n{node_output['draft']}")

# ==================================================================
# interrupt_before 生效 – 图在此暂停
# 此时 approval 节点尚未执行，可以通过 get_state 查看当前状态
# ==================================================================
state = app.get_state(thread)
print(f"\\n下一步节点: {state.next}") # 输出 ('approval',)，证明已暂停

# ——————– 第二步：人工审批（update_state） ——————–
# ==================================================================
# update_state 动态修改状态后继续执行
# – as_node="approval"：告诉图"这次更新等同于 approval 节点的输出"
# – 可以同时修改多个字段（draft + approved）
# – 类似 Command 的"编辑后重放"效果
# ==================================================================
print("\\n" + "=" * 50)
print(" 【知识点 2.1】人工审批 – update_state")
print("=" * 50)
print(f"\\n草案内容：\\n{state.values['draft']}")
user_input = input("\\n是否批准？(y/n): ").strip().lower()
approved = user_input == 'y'

if approved:
new_draft = input("请输入修改意见（直接回车则保持不变）：").strip()
if new_draft:
# 修改草案并批准
app.update_state(thread, {"draft": new_draft, "approved": True}, as_node="approval")
print(f"\\n草案已修改为：{new_draft}")
else:
# 保持原草案，仅批准
app.update_state(thread, {"approved": True}, as_node="approval")
else:
print("\\n草案未批准，流程结束。")
app.update_state(thread, {"approved": False}, as_node="approval")
exit()

# ——————– 第三步：Send 并行分叉执行 ——————–
# ==================================================================
# Send API 并行分叉 – 观察输出
# – parallel 节点：LLM 推荐 3 家供应商
# – route_to_suppliers：生成 3 个 Send，并发执行 fetch_supplier
# – aggregate：所有 Send 完成后自动汇总
# ==================================================================
print("\\n=== Send 并行分叉 – 并发查询供应商 ===")
for event in app.stream(None, thread):
for node_name, node_output in event.items():
if node_name == "parallel":
print(f"\\n询价供应商：{node_output.get('suppliers_list', [])}")
elif node_name == "fetch_supplier":
for r in node_output.get("supplier_results", []):
print(f" -> {r}")

# ——————– 第四步：查看最终结果 ——————–
final_state = app.get_state(thread)
print("\\n" + "=" * 50)
print(" 最终报价结果")
print("=" * 50)
print(f"\\n{final_state.values['final_quote']}")

# ——————– 第五步：时间旅行与分支回溯 ——————–
# ==================================================================
# 时间旅行（Time Travel）
# – get_state_history：获取所有历史检查点（快照）
# – 每个检查点记录了图在某个节点的完整状态
# – 可以回退到任意历史点，修改状态后重新执行（相当于 fork 新分支）
#
# 历史检查点顺序（从旧到新）：
# checkpoints[0] → START
# checkpoints[1] → generate 执行完毕（草案生成后）
# checkpoints[-2] → 倒数第二个（通常是 aggregate 之前）
# checkpoints[-1] → 最新状态（END）
# ==================================================================
print("\\n" + "=" * 50)
print(" 时间旅行 – 回退到审批节点前重新决策")
print("=" * 50)

# 列出所有历史检查点
checkpoints = list(app.get_state_history(thread))
print(f"\\n历史检查点数量：{len(checkpoints)}")
for i, cp in enumerate(checkpoints):
print(f" [{i}] 下一步节点: {cp.next}")

# 回退到 generate 之后、approval 之前的检查点（索引 1）
# 这相当于"fork"出一条新分支，从审批前重新开始
if len(checkpoints) >= 2:
rollback_checkpoint = checkpoints[1].config # generate 完成后的检查点
print(f"\\n回退到检查点 [1]，修改 approved=False（模拟拒绝）…")

# 在历史检查点上修改状态 → 创建新分支
app.update_state(rollback_checkpoint, {"approved": False}, as_node="approval")

# 从回退点继续执行（走新分支）
print("从回退点继续执行：")
for event in app.stream(None, config=rollback_checkpoint):
print(f" {event}")

# 验证新分支的最终状态
new_state = app.get_state(thread)
print(f"\\n时间旅行后的最终状态 – draft: {new_state.values.get('draft', 'N/A')}")
print(f"approved: {new_state.values.get('approved', 'N/A')}")
print("\\n✅ 时间旅行完成！原始分支的结果不受影响。")
else:
print("检查点不足，无法执行时间旅行。")