Edit graph state and inject human feedback
update_state · as_node · Human Feedback

1Review: Three modes of human-machine collaboration

In the previous lessons of Module 3, we discussed why we need to include humans in the AI Agent process. LangGraph's human-in-the-loop supports three core scenarios:

The breakpoints learned in the previous lesson allow us toPause graph execution, but after pausing, you can only choose "continue" or "terminate". The problem to be solved in this lesson is:After pausing, how to actively modify the State and then continue execution with the modified state?

The key APIs covered in this lesson aregraph.update_state(config, values), which is the core state editing interface provided by LangGraph. Used with breakpoints, it can achieve powerful human-computer collaborative interaction.

2graph.update_state() — directly edit graph state

Build an Agent graph for testing

Notebook first built a computing assistant Agent with breakpoints, usinginterrupt_before=["assistant"]Pause before helper node:

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)

Trigger breakpoint: the graph pauses before the assistant node

# 发送初始输入，图会在断点处停止
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 节点前的断点

The graph is paused at this time. We can check the current status:

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

Core API: graph.update_state()

Now let’s look at the most critical call in this lesson——update_state. Its full signature is as follows:

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

In the Notebook, we appended a new manual message to the message list, correcting the computational requirements:

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

update_stateReturns a new checkpoint configuration indicating that the state has been saved to a new checkpoint.

3add_messages reducer and message coverage mechanism

update_stateRather than simply "replacing" the entire State, it follows thereducer rules. It is crucial to understand this.

reducer in MessagesState

MessagesStateis LangGraph's built-in state class, whichmessagesField usageadd_messagesAs a 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]

Comparison of two message update behaviors

add_messagesThe way the reducer works depends on whether your incoming message carriesid：

Scenario 1: Append new message (without id)

First in Notebookupdate_stateThe call passes in aNew HumanMessage without id, soadd_messagesput itAppendReaching the end of the existing list:

Verification result:

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

Scenario 2: Overwriting existing message (with id)

if we wantreplaceInstead of appending, just keep the original message in the incoming message object.id. This wayadd_messageswill find the message with that id and overwrite it:

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

4Resume graph execution after editing state

Modifying the state is only the first step. The complete process also requires the diagram to continue execution from the modified state. The method to resume execution is very simple:incomingNoneas inputcall againgraph.stream()。

Complete edit - restore process code

# ① 发送初始输入，图在 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()

Since the graph is still setinterrupt_before=["assistant"], which pauses again in front of the assistant node after the tool results return. incomingNoneRestore again:

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

5as_node parameter: disguised as an update of a specific node

update_stateThe third parameter ofas_nodeIt is an advanced feature and the key to realizing "artificial feedback node".

What problem does as_node solve?

During the execution of the graph, "who was the last executed node" is recorded, because the jump logic of the edge depends on this information. When you call directlyupdate_stateAt this time, LangGraph needs to know: "Which node" counts as completing the execution of this update?

passas_nodeParameters, you can specify that this status update "pretends" to be completed by a certain node, so that the graph determines where to go next according to the correct edge logic.

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

Illustration of how as_node works

as_node and the next field of the graph

callupdate_state(as_node="human_feedback")After, figure ofstate.nextwill become('assistant',), because fromhuman_feedbackAfter the node, the edge is directly connected toassistant：

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

6Add human_feedback node to implement manual input

The previous example was a message that the graph was "quietly modified" externally. A more elegant design is: in the graph structureExplicitly add a node representing human feedback, making human-machine collaboration part of the graph architecture.

design ideas

The new diagram introduces ahuman_feedbackNode, itself is an empty function (no-op) and does not do any actual processing. Its role is just as a "placeholder" in the diagram - when the diagram is executed here, passinterrupt_beforePause, waiting for manual callupdate_state(as_node="human_feedback")to inject feedback content and then continue execution.

# 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)

Complete manual feedback workflow code

# ① 启动图，图执行到 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()

When the user enters "no, multiply 3 and 3", the output is as follows:

7Edit status remotely via LangSmith Studio API

In addition to calling it directly in Python codegraph.update_state(), LangGraph also provides an HTTP-based API (vialanggraph_sdkclient), you can remotely edit the state of the graph running on the server. This is the underlying mechanism behind LangSmith Studio's visual interface.

Using the LangGraph SDK client

from langgraph_sdk import get_client

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

Trigger breakpoints and get status via 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])

Get and modify status via 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}
)

Note that the API in the SDK isclient.threads.update_state(), corresponding to the localgraph.update_state(), the function is exactly the same, but it is called remotely through HTTP.

Resume execution via 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])

8Complete workflow summary and best practices

Complete human editing workflow

Quick Check on Key Concepts

When to choose which status editing method

Panorama of knowledge points in this course