编辑图状态与注入人工反馈
update_state · as_node · Human Feedback

1 回顾：人机协作的三种模式

在 Module 3 的前几课中，我们讨论了为什么需要在 AI Agent 流程中引入人的参与。LangGraph 的 human-in-the-loop 支持三种核心场景：

前一课学习的断点（Breakpoint）让我们能够暂停图的执行，但暂停之后只能选择"继续"或"终止"。而本课要解决的问题是：暂停之后，如何主动修改 State，然后带着修改后的状态继续执行？

本课涉及的关键 API 是 graph.update_state(config, values)，它是 LangGraph 提供的核心状态编辑接口，配合断点使用可以实现强大的人机协作交互。

2 graph.update_state() — 直接编辑图状态

搭建测试用的 Agent 图

Notebook 首先构建了一个带有断点的计算助手 Agent，使用 interrupt_before=["assistant"] 在助手节点之前暂停：

from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState, START, StateGraph
from langgraph.prebuilt import tools_condition, ToolNode
from langchain_core.messages import HumanMessage, SystemMessage

# 定义三个算术工具
def multiply(a: int, b: int) -> int:
    """Multiply a and b."""
    return a * b

def add(a: int, b: int) -> int:
    """Adds a and b."""
    return a + b

def divide(a: int, b: float) -> float:
    """Divide a by b."""
    return a / b

tools = [add, multiply, divide]
llm = ChatDeepSeek(model="deepseek-v4-pro")
llm_with_tools = llm.bind_tools(tools)

# 系统消息
sys_msg = SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")

# assistant 节点：调用绑定了工具的 LLM
def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

# 构建图
builder = StateGraph(MessagesState)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", tools_condition)
builder.add_edge("tools", "assistant")

memory = MemorySaver()
# 关键：在 assistant 节点之前设置断点
graph = builder.compile(interrupt_before=["assistant"], checkpointer=memory)

触发断点：图在 assistant 节点前暂停

# 发送初始输入，图会在断点处停止
initial_input = {"messages": "Multiply 2 and 3"}
thread = {"configurable": {"thread_id": "1"}}

for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()
# 输出显示图已收到消息，正在等待 assistant 节点前的断点

此时图已暂停。我们可以检查当前状态：

state = graph.get_state(thread)
print(state.next)  # ('assistant',) — 下一步等待执行 assistant

核心 API：graph.update_state()

现在来看本课最关键的调用——update_state。它的完整签名如下：

graph.update_state(
    config,          # thread 配置，指向哪个 checkpoint（哪条线程）
    values,          # 要更新的 State 字段的字典
    as_node=None    # 可选：伪装成哪个节点执行了此更新（后面详述）
)

在 Notebook 中，我们向消息列表追加了一条新的人工消息，修正了计算需求：

graph.update_state(
    thread,
    {"messages": [HumanMessage(content="No, actually multiply 3 and 3!")]},
)

update_state 返回一个新的 checkpoint 配置，说明状态已经被保存到一个新的检查点中。

3 add_messages Reducer 与消息覆盖机制

update_state 不是简单地"替换"整个 State，它遵循 State 中为每个字段定义的 Reducer 规则。理解这一点至关重要。

MessagesState 中的 Reducer

MessagesState 是 LangGraph 内置的状态类，其 messages 字段使用 add_messages 作为 Reducer：

# MessagesState 的内部定义（LangGraph 源码简化版）
from typing import Annotated
from langgraph.graph.message import add_messages

class MessagesState(TypedDict):
    # Annotated 声明 reducer：add_messages 函数处理合并逻辑
    messages: Annotated[list, add_messages]

两种消息更新行为的对比

add_messages Reducer 的工作方式取决于你传入的消息是否携带 id：

场景一：追加新消息（无 id）

Notebook 中的第一个 update_state 调用传入了一个没有 id 的新 HumanMessage，所以 add_messages 将它追加到了现有列表的末尾：

验证结果：

new_state = graph.get_state(thread).values
for m in new_state['messages']:
    m.pretty_print()

场景二：覆盖现有消息（带 id）

如果我们想替换而不是追加，只需要在传入的消息对象中保留原消息的 id。这样 add_messages 会找到具有该 id 的消息并覆盖它：

# 假设已有消息的 id 是 "msg-123"
# 传入同 id 的新消息 → 覆盖，而非追加
graph.update_state(
    thread,
    {"messages": [HumanMessage(
        content="No, actually multiply 3 and 3!",
        id="msg-123"   # 与原消息 id 相同 → 覆盖该消息
    )]},
)

4 编辑状态后恢复图执行

修改状态只是第一步，完整的流程还需要让图从修改后的状态继续执行。恢复执行的方法非常简单：传入 None 作为输入再次调用 graph.stream()。

完整的编辑-恢复流程代码

# ① 发送初始输入，图在 assistant 节点前暂停
initial_input = {"messages": "Multiply 2 and 3"}
thread = {"configurable": {"thread_id": "1"}}

for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()
# 此时图暂停，等待人工干预

# ② 编辑状态：注入一条新的用户消息
graph.update_state(
    thread,
    {"messages": [HumanMessage(content="No, actually multiply 3 and 3!")]},
)

# ③ 传入 None，从当前状态继续执行
for event in graph.stream(None, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

由于图仍然设置了 interrupt_before=["assistant"]，它在工具结果返回后再次在 assistant 节点前暂停。传入 None 再次恢复：

# ④ 图再次在断点处停下，再次传 None 让它完成最后一步
for event in graph.stream(None, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

5 as_node 参数：伪装成特定节点的更新

update_state 的第三个参数 as_node 是一个高级特性，也是实现"人工反馈节点"的关键。

as_node 解决了什么问题？

图在执行过程中会记录"上一个执行的节点是谁"，因为边（Edge）的跳转逻辑依赖于此信息。当你直接调用 update_state 时，LangGraph 需要知道：这次更新算作"哪个节点"完成了执行？

通过 as_node 参数，你可以指定这次状态更新"假装"是某个节点完成的，从而让图按照正确的边的逻辑来决定下一步去哪里。

# 指定 as_node="human_feedback"
# 意思是：假装 human_feedback 节点刚刚执行完，并产生了这些状态更新
graph.update_state(
    thread,
    {"messages": user_input},
    as_node="human_feedback"   # 关键参数
)

as_node 的工作原理图示

as_node 与图的 next 字段

调用 update_state(as_node="human_feedback") 后，图的 state.next 会变成 ('assistant',)，因为从 human_feedback 节点之后，边直接连接到了 assistant：

# 调用 update_state(as_node="human_feedback") 之后
state = graph.get_state(thread)
print(state.next)   # ('assistant',) — 下一步执行 assistant

6 添加 human_feedback 节点实现人工输入

前面的例子是在外部"悄悄修改"了图的消息。更优雅的设计是：在图结构中显式地加入一个代表人类反馈的节点，让人机协作成为图架构的一部分。

设计思路

新的图引入了一个 human_feedback 节点，它本身是一个空函数（no-op），不做任何实际处理。它的作用只是作为图中一个"占位符"——当图执行到这里时，通过 interrupt_before 暂停，等待人工调用 update_state(as_node="human_feedback") 来注入反馈内容，然后继续执行。

# no-op 节点：本身不做任何事，只作为占位符
def human_feedback(state: MessagesState):
    pass   # 什么都不做，真正的内容由 update_state 注入

def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

# 构建新图
builder = StateGraph(MessagesState)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_node("human_feedback", human_feedback)

# 关键：图从 human_feedback 开始，工具完成后也回到 human_feedback
builder.add_edge(START, "human_feedback")
builder.add_edge("human_feedback", "assistant")
builder.add_conditional_edges("assistant", tools_condition)
builder.add_edge("tools", "human_feedback")  # 工具结束后回到人工反馈节点

memory = MemorySaver()
# 在 human_feedback 节点之前设置断点
graph = builder.compile(interrupt_before=["human_feedback"], checkpointer=memory)

完整的人工反馈工作流代码

# ① 启动图，图执行到 human_feedback 前暂停
initial_input = {"messages": "Multiply 2 and 3"}
thread = {"configurable": {"thread_id": "5"}}

for event in graph.stream(initial_input, thread, stream_mode="values"):
    event["messages"][-1].pretty_print()

# ② 获取真实的人工输入（这里用 input() 模拟）
user_input = input("Tell me how you want to update the state: ")
# 实际应用中，这个 user_input 来自 Web UI、API 等

# ③ 注入人工反馈，伪装成 human_feedback 节点的输出
graph.update_state(thread, {"messages": user_input}, as_node="human_feedback")

# ④ 继续执行图（从 assistant 节点开始）
for event in graph.stream(None, thread, stream_mode="values"):
    event["messages"][-1].pretty_print()

当用户输入 "no, multiply 3 and 3" 时，输出如下：

7 通过 LangSmith Studio API 远程编辑状态

除了在 Python 代码中直接调用 graph.update_state()，LangGraph 还提供了基于 HTTP 的 API（通过 langgraph_sdk 客户端），可以远程编辑运行在服务器上的图的状态。这是 LangSmith Studio 可视化界面背后的底层机制。

使用 LangGraph SDK 客户端

from langgraph_sdk import get_client

# 连接到本地运行的 LangGraph 开发服务器
client = get_client(url="http://127.0.0.1:2024")
# 生产环境中换成你的 LangGraph Cloud URL

通过 API 触发断点并获取状态

initial_input = {"messages": "Multiply 2 and 3"}

# 创建一个新线程
thread = await client.threads.create()

# 流式运行，并传入 interrupt_before（无需在代码里预定义断点）
async for chunk in client.runs.stream(
    thread["thread_id"],
    "agent",
    input=initial_input,
    stream_mode="values",
    interrupt_before=["assistant"],   # API 调用时动态传入断点
):
    messages = chunk.data.get('messages', [])
    if messages:
        print(messages[-1])

通过 API 获取和修改状态

# 获取当前状态
current_state = await client.threads.get_state(thread['thread_id'])

# 取出最后一条消息
last_message = current_state['values']['messages'][-1]
print(last_message)
# {'content': 'Multiply 2 and 3', 'id': '882dabe4-...', 'type': 'human', ...}

# 修改消息内容（保留 id → 触发覆盖）
last_message['content'] = "No, actually multiply 3 and 3!"

# 将修改后的消息写回（同 id → 覆盖原消息）
await client.threads.update_state(
    thread['thread_id'],
    {"messages": last_message}
)

注意 SDK 中的 API 是 client.threads.update_state()，对应本地的 graph.update_state()，功能完全相同，只是通过 HTTP 远程调用。

通过 API 恢复执行

# 传入 input=None，从当前最新状态继续执行
async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="agent",
    input=None,              # None = 从当前 checkpoint 继续
    stream_mode="values",
    interrupt_before=["assistant"],
):
    messages = chunk.data.get('messages', [])
    if messages:
        print(messages[-1])

8 完整工作流总结与最佳实践

完整的人工编辑工作流

关键概念速查

何时选择哪种状态编辑方式

本课知识点全景