本页目录
Advanced / 高级能力Note 80

第三章：Event Streaming

Event Streaming 是把一次 Agent 运行拆成连续事件：模型消息、工具开始/结束、状态快照和最终输出都会按运行过程吐出来。它解决的是前端不能只等最终答案的问题，IDE、聊天窗口、工具执行面板都需要边跑边更新。Deep Agents 继承 LangGraph 的流式协议，并额外提供面向子代理的 subag

全部笔记查看源文档

Event Streaming 是把一次 Agent 运行拆成连续事件：模型消息、工具开始/结束、状态快照和最终输出都会按运行过程吐出来。它解决的是前端不能只等最终答案的问题，IDE、聊天窗口、工具执行面板都需要边跑边更新。Deep Agents 继承 LangGraph 的流式协议，并额外提供面向子代理的 subagents 投影。

最小代码

代码在 deepagent_src/advanced_teach/03_event_streaming.py。

这个例子只做一件事：让真实 LLM 必须调用 echo_topic 工具，然后通过 agent.stream_events(..., version="v3") 收集同一次运行里的 messages、tools、values 和最终 output。

关键逻辑：

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

for event in stream:
    method = event.get("method")
    if method == "messages":
        # AIMessage，可能包含 tool_calls，也可能包含最终文本
        ...
    if method == "tools":
        # tool-started / tool-finished
        ...

output = stream.output

这里直接迭代原始事件，而不是同时消费多个 projection。原因很简单：GraphRunStream 的 projection 是单消费者模型，同一条流不要被多个循环抢着读，复杂 UI 再按需要用 interleave(...) 或 tee(...)。

stream_events、stream、astream 的关系

这几个 API 都是在驱动同一个 LangGraph/Deep Agents 运行，只是返回形态不同。stream / astream 是 stream-mode API：你传 stream_mode="updates"、"values"、"messages"、"custom" 等模式，然后自己按 chunk 类型分支处理。stream_events / astream_events 是 Event Streaming API：它返回 run stream 对象，直接给你 messages、tool_calls、values、subagents、output 这些投影。

API 同步/异步 返回什么 适合场景 当前建议
agent.stream(...) 同步 Iterator，按 stream_mode 吐 chunk 旧代码、调试 LangGraph runtime、只要某个 stream mode 能用，但新前端优先不用它
agent.astream(...) 异步 AsyncIterator，按 stream_mode 吐 chunk async 服务端、WebSocket/SSE 中直接异步转发 和 stream 一样，是 stream-mode 路线
agent.stream_events(..., version="v3") 同步 GraphRunStream，有 typed projections 和原始事件 本地脚本、同步服务端、教学验证、前端协议整理 新应用和前端优先选它
agent.astream_events(..., version="v3") 异步 AsyncGraphRunStream，异步 typed projections FastAPI、SSE、WebSocket、并发消费多个投影 生产服务端最常用

它们的关系可以这么记：stream/astream 是“按模式吐运行片段”，stream_events/astream_events 是“按语义通道吐运行事件”。官方新文档更推荐新应用使用 Event Streaming，因为前端通常关心的是消息、工具、状态、子代理这些语义通道，而不是自己解析一堆 stream_mode chunk。

同步和异步只是消费方式不同，不改变 Agent 本身：脚本里用 stream_events 简单；FastAPI 这类异步后端用 astream_events，这样不会堵住事件循环。stream 和 astream 仍然有价值，尤其是你已经有基于 stream_mode="updates"、stream_mode="messages" 的旧代码，或者你正在调 LangGraph 底层 stream mode。

一个最小对照：

# stream-mode：自己根据 chunk 类型分支
for chunk in agent.stream(input, stream_mode=["messages", "updates"], version="v2"):
    print(chunk["type"], chunk["data"])

# event-streaming：直接消费语义投影
stream = agent.stream_events(input, version="v3")
for message in stream.messages:
    print(message.text)
final_state = stream.output

运行命令

uv run python -m deepagent_src.advanced_teach.03_event_streaming

这会触发一次真实 LLM 调用，使用项目里的 get_gpt_model(disable_tool_streaming=True)。如果没有 CHATGPT_API_KEY 或 CHATGPT_API_URL，真实调用会失败。

预期现象

输出里应该能看到：

methods: values, messages, values, tools, tools, values, messages, values
tool_calls: echo_topic
tool_results: 工具已收到主题: event streaming
final: ... event streaming ... 前端 ...
message_count: 4
event streaming real call ok

顺序可能随 beta 版本略有变化，但核心应该成立：先有状态快照，再有模型 tool call，再有工具事件，最后有模型最终消息。

常见误区

不要把 stream.messages、stream.tool_calls、stream.values 当成可以随便重复遍历的列表。GraphRunStream 是运行中的流，projection 是单消费者；你遍历一次就等于驱动 Agent 往前跑。前端要多面板同时显示时，用官方的并发消费方式，或者在自己的事件总线里把原始事件分发出去。

另一个误区是把 stream.subgraphs 当成产品 UI。Deep Agents 文档更推荐用户界面使用 stream.subagents，因为它表达的是“代理委派任务”，不是 LangGraph 内部节点结构。

验证

本章验证点：

  1. messages 事件出现，说明模型消息可流出。
  2. tools 事件出现，说明工具开始/结束可被前端观察。
  3. values 事件出现，说明状态快照可被观察。
  4. 最终 stream.output 能拿到完整 state。
  5. 断言 echo_topic 被真实调用，最终文本包含 event streaming 和 前端。

官方依据：/oss/python/deepagents/event-streaming 说明 Deep Agents 使用 agent.stream_events(input, version="v3")，并暴露 messages、tool_calls、values、subagents、output、interleave(...) 等投影；其中 subagents 是 Deep Agents 面向产品 UI 的重点能力。

相关资源

  • 查看示例代码：deepagent_src/advanced_teach/03_event_streaming.py 
    from __future__ import annotations

import os
from typing import Any

from deepagents import create_deep_agent
from langchain.messages import HumanMessage
from langchain_core.tools import tool

from deepagent_src.llms import get_gpt_model


@tool
def echo_topic(topic: str) -> str:
    """Echo a short topic for event streaming teaching."""
    return f"工具已收到主题: {topic}"


def message_text(message: Any) -> str:
    text = getattr(message, "text", None)
    if isinstance(text, str):
        return text
    content = getattr(message, "content", "")
    return content if isinstance(content, str) else str(content)


def collect_stream_events() -> dict[str, Any]:
    os.environ["LANGSMITH_TRACING"] = "false"
    os.environ["LANGCHAIN_TRACING_V2"] = "false"

    agent = create_deep_agent(
        model=get_gpt_model(disable_tool_streaming=True),
        tools=[echo_topic],
        subagents=[],
        system_prompt=(
            "你是 Deep Agents Event Streaming 教学助手。必须先调用 echo_topic 工具，"
            "topic 填 event streaming，然后用一句中文回答。最终回答必须包含：event streaming 前端。"
        ),
    )

    stream = agent.stream_events(
        {
            "messages": [
                HumanMessage(
                    content="请调用工具，然后说明 event streaming 能给前端带来什么。"
                )
            ]
        },
        version="v3",
    )

    methods: list[str] = []
    tool_calls: list[str] = []
    tool_results: list[str] = []
    final_text = ""

    for event in stream:
        method = event.get("method")
        if method:
            methods.append(method)

        params = event.get("params", {})
        data = params.get("data")

        if method == "messages" and isinstance(data, tuple):
            message = data[0]
            for call in getattr(message, "tool_calls", None) or []:
                tool_calls.append(call.get("name", ""))
            text = message_text(message)
            if text:
                final_text = text

        if method == "tools" and isinstance(data, dict):
            if data.get("event") == "tool-finished":
                output = data.get("output")
                tool_results.append(message_text(output))

    output = stream.output or {}
    return {
        "methods": methods,
        "tool_calls": tool_calls,
        "tool_results": tool_results,
        "final_text": final_text,
        "message_count": len(output.get("messages", [])),
    }


def main() -> None:
    result = collect_stream_events()
    print("methods:", ", ".join(result["methods"]))
    print("tool_calls:", ", ".join(result["tool_calls"]))
    print("tool_results:", " | ".join(result["tool_results"]))
    print("final:", result["final_text"])
    print("message_count:", result["message_count"])

    assert "messages" in result["methods"], result
    assert "tools" in result["methods"], result
    assert "values" in result["methods"], result
    assert "echo_topic" in result["tool_calls"], result
    assert any("event streaming" in item for item in result["tool_results"]), result
    assert "event streaming" in result["final_text"].lower(), result
    assert "前端" in result["final_text"], result
    assert result["message_count"] >= 3, result
    print("event streaming real call ok")


if __name__ == "__main__":
    main()