一个 Agent 跑完一次可能要几十秒：检索、调用工具、反复思考。对用户来说，这几十秒里屏幕上什么都没有，体验和"程序卡死了"没有区别。另一方面，有些步骤不能完全交给模型——转账、发邮件、删除数据，执行之前必须有人看一眼。这一篇处理这两件和"人"有关的事：流式输出让用户看见过程，**中断（interrupt）**让人在关键位置做决定。后者完全建立在上一篇的持久化之上，建议先读完上一篇。

一、流式输出：边执行边汇报#

invoke 的工作方式是"全部跑完，一次性交出结果"。stream 则在执行过程中不断地吐出数据，调用方可以用 for 循环一条条接收。第一篇已经用过一次：

Python
for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)

stream_mode 决定"汇报什么内容"。下面用一个"检索 → 生成"的两步小图，对比几种最常用的模式。

1.1 关心数据：values 与 updates#

  • values：每个超步结束后，交出完整的 State。三次输出分别是：只有问题 → 问题加检索结果 → 全部字段。适合需要随时展示完整状态的界面。
  • updates：每个节点完成后，只交出它返回的那部分，并标明是哪个节点：{'retrieve': {'docs': [...]}}。数据量小，最适合做"步骤进度条"。

1.2 关心模型输出：messages#

聊天界面最想要的是"打字机效果"：模型生成一个字，就显示一个字。messages 模式就是做这个的，它交出的是模型生成的一个个片段，以及片段来自哪个节点：

Python
for chunk, metadata in graph.stream(inputs, stream_mode="messages"):
    print(chunk.content, end="")              # 逐段打印模型输出
    # metadata["langgraph_node"] 是产生这段输出的节点名，可以只显示某个节点的内容

值得一提的是，节点里调用模型时用的是普通的 model.invoke(...)，并不需要改成 model.stream(...)——实测 LangGraph 会自动捕获模型的逐段输出，照样一段段地流出来。业务代码不用为了流式输出做任何改动。

1.3 关心自定义进度：custom#

有些进度信息既不是 State，也不是模型输出，比如"正在检索第 2 个数据源"。这时可以在节点里拿到一个"写入器"，主动往流里推送任何内容：

Python
from langgraph.config import get_stream_writer

 ():
    writer = get_stream_writer()
     i, source  ([, , ], ):
        writer({: })
        ...
     {: docs}