Streaming Output and Interruption Mechanism
Streaming & Interruption in LangGraph

1Why streaming output is needed: from "waiting for results" to "real-time observation"

By default, callinggraph.invoke()yesblocking: You can’t get the final State until the graph is completely run. This is fine for simple tasks, but for:

• LLM generates long text: The user needs to wait for several seconds and the experience is extremely poor.
• Multi-node complex process: You have absolutely no idea which node the graph is currently running to.
• Agent cycle calling tool: The intermediate steps are completely black box
• Human-in-the-loop scene: Requires manual intervention such as "pause" at a specific node

LangGraph providesFirst-class streaming support (First-class Streaming), allowing you to obtain output of various granularities in real time during graph execution. The ultimate goal of this section is to understand how Streaming paves the way for the core feature of Module 3: Human-in-the-loop.

2Graph structure review: Chatbot example used in this section

This notebook reuses the Chatbot diagram with summary from Module 2. You must first understand the structure of this picture before you can understand the subsequent streaming output.

from langchain_deepseek import ChatDeepSeek
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph import MessagesState

# ─── State：继承 MessagesState，额外加 summary 字段 ───
class State(MessagesState):
    summary: str

# ─── Node 1：调用 LLM 进行对话 ───
def call_model(state: State, config: RunnableConfig):
    summary = state.get("summary", "")
    if summary:
        system_message = f"Summary of conversation earlier: {summary}"
        messages = [SystemMessage(content=system_message)] + state["messages"]
    else:
        messages = state["messages"]
    response = model.invoke(messages, config)  # config 支持 token 流式
    return {"messages": response}

# ─── Node 2：对历史对话做摘要（消息过多时触发）───
def summarize_conversation(state: State):
    summary = state.get("summary", "")
    if summary:
        summary_message = f"This is summary of the conversation to date: {summary}\n\n" \
                          "Extend the summary by taking into account the new messages above:"
    else:
        summary_message = "Create a summary of the conversation above:"
    messages = state["messages"] + [HumanMessage(content=summary_message)]
    response = model.invoke(messages)
    # 保留最近 2 条消息，其余删除
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    return {"summary": response.content, "messages": delete_messages}

# ─── 路由函数：消息超过 6 条时触发摘要 ───
def should_continue(state: State):
    if len(state["messages"]) > 6:
        return "summarize_conversation"
    return END

# ─── 构建图 ───
workflow = StateGraph(State)
workflow.add_node("conversation", call_model)
workflow.add_node(summarize_conversation)

workflow.add_edge(START, "conversation")
workflow.add_conditional_edges("conversation", should_continue)
workflow.add_edge("summarize_conversation", END)

# 使用 MemorySaver 作为 checkpointer，支持线程级别的对话记忆
memory = MemorySaver()
graph = workflow.compile(checkpointer=memory)

3stream_mode="updates": only look at the incremental changes of each node

stream_mode="updates"It is the most commonly used streaming mode. Every time a node in the graph is executed, achunk, the chunk only contains the changes made by this node to StateChange part, rather than the full State.

Basic usage

# 创建一个对话线程（thread_id 用于区分不同对话）
config = {"configurable": {"thread_id": "1"}}

# 使用 stream() 方法，指定 stream_mode="updates"
for chunk in graph.stream(
    {"messages": [HumanMessage(content="hi! I'm Lance")]},
    config,
    stream_mode="updates"
):
    print(chunk)

The structure of chunk is a dictionary,The key is the node name，The value is the updated content returned by the node. For example, the aboveconversationThe node returnedmessagesUpdate of fields to include LLM's response.

Extract and format LLM responses

for chunk in graph.stream(
    {"messages": [HumanMessage(content="hi! I'm Lance")]},
    config,
    stream_mode="updates"
):
    # 从 conversation 节点的更新中取出消息，美化打印
    chunk['conversation']["messages"].pretty_print()

Data flow diagram of updates mode

4stream_mode="values": Output the complete status of the graph at each step

stream_mode="values"with"updates"The difference is: each push is notChanges to a certain node, butThe complete State of the current entire graph。

code example

# 使用新的 thread_id 开启新对话
config = {"configurable": {"thread_id": "2"}}

input_message = HumanMessage(content="hi! I'm Lance")
for event in graph.stream(
    {"messages": [input_message]},
    config,
    stream_mode="values"
):
    # event 就是完整的 State 字典
    for m in event['messages']:
        m.pretty_print()
    print("---"*25)  # 分隔符，区分不同节点执行后的状态

Note:first pushOccurs after the graph receives input and before the first node is executed (there is only Human Message in the state);Second pushOccurs after the conversation node is executed (there is Human + AI Message in the state).

Key differences between updates vs values

5astream_events: Token-level fine-grained streaming output

The first two modes (updates/values) areNode granularityStreaming - you must wait for the node to finish executing before pushing. However, chat applications often require LLM to immediately display each token to the user when it generates it, which requiresastream_events。

How astream_events works

graph.astream_events()is aasynchronous method(Requiredasync for), which triggers various events synchronously when executed inside the node. You can filter out the information you care about based on the event type.

Step 1: Observe all event types

config = {"configurable": {"thread_id": "3"}}
input_message = HumanMessage(content="Tell me about the 49ers NFL team")

# version="v2" 是目前推荐的版本，事件格式更稳定
async for event in graph.astream_events(
    {"messages": [input_message]}, config, version="v2"
):
    # 每个 event 有 metadata.langgraph_node（哪个节点触发）
    # event.event（事件类型）
    # event.name（具体名称）
    print(f"Node: {event['metadata'].get('langgraph_node','')}. "
          f"Type: {event['event']}. Name: {event['name']}")

Each event dictionary contains the following key fields:

Step 2: Filter out the Token stream of specific nodes

node_to_stream = 'conversation'  # 指定只监听哪个节点的输出
config = {"configurable": {"thread_id": "4"}}
input_message = HumanMessage(content="Tell me about the 49ers NFL team")

async for event in graph.astream_events(
    {"messages": [input_message]}, config, version="v2"
):
    # 只处理聊天模型产生的 token 事件，且来自 conversation 节点
    if (event["event"] == "on_chat_model_stream"
            and event['metadata'].get('langgraph_node', '') == node_to_stream):
        data = event["data"]
        # data["chunk"] 是一个 AIMessageChunk，.content 是这个 token 的文本
        print(data["chunk"].content, end="|")  # 用 | 分隔每个 token

Complete Token streaming printing code

config = {"configurable": {"thread_id": "5"}}
input_message = HumanMessage(content="Tell me about the 49ers NFL team")

async for event in graph.astream_events(
    {"messages": [input_message]}, config, version="v2"
):
    if (event["event"] == "on_chat_model_stream"
            and event['metadata'].get('langgraph_node', '') == 'conversation'):
        data = event["data"]
        # end="" 让 token 连续显示，不换行
        print(data["chunk"].content, end="|")

6LangGraph API’s messages mode: designed for chatting

when you passLangGraph API (LangSmith Studio or deployed service)When calling a graph remotely, exceptvaluesandupdatesIn addition, there is a special mode that is only supported at the API level:stream_mode="messages"。

Using the values pattern via the API

from langgraph_sdk import get_client
from langchain_core.messages import convert_to_messages

client = get_client(url="http://127.0.0.1:2024")

# 创建新线程
thread = await client.threads.create()
input_message = HumanMessage(content="Multiply 2 and 3")

async for event in client.runs.stream(
    thread["thread_id"],
    assistant_id="agent",
    input={"messages": [input_message]},
    stream_mode="values"     # 远程 API 也支持 values 模式
):
    messages = event.data.get('messages', None)
    if messages:
        print(convert_to_messages(messages)[-1])  # 打印最新消息
    print('='*25)

The event object structure returned by the API:event.event(type string) andevent.data(data content).

messages mode: fine-grained event streaming

stream_mode="messages"specially forMessage LangGraph(Contained in statemessagesField) design will push four types of events:

thread = await client.threads.create()
input_message = HumanMessage(content="Multiply 2 and 3")

async for event in client.runs.stream(
    thread["thread_id"],
    assistant_id="agent",
    input={"messages": [input_message]},
    stream_mode="messages"
):
    print(event.event)  # 查看事件类型

Complete messages mode processing code

def format_tool_calls(tool_calls):
    """格式化工具调用列表为可读字符串"""
    if tool_calls:
        formatted_calls = []
        for call in tool_calls:
            formatted_calls.append(
                f"Tool Call ID: {call['id']}, Function: {call['name']}, Arguments: {call['args']}"
            )
        return "\n".join(formatted_calls)
    return "No tool calls"

thread = await client.threads.create()
input_message = HumanMessage(content="Multiply 2 and 3")

async for event in client.runs.stream(
    thread["thread_id"],
    assistant_id="agent",
    input={"messages": [input_message]},
    stream_mode="messages"
):
    # 处理运行元信息
    if event.event == "metadata":
        print(f"Metadata: Run ID - {event.data['run_id']}")
        print("-" * 50)

    # 处理流式消息片段
    elif event.event == "messages/partial":
        for data_item in event.data:
            # 用户输入消息
            if "role" in data_item and data_item["role"] == "user":
                print(f"Human: {data_item['content']}")
            else:
                content = data_item.get("content", "")
                tool_calls = data_item.get("tool_calls", [])
                if content:
                    print(f"AI: {content}")
                if tool_calls:
                    print("Tool Calls:")
                    print(format_tool_calls(tool_calls))

7Interrupt mechanism: interrupt_before and interrupt_after

Streaming output lets youobserveThe execution of the graph; and the interruption mechanism (Interruption) allows you tocontrolGraph execution - before and after a specific nodepause chart, wait for external decisions (usually humans) to be made, and then resume execution. This is exactlyHuman-in-the-loopcore mechanism.

Declare breakpoints during compile()

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

# interrupt_before：在执行指定节点"之前"暂停
graph = workflow.compile(
    checkpointer=memory,
    interrupt_before=["tools"]    # 在执行 tools 节点之前暂停
)

# interrupt_after：在执行指定节点"之后"暂停
graph = workflow.compile(
    checkpointer=memory,
    interrupt_after=["conversation"]  # 在 conversation 节点执行完后暂停
)

interrupt_beforeandinterrupt_afterAccept oneNode name list, multiple break points can be declared at the same time:

# 同时声明多个中断点
graph = workflow.compile(
    checkpointer=memory,
    interrupt_before=["tools", "human_review"],  # 在这两个节点前都暂停
    interrupt_after=["planner"]                   # 在 planner 执行后暂停
)

The difference between interrupt_before vs interrupt_after

Graph state after interruption

After the graph is interrupted, you can view the current State:

config = {"configurable": {"thread_id": "1"}}

# invoke 会在中断点停下，不会继续执行后续节点
result = graph.invoke(
    {"messages": [HumanMessage(content="search for the latest AI news")]},
    config
)

# 获取当前的图状态快照
snapshot = graph.get_state(config)

# 查看下一步将执行哪个节点（中断在哪里）
print(snapshot.next)      # 例如：('tools',) 表示即将执行 tools 节点

# 查看当前 State
print(snapshot.values)    # 完整的 State 字典

snapshot.nextReturns a tuple representingWhen the graph is paused, the name of the node to be executed next. If the graph has been executed (normally ended),snapshot.nextis an empty tuple()。

8Human-in-the-loop full mode: pause, approve, resume

After mastering the interruption mechanism, let's look at the complete Human-in-the-loop workflow:Graph paused → Manual inspection/modification of State → Resume execution。

Step 1: Resume execution (without modifying State)

If LLM’s decision is deemed correct after manual review, let the diagram continue directly:

# 以 None 作为输入调用 invoke，表示"继续从中断处运行，不改变 State"
# 关键：必须使用同一个 config（同一个 thread_id），图才能找到存档点
result = graph.invoke(None, config)

# 图从 tools 节点继续执行，完成后返回最终 State
print(result)

Step 2: Modify the State and then restore it (manual intervention)

This is the essence of Human-in-the-loop: manually modify the State of the graph and then resume execution:

# 1. 先查看当前 State
snapshot = graph.get_state(config)
current_messages = snapshot.values["messages"]
print("当前最后一条消息（LLM 的工具调用请求）：")
print(current_messages[-1])

# 2. 人工修改 State（例如：修改工具调用的参数）
# 假设 LLM 要搜索 "latest AI news"，但人工觉得应该搜索更具体的内容
from langchain_core.messages import AIMessage

# 构造修改后的消息
modified_message = AIMessage(
    content="",
    id=current_messages[-1].id,           # 复用原消息的 id，触发覆盖而非追加
    tool_calls=[{
        "id": current_messages[-1].tool_calls[0]["id"],
        "name": "tavily_search",
        "args": {"query": "GPT-5 release date 2024"}  # 人工修改了搜索词
    }]
)

# 3. 更新图的 State（使用 update_state）
graph.update_state(
    config,
    {"messages": [modified_message]},   # id 相同 → add_messages reducer 原地覆盖该条消息
    as_node="agent"                       # 以 agent 节点的视角更新
)

# 4. 恢复执行（图会使用修改后的 State 继续运行）
result = graph.invoke(None, config)

Complete Human-in-the-loop life cycle

Typical application scenarios

Summary of knowledge points in this section