BytesRouter

主题

LangChain 文档版本 跟随上游 源快照 2f3fa160 更新于 2026-10-08

事件流式传输

使用针对消息、状态、子图、输出和扩展的类型化投影来流式输出 LangGraph 运行。

事件流式传输是大多数 LangGraph 应用代码推荐的进程内流式传输模型。它返回一个运行流对象，可以同时以多种方式消费。

快速开始

py
stream = graph.stream_events({
    "messages": [{"role": "user", "content": "What is 42 * 17?"}],
}, version="v3")

for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

final_state = stream.output

要针对部署在 Agent Server 后面的图进行流式传输，请参阅 LangSmith 流式传输 API。

各部分如何组合在一起

流式传输栈有两个主要层次：

  1. 流式传输（Streaming）从 Pregel 引擎发出原始的图执行事件。
  2. 事件流式传输（Event streaming）对这些事件进行规范化，将它们送经流转换器，并暴露类型化投影。
Pregel 引擎
运行图步骤
<div class="text-center text-xs font-medium uppercase tracking-wide text-slate-500 dark:text-slate-400">发出</div>

<div class="rounded-lg border border-orange-300 bg-orange-50 p-3 text-center dark:border-orange-800 dark:bg-orange-950">
  <div class="font-semibold text-orange-900 dark:text-orange-100">原始 Pregel 事件</div>
  <div class="mt-1 text-xs text-orange-700 dark:text-orange-300"><code>updates</code>, <code>values</code>, <code>messages</code>, <code>custom</code>, <code>checkpoints</code>, <code>tasks</code>, <code>debug</code></div>
</div>

<div class="text-center text-xs font-medium uppercase tracking-wide text-slate-500 dark:text-slate-400">发送至</div>

<div class="rounded-lg border border-blue-300 bg-blue-50 p-3 text-center dark:border-blue-800 dark:bg-blue-950">
  <div class="font-semibold text-blue-900 dark:text-blue-100">事件路由器</div>
  <div class="mt-1 text-xs text-blue-700 dark:text-blue-300">让每个事件经过转换器流水线</div>
</div>

<div class="text-center text-xs font-medium uppercase tracking-wide text-slate-500 dark:text-slate-400">级联经过</div>

<div class="rounded-lg border border-green-300 bg-green-50 p-3 dark:border-green-800 dark:bg-green-950">
  <div class="font-semibold text-green-900 dark:text-green-100">流转换器</div>

  <div class="mt-2 grid gap-2 text-xs text-green-800 dark:text-green-200 sm:grid-cols-4">
    <div class="rounded border border-green-200 bg-white px-2 py-1 dark:border-green-800 dark:bg-green-950">ValuesTransformer</div>
    <div class="rounded border border-green-200 bg-white px-2 py-1 dark:border-green-800 dark:bg-green-950">MessagesTransformer</div>
    <div class="px-2 py-1 text-center text-green-600 dark:text-green-300">...</div>
    <div class="rounded border border-green-200 bg-white px-2 py-1 dark:border-green-800 dark:bg-green-950">自定义转换器</div>
  </div>
</div>

<div class="text-center text-xs font-medium uppercase tracking-wide text-slate-500 dark:text-slate-400">产出</div>

<div class="rounded-lg border border-purple-300 bg-purple-50 p-3 text-center dark:border-purple-800 dark:bg-purple-950">
  <div class="font-semibold text-purple-900 dark:text-purple-100">事件流</div>
  <div class="mt-1 text-xs text-purple-700 dark:text-purple-300">为应用代码投影的事件</div>
</div>

事件路由是两个层之间的桥梁。它接收规范化后的 Pregel 事件，并将每个事件依次传递给已注册的流转换器。内置转换器会创建标准投影，例如 stream.messages、stream.values、stream.subgraphs 和 stream.output。自定义转换器可以在 stream.extensions 下添加应用专属的投影。

事件流式传输提供什么

运行流在一个底层事件流之上暴露了类型化的投影：

投影用途
stream迭代每个协议事件。
stream.messages流式输出聊天模型消息和 token 增量。
stream.values迭代状态快照并等待最终值。
stream.output等待最终输出。
stream.subgraphs发现并观察嵌套图的执行。
stream.interrupts检查人在回路的中断载荷。
stream.interrupted检查运行是否因等待人工输入而暂停。
stream.extensions消费自定义流转换器的投影。

多个消费者可以并发读取这些投影。读取 stream.messages 不会消耗 stream.values、stream.subgraphs 或 stream.output 所需的事件。

事件流式传输位于流式传输之上，后者通过 stream_mode 模式（例如 updates、values、messages、custom、checkpoints、tasks 和 debug）暴露原始的图执行事件。当你需要对这些模式进行底层访问时，请使用流式传输；当应用代码受益于类型化投影时，请使用事件流式传输。

流式输出消息

使用 stream.messages 获取聊天模型的输出：

py
stream = graph.stream_events(input, version="v3")

for message in stream.messages:
    text = str(message.text)
    usage = message.output.usage_metadata

    print(text)
    print(usage)

在同步代码中，message.text 是可迭代的。迭代它以获得逐 token 的输出，或调用 str(message.text) 获取完整文本。

message.reasoning 暴露推理增量，message.tool_calls 暴露工具调用参数分块。如果你需要按精确的到达顺序获取文本、推理和工具调用分块，请迭代消息流的原始事件，而不是分别迭代每个投影。

流式输出子图

使用 stream.subgraphs 观察嵌套图的工作，而无需解析命名空间字符串：

py
stream = graph.stream_events(input, version="v3")

for subgraph in stream.subgraphs:
    print(subgraph.graph_name, subgraph.path)

    for message in subgraph.messages:
        print(message.text)

subgraph.graph_name 是编译后的图或智能体的 name。从工具派发的具名智能体（例如通过 Deep Agents 的 task 工具调用的 create_agent(name=...)）会以该名称在此处出现，而开启该作用域的 lifecycle 事件会携带一个 cause，链接回派发它的工具调用。更多信息请参见生命周期。

关于产品专属的流，请参阅Deep Agents 流式传输了解子智能体流，以及 LangChain 智能体流式传输了解工具调用和中间件事件。

流式输出状态

使用 stream.values 在每个步骤之后流式输出完整的状态快照：

py
stream = graph.stream_events(input, version="v3")

for snapshot in stream.values:
    print(snapshot)

final_state = stream.output

流式输出多个投影

在异步代码中并发消费时，请把 astream_events 与 asyncio.gather 配合使用：

py
import asyncio

stream = await graph.astream_events(input, version="v3")

async def consume_messages():
    async for message in stream.messages:
        print(f"[llm] node={message.node}")

async def consume_subgraphs():
    async for subgraph in stream.subgraphs:
        print(f"[subgraph] path={subgraph.path}")

await asyncio.gather(consume_messages(), consume_subgraphs())

对于同步代码，使用 stream.interleave(...) 按严格的到达顺序消费多个投影：

py
stream = graph.stream_events(input, version="v3")

for name, item in stream.interleave("values", "messages", "subgraphs"):
    if name == "values":
        print(f"[state] keys={list(item)}")
    elif name == "messages":
        print(f"[llm] node={item.node}")
    elif name == "subgraphs":
        print(f"[subgraph] path={item.path}")

中断后恢复

当图因等待人工输入而暂停时，检查 stream.interrupted 和 stream.interrupts，然后再次调用 stream_events(..., version="v3") 并传入 Command 来恢复。

恢复需要图使用检查点器编译，并且配置中携带线程 ID —— 参见持久化。

py
from langgraph.types import Command

stream = graph.stream_events(input, version="v3")

for message in stream.messages:
    print(message.text)

if stream.interrupted:
    print(stream.interrupts)

stream = graph.stream_events(
    Command(resume={"decisions": [{"type": "approve"}]}),
    version="v3",
)
final_state = stream.output

流式输出所有协议事件

当你需要原始协议事件流时，直接使用运行对象本身：

py
stream = graph.stream_events({
    "messages": [{"role": "user", "content": "What is 42 * 17?"}],
}, version="v3")

for event in stream:
    namespace = event["params"]["namespace"]
    print(namespace, event["method"], event["params"]["data"])

每个事件都是一个 ProtocolEvent 信封，包裹着特定于通道的载荷。转换器的 process(event) 接收的也是同样的结构。

py
class ProtocolEvent(TypedDict):
    seq: int                    # strictly increasing within a run; use for ordering
    method: str                 # channel name: "messages", "values", "updates", "custom", "tools", "lifecycle", ...
    params: ProtocolEventParams

class ProtocolEventParams(TypedDict):
    namespace: list[str]        # path of "<name>:<runtime_id>" segments from the root graph; [] is the root
    timestamp: int              # wall-clock milliseconds; can drift, don't rely on for ordering
    data: Any                   # channel-specific payload; shape depends on `method`

namespace 是从根图到发出该事件的作用域的路径。根是空数组 []。每个子执行都会添加一个 "name:runtime_id" 段，因此子图内嵌套的工具调用看起来像 ["researcher:6f4d", "tools:91ac"]。: 之前的名称是稳定的图或节点名称；后缀是每次调用的运行时 ID。当你只关心特定子树时，请自行按命名空间过滤原始事件 —— stream.subgraphs 已经为嵌套图执行做了这件事。

通道与事件生命周期

原始事件在通道上流动。通道名称会作为事件的 method 出现；每个通道发出特定的事件结构。

通道用途
values完整图状态快照。
updates每个节点的状态增量。
messages以内容块为中心的聊天模型输出。
tools工具调用的开始、流式输出、完成和错误事件。
lifecycle运行、子图和子智能体的状态变化。
checkpoints用于分支和时间旅行的轻量级检查点信封。
input人在回路的输入请求和响应。
tasksPregel 任务创建和结果事件。
custom图代码中用户定义的载荷。
custom:<name>应用定义的流转换器输出。

类型化投影（stream.messages、stream.values 等）由这些通道构建。当你直接迭代运行对象时，通道名称会作为原始事件上的 method 字段出现。

消息

messages 通道将输出建模为内容块。数据的 event 字段是以下之一：

  • message-start
  • content-block-start
  • content-block-delta
  • content-block-finish
  • message-finish

内容块具有明确的边界：一个块开始、发出零个或多个增量，并在同一条消息中的下一个块开始之前结束。这使得 token 流式传输、推理块、工具调用块和多模态内容变得明确，而无需依赖特定服务商的格式。message-finish 可能包含 token 用量；不可恢复的模型调用失败会以消息错误事件的形式到达。

若要直接消费原始的内容块事件，而不是使用 stream.messages 投影：

py
for event in stream:
    if event["method"] != "messages":
        continue

    data = event["params"]["data"][0]
    if not isinstance(data, dict):
        continue
    if data.get("event") != "content-block-delta":
        continue

    block = data.get("delta") or {}
    if block.get("type") == "text-delta":
        print(block.get("text", ""), end="", flush=True)
    elif block.get("type") == "reasoning-delta":
        print(f"[thinking]{block.get('reasoning', '')}", end="", flush=True)

工具

tools 通道暴露工具执行。数据的 event 字段是以下之一：

  • tool-started
  • tool-output-delta
  • tool-finished
  • tool-error

工具事件通过工具调用 ID 进行关联，因此一次工具执行可以与其在 messages 通道上发起它的工具调用内容块关联起来。

生命周期

lifecycle 通道跟踪根运行、子图和子智能体的状态。数据的 event 字段是以下之一：

  • started
  • running
  • completed
  • failed
  • interrupted

除 event 之外，生命周期数据还可能包含可选的 graph_name、error 和 cause，描述子作用域为何启动（父工具调用、扇出发送、边转移）。

构建你自己的投影

流转换器是事件流式传输中的投影层。它们观察协议事件，维护自己的状态，并暴露一次运行的派生视图 —— 例如工具活动、token 总数、进度事件、产物，或供另一种协议使用的消息。StreamChannel 是转换器用来发布这些视图的投影原语。

内置投影（stream.messages、stream.values、stream.subgraphs、stream.output）和产品专属投影（LangChain 的 stream.tool_calls、Deep Agents 的 stream.subagents）本身就是使用同一契约的转换器。用户转换器通过编译时或调用时注册叠加在其上，它们的投影会出现在 stream.extensions 下。

当现有投影不符合应用所需的结构时，就编写一个。

转换器如何工作

事件流式传输始于 LangGraph Pregel 引擎的流式输出。运行时会将这些分块规范化为协议事件，然后由流处理器将每个事件路由通过一叠流转换器。

flowchart TD
    A[Pregel modes] --> B[Events]
    B --> C[Built-in projections]
    C --> D[User transformers]
    D --> E[Run projections]

流处理器是单个流的中央调度器。对于每个协议事件，它会：

  1. 按顺序调用每个已注册转换器的 process(event) 钩子。
  2. 将具名 StreamChannel 的推送接回协议事件流。
  3. 将事件存储在运行流中，除非某个转换器将其抑制。
  4. 当运行结束时，对每个转换器调用 finalize() 或 fail()。

转换器是观察式的。它们不会回调图运行时。相反，它们消费事件并将派生值推入 StreamChannel、promise 或其他投影对象。

转换器的形态

转换器实现 StreamTransformer 接口：

py
from langgraph.stream import ProtocolEvent, StreamTransformer

class MyTransformer(StreamTransformer):
    def init(self) -> dict:
        ...

    def process(self, event: ProtocolEvent) -> bool:
        ...

    def finalize(self) -> None:
        ...

    def fail(self, err: BaseException) -> None:
        ...
  • init() 创建投影对象。用户转换器的投影会出现在 stream.extensions 下。
  • process() 观察每个协议事件。有关 ProtocolEvent 的结构，请参见流式输出所有协议事件。仅当你有意要抑制原始事件时才返回 false。
  • finalize() 在流成功结束后关闭或解析非通道投影。
  • fail() 将错误传播到非通道投影。

声明所需的流模式

required_stream_modes 控制底层图在流式传输期间发出哪些 Pregel 流模式。运行时会取每个已注册转换器的 required_stream_modes 的并集，并将该并集作为 stream_mode 参数传递给图的 .stream() 调用。没有任何转换器请求的模式永远不会被发出 —— 声明 ("custom",) 才使得 custom 事件能够流经该运行。

py
class CustomTransformer(StreamTransformer):
    required_stream_modes = ("custom",)  

    def process(self, event: ProtocolEvent) -> bool:
        if event["method"] == "custom":
            ...
        return True

process() 接收图发出的每个事件，并负责按 event["method"] 进行过滤。该声明会开启上游的发出；它不会缩小 process() 看到的内容。有效值是 Pregel 流模式："messages"、"tools"、"custom"、"values"、"updates"、"checkpoints"、"tasks"、"debug"。每个转换器都必须声明它作用到的每一种模式 —— 被省略的模式不会被图发出，也永远不会到达 process()。

StreamChannel

StreamChannel 是转换器用于流式输出值的投影原语。它始终在 stream.extensions.<name> 上暴露一个可迭代的流。构造函数参数决定每次 push() 是否也会作为 custom:<name> 事件流入运行的主事件流 —— 也就是说，投影的值是否会在迭代原始协议事件时出现。

需求用法
仅侧通道投影StreamChannel()
同时将每次推送流入主事件流StreamChannel(name)

具名通道的载荷必须可序列化，因为每个推送的值也会成为主事件流中的一个 custom:<name> 协议事件。请将 promise、异步可迭代对象、类实例以及其他进程内句柄保留在未命名通道中。

流处理器负责通道的生命周期。一旦 init() 返回一个通道，当运行结束时，处理器会为你关闭或使其失败。转换器只负责推送值。

示例：具名通道

向 StreamChannel 传入字符串名称，以通过 stream.extensions 暴露一个流式投影，并将每个推送的值作为 custom:<name> 协议事件转发到运行的主事件流中：

py
from typing import TypedDict

from langgraph.stream import ProtocolEvent, StreamChannel, StreamTransformer

class ToolActivity(TypedDict):
    name: str
    status: str

class ToolActivityTransformer(StreamTransformer):
    required_stream_modes = ("tools",)

    def __init__(self, scope: tuple[str, ...] = ()) -> None:
        super().__init__(scope)
        self.activity = StreamChannel[ToolActivity]("tool_activity")

    def init(self) -> dict:
        return {"tool_activity": self.activity}

    def process(self, event: ProtocolEvent) -> bool:
        if event["method"] != "tools":
            return True

        data = event["params"]["data"]
        if isinstance(data, dict) and data.get("tool_name") and data.get("event"):
            status = "error" if data["event"] == "tool-error" else "started"
            self.activity.push({"name": data["tool_name"], "status": status})
        return True

示例：未命名通道

不指定名称时，该通道仅是一个侧通道投影 —— 可通过 stream.extensions 访问，但对迭代原始事件的消费者不可见。对于持有无法序列化到主事件流上的进程内句柄（promise、异步可迭代对象、类实例）的投影来说，这是正确的选择。

下面的示例将一个未命名通道与 get_stream_writer 配对使用，后者让图节点能够发出 custom 通道事件，然后转换器将这些事件排入投影：

py
from langgraph.config import get_stream_writer
from langgraph.stream import ProtocolEvent, StreamChannel, StreamTransformer

def node(state):
    writer = get_stream_writer()
    writer({"kind": "progress", "message": "retrieving context"})
    return state

class CustomTransformer(StreamTransformer):
    required_stream_modes = ("custom",)

    def __init__(self, scope: tuple[str, ...] = ()) -> None:
        super().__init__(scope)
        self.log = StreamChannel()

    def init(self) -> dict:
        return {"custom": self.log}

    def process(self, event: ProtocolEvent) -> bool:
        if event["method"] == "custom":
            self.log.push(event["params"]["data"])
        return True

stream = graph.stream_events(input, version="v3", transformers=[CustomTransformer])

for item in stream.extensions["custom"]:
    print(item)

示例：最终值投影

当投影不应流入主事件流时，请使用未命名流、promise 或其他进程内对象：

py
from langgraph.stream import ProtocolEvent, StreamChannel, StreamTransformer

class StatsTransformer(StreamTransformer):
    required_stream_modes = ("messages",)

    def __init__(self, scope: tuple[str, ...] = ()) -> None:
        super().__init__(scope)
        self.total_tokens = 0
        self.total_tokens_log = StreamChannel[int]()

    def init(self) -> dict:
        return {"total_tokens": self.total_tokens_log}

    def process(self, event: ProtocolEvent) -> bool:
        data = event["params"]["data"]
        if isinstance(data, dict):
            usage = data.get("usage") or {}
            self.total_tokens += usage.get("output_tokens") or 0
        return True

    def finalize(self) -> None:
        self.total_tokens_log.push(self.total_tokens)
        self.total_tokens_log.close()

在调用时或编译时注册

在调用时传入转换器，以便在本地进行试验：

py
stream = graph.stream_events(
    input,
    version="v3",
    transformers=[StatsTransformer, ToolActivityTransformer],
)

当该图的每次运行都应产生该投影时，将转换器编译进图中：

py
graph = builder.compile(
    transformers=[StatsTransformer, ToolActivityTransformer],
)

内置：ToolCallTransformer

LangGraph 内置提供了 ToolCallTransformer。注册它以在普通的 StateGraph 上暴露 stream.tool_calls：

py
from langgraph.prebuilt import ToolCallTransformer

stream = graph.stream_events(input, version="v3", transformers=[ToolCallTransformer])

for tool_call in stream.tool_calls:
    print(tool_call.tool_name, tool_call.input)

相关

LangGraph 定义了流式传输原语。若要在 LangChain 或 Deep Agents 中使用流式传输，请查阅相关的产品文档：

线级事件和命令格式定义在 Agent Protocol 仓库中，并可作为 PyPI 上的 langchain-protocol 和 npm 上的 @langchain/protocol 使用。

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可