流式输出与中断机制
Streaming & Interruption in LangGraph

1 为什么需要流式输出：从"等待结果"到"实时观察"

默认情况下，调用 graph.invoke() 是阻塞式的：你等图完整跑完，才能拿到最终 State。对于简单任务这没问题，但对于：

• LLM 生成长文本：用户需要等待数秒，体验极差
• 多节点复杂流程：你完全不知道图当前跑到哪个节点
• Agent 循环调用工具：中间步骤完全黑箱
• Human-in-the-loop 场景：需要在特定节点"暂停"等人工干预

LangGraph 提供了一流的流式支持（First-class Streaming），让你可以在图执行过程中实时获取各种粒度的输出。本节的终极目标是理解 Streaming 是如何为 Module 3 的核心特性——Human-in-the-loop——铺路的。

2 图的结构回顾：本节使用的 Chatbot 示例

本节 notebook 复用了 Module 2 的带摘要的 Chatbot 图。先理解这张图的结构，才能看懂后面的流式输出。

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)

3 stream_mode="updates"：只看每个节点的增量变化

stream_mode="updates" 是最常用的流式模式。图每执行完一个节点，就会推送一个 chunk，chunk 里只包含这个节点对 State 做出的变化部分，而不是完整 State。

基本用法

# 创建一个对话线程（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)

chunk 的结构是一个字典，键是节点名称，值是该节点返回的更新内容。比如上面的 conversation 节点返回了 messages 字段的更新，里面包含 LLM 的回复。

提取并格式化 LLM 回复

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

updates 模式的数据流图示

4 stream_mode="values"：每步输出图的完整状态

stream_mode="values" 与 "updates" 的区别在于：每次推送的不是某个节点的变更，而是当前整张图的完整 State。

代码示例

# 使用新的 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)  # 分隔符，区分不同节点执行后的状态

注意：第一次推送发生在图接收到输入之后、第一个节点执行之前（状态里只有 Human Message）；第二次推送发生在 conversation 节点执行完之后（状态里有 Human + AI Message）。

updates vs values 的关键区别

5 astream_events：Token 级别的细粒度流式输出

前面两种模式（updates / values）都是节点粒度的流式——必须等节点执行完才推送。但聊天应用往往需要 LLM 生成每个 token 时就立刻显示给用户，这就需要 astream_events。

astream_events 的工作原理

graph.astream_events() 是一个异步方法（需要 async for），它在节点内部执行时同步触发各类事件，你可以根据事件类型过滤出你关心的信息。

第一步：观察所有事件类型

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']}")

每个 event 字典包含以下关键字段：

第二步：过滤出特定节点的 Token 流

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

完整的 Token 流式打印代码

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="|")

6 LangGraph API 的 messages 模式：专为聊天设计

当你通过 LangGraph API（LangSmith Studio 或部署后的服务） 远程调用图时，除了 values 和 updates 之外，还有一种仅在 API 层面支持的特殊模式：stream_mode="messages"。

通过 API 使用 values 模式

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)

API 返回的 event 对象结构：event.event（类型字符串）和 event.data（数据内容）。

messages 模式：细粒度事件流

stream_mode="messages" 专门为消息型 LangGraph（state 里有 messages 字段）设计，会推送四类事件：

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)  # 查看事件类型

完整的 messages 模式处理代码

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

7 中断机制：interrupt_before 与 interrupt_after

流式输出让你观察图的执行；而中断机制（Interruption）则让你控制图的执行——在特定节点前后暂停图，等待外部（通常是人）做出决策，再恢复执行。这正是 Human-in-the-loop 的核心机制。

在 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_before 和 interrupt_after 都接受一个节点名列表，可以同时声明多个中断点：

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

interrupt_before vs interrupt_after 的区别

中断后的图状态

图被中断后，你可以查看当前 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.next 返回一个元组，表示图暂停时，下一步待执行的节点名称。如果图已经执行完毕（正常结束），snapshot.next 为空元组 ()。

8 Human-in-the-loop 完整模式：暂停、审批、恢复

掌握了中断机制后，我们来看完整的 Human-in-the-loop 工作流：图暂停 → 人工检查/修改 State → 恢复执行。

第一步：恢复执行（不修改 State）

如果人工审查后认为 LLM 的决策正确，直接让图继续：

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

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

第二步：修改 State 后再恢复（人工干预）

这是 Human-in-the-loop 的精髓：人工修改图的 State，然后恢复执行：

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

完整的 Human-in-the-loop 生命周期

典型应用场景

本节知识点总结