环企首页
LangGraph 中外部控制修改图状态
LangGraph 提供了多种方式从外部（即图执行之外）修改图的状态，这对于实现“人在回路”（Human-in-the-Loop）、状态恢复、调试等功能至关重要。
📌 核心概念
LangGraph 的状态管理基于持久化检查点机制。图在执行过程中会保存状态快照（checkpoint），外部控制就是通过更新、恢复或修改这些检查点来实现的。
1️⃣ 基本状态访问和修改
获取当前状态
# 执行后获取最终状态
result = graph.invoke(initial_state, config)
# 或者单独获取状态快照
state_snapshot = graph.get_state(config)
print(state_snapshot.values) # 当前状态值
print(state_snapshot.next) # 下一个要执行的节点
更新状态（从外部注入）
使用 graph.update_state() 可以在图执行过程中注入新状态：
# 更新状态
graph.update_state(
config, # 线程配置
{"messages": [HumanMessage(content="人工干预")]}, # 要更新的状态
as_node="human_input" # 模拟从哪个节点更新
)
2️⃣ 人在回路干预（Human-in-the-Loop）
中断执行，等待人工输入
在图编译时设置中断点：
from langgraph.graph import StateGraph, START, END
# 编译时设置中断节点
graph = builder.compile(
checkpointer=checkpointer,
interrupt_before=["assistant"], # 在 assistant 前中断
# 或 interrupt_after=["tool_node"] # 在工具节点后中断
)
外部干预流程
# 第1步：执行图，会在中断点停止
thread_config = {"configurable": {"thread_id": "abc123"}}
for event in graph.stream(
{"messages": [HumanMessage(content="计算3+5")]},
config=thread_config,
stream_mode="values"
):
print(event)
# 第2步：图会停在中断点，你可以检查和修改状态
state = graph.get_state(thread_config)
print("当前状态:", state.values)
print("下一个节点:", state.next) # 应该是 ['assistant']
# 第3步：从外部修改状态
graph.update_state(
thread_config,
{"messages": [HumanMessage(content="修改后的输入")]},
as_node="human_input"
)
# 第4步：继续执行
for event in graph.stream(None, config=thread_config, stream_mode="values"):
print(event)
3️⃣ 使用 Command 进行精细控制
LangGraph 提供了 Command 对象来实现更复杂的控制流：
from langgraph.types import Command
def human_feedback_node(state):
# 等待人工输入
user_input = get_user_input() # 外部获取输入
return Command(
update={"messages": [HumanMessage(content=user_input)]},
goto="assistant" # 指定下一个节点
)
Command 的常用参数
| 参数 | 疫苗作用 |
|---|
4️⃣ 状态版本回滚和时间旅行
利用检查点机制，你可以回到历史的任何状态：
# 获取所有历史检查点
history = list(graph.get_state_history(config))
for i, state in enumerate(history):
print(f"步骤 {i}: {state.values}")
# 回到上一个状态
previous_state = history[1] # index 0 是最新状态
config = previous_state.config
# 从该状态重新执行
for event in graph.stream(None, config=config):
print(event)
5️⃣ 完整示例：人工审批流程
from typing import Annotated, Literal
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
class ApprovalState(TypedDict):
messages: Annotated[list, add_messages]
approved: bool
feedback: str
def assistant(state: ApprovalState):
response = llm.invoke(state["messages"])
return {"messages": [response]}
def human_approval(state: ApprovalState) -> Literal["assistant", "END"]:
# 这个节点实际上会被中断，由外部决策
if state.get("approved", False):
return "assistant"
return END
# 构建图
builder = StateGraph(ApprovalState)
builder.add_node("assistant", assistant)
builder.add_node("human_approval", human_approval)
builder.add_edge(START, "assistant")
builder.add_edge("assistant", "human_approval")
builder.add_conditional_edges("human_approval", human_approval)
# 编译时设置中断
memory = MemorySaver()
graph = builder.compile(checkpointer=memory, interrupt_before=["human_approval"])
# 执行
config = {"configurable": {"thread_id": "approval_001"}}
# 第一次执行，会在 human_approval 前停止
for event in graph.stream(
{"messages": [HumanMessage(content="生成报告")]},
config=config
):
print(event)
# 外部获取状态，询问用户
state = graph.get_state(config)
print("等待审批的内容:", state.values["messages"][-1].content)
# 用户审批
user_approved = input("是否批准？(y/n): ")
# 外部更新状态
graph.update_state(
config,
{"approved": user_approved.lower() == "y"},
as_node="human_approval"
)
# 继续执行
if user_approved.lower() == "y":
for event in graph.stream(None, config=config):
print(event)
📊 方法对比
| 方法 | 疫苗用途 | 疫苗适用场景 |
|---|
⚠️ 重要注意事项
- 需要持久化存储：要支持状态修改和历史查询，必须传入
checkpointer参数编译图 thread_id是必需的：每个会话（线程）需要有唯一的thread_id才能保存状态- 状态合并规则：使用
update_state时，新增的状态会按照 Reducer 规则（如add_messages）合并，而非直接替换 as_node的意义：模拟状态是从哪个节点发起的更新，影响后续图的执行逻辑stream(None)恢复执行：中断后继续执行时，调用graph.stream(None, config)即可
💎 总结
外部控制修改图状态的核心方法是：
get_state(config)：获取当前状态update_state(config, update, as_node)：注入新状态interrupt_before/after：设置中断点- 状态历史回滚：回到任意历史状态
这为构建需要人工审批、错误恢复、动态修正的复杂 Agent 系统提供了强大的支持。