> ## Documentation Index
> Fetch the complete documentation index at: https://nvd-54.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 流式输出

> 从深度智能体运行和子智能体执行中流式获取实时更新

<Info>
  **预览中：** 尝试事件流式的深度智能体类型投影，涵盖协调器消息、委派子智能体、工具调用和最终输出。从[深度智能体事件流](/oss/python/deepagents/event-streaming)开始，或在[流式 cookbook](https://github.com/langchain-ai/streaming-cookbook) 中探索可运行示例。
</Info>

深度智能体基于 LangGraph 的流式基础设施构建，原生支持子智能体流。当深度智能体将工作委派给子智能体时，你可以独立流式获取每个子智能体的更新 — 实时跟踪进度、LLM Token 和工具调用。

深度智能体流式可以实现的功能：

* <Icon icon="diagram-subtask" size={16} /> [**流式子智能体进度**](#subagent-progress) — 跟踪每个子智能体并行运行时的执行情况。
* <Icon icon="square-binary" size={16} /> [**流式 LLM Token**](#llm-tokens) — 从主智能体和每个子智能体流式获取 Token。
* <Icon icon="screwdriver-wrench" size={16} /> [**流式工具调用**](#tool-calls) — 查看子智能体执行中的工具调用和结果。
* <Icon icon="table" size={16} /> [**流式自定义更新**](#custom-updates) — 从子智能体节点内部发出用户定义的信号。

## 启用子图流式

深度智能体使用 LangGraph 的子图流式来展示来自子智能体执行的事件。要接收子智能体事件，在流式时启用 `stream_subgraphs`。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from deepagents import create_deep_agent

agent = create_deep_agent(
    model="google_genai:gemini-3.1-pro-preview",
    system_prompt="You are a helpful research assistant",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="updates",
    subgraphs=True,  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            # 子智能体事件 - 命名空间标识来源
            print(f"[subagent: {chunk['ns']}]")
        else:
            # 主智能体事件
            print("[main agent]")
        print(chunk["data"])
```

## 命名空间

当启用 `subgraphs` 时，每个流式事件包含一个**命名空间**，标识哪个智能体产生了它。命名空间是节点名称和任务 ID 的路径，表示智能体层级。

| 命名空间                                       | 来源                                  |
| ------------------------------------------ | ----------------------------------- |
| `()`（空）                                    | 主智能体                                |
| `("tools:abc123",)`                        | 由主智能体的 `task` 工具调用 `abc123` 生成的子智能体 |
| `("tools:abc123", "model_request:def456")` | 子智能体内部的模型请求节点                       |

使用命名空间将事件路由到正确的 UI 组件：

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Plan my vacation"}]},
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "updates":
        # 检查此事件是否来自子智能体
        is_subagent = any(
            segment.startswith("tools:") for segment in chunk["ns"]
        )

        if is_subagent:
            # 从命名空间提取工具调用 ID
            tool_call_id = next(
                s.split(":")[1] for s in chunk["ns"] if s.startswith("tools:")
            )
            print(f"Subagent {tool_call_id}: {chunk['data']}")
        else:
            print(f"Main agent: {chunk['data']}")
```

## 子智能体进度

使用 `stream_mode="updates"` 在每个步骤完成时跟踪子智能体进度。这对于显示哪些子智能体处于活动状态及其已完成的工作很有用。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from deepagents import create_deep_agent

agent = create_deep_agent(
    model="google_genai:gemini-3.1-pro-preview",
    system_prompt=(
        "You are a project coordinator. Always delegate research tasks "
        "to your researcher subagent using the task tool. Keep your final response to one sentence."
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "Researches topics thoroughly",
            "system_prompt": (
                "You are a thorough researcher. Research the given topic "
                "and provide a concise summary in 2-3 sentences."
            ),
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "updates":
        # 主智能体更新（空命名空间）
        if not chunk["ns"]:
            for node_name, data in chunk["data"].items():
                if node_name == "tools":
                    # 子智能体结果返回给主智能体
                    for msg in data.get("messages", []):
                        if msg.type == "tool":
                            print(f"\n子智能体完成: {msg.name}")
                            print(f"  结果: {str(msg.content)[:200]}...")
                else:
                    print(f"[主智能体] 步骤: {node_name}")

        # 子智能体更新（非空命名空间）
        else:
            for node_name, data in chunk["data"].items():
                print(f"  [{chunk['ns'][0]}] 步骤: {node_name}")
```

```shell title="输出" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
[主智能体] 步骤: model_request
  [tools:call_abc123] 步骤: model_request
  [tools:call_abc123] 步骤: tools
  [tools:call_abc123] 步骤: model_request

子智能体完成: task
  结果: ## AI Safety Report...
[主智能体] 步骤: model_request
```

## LLM Token

使用 `stream_mode="messages"` 从主智能体和子智能体流式获取单个 Token。每个消息事件包含标识来源智能体的元数据。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
current_source = ""

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="messages",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]

        # 检查此事件是否来自子智能体（命名空间包含 "tools:"）
        is_subagent = any(s.startswith("tools:") for s in chunk["ns"])

        if is_subagent:
            # 来自子智能体的 Token
            subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
            if subagent_ns != current_source:
                print(f"\n\n--- [子智能体: {subagent_ns}] ---")
                current_source = subagent_ns
            if token.content:
                print(token.content, end="", flush=True)
        else:
            # 来自主智能体的 Token
            if "main" != current_source:
                print("\n\n--- [主智能体] ---")
                current_source = "main"
            if token.content:
                print(token.content, end="", flush=True)

print()
```

## 工具调用

当子智能体使用工具时，你可以流式获取工具调用事件来显示每个子智能体正在做什么。工具调用块出现在 `messages` 流模式中。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research recent quantum computing advances"}]},
    stream_mode="messages",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]

        # 标识来源："main" 或子智能体命名空间段
        is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
        source = next((s for s in chunk["ns"] if s.startswith("tools:")), "main") if is_subagent else "main"

        # 工具调用块（流式工具调用）
        if token.tool_call_chunks:
            for tc in token.tool_call_chunks:
                if tc.get("name"):
                    print(f"\n[{source}] 工具调用: {tc['name']}")
                # 参数分块流入 - 增量写入
                if tc.get("args"):
                    print(tc["args"], end="", flush=True)

        # 工具结果
        if token.type == "tool":
            print(f"\n[{source}] 工具结果 [{token.name}]: {str(token.content)[:150]}")

        # 普通 AI 内容（跳过工具调用消息）
        if token.type == "ai" and token.content and not token.tool_call_chunks:
            print(token.content, end="", flush=True)

print()
```

## 自定义更新

在子智能体工具内部使用 [`get_stream_writer`](https://reference.langchain.com/python/langgraph/config/get_stream_writer) 发出自定义进度事件：

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import time
from langchain.tools import tool
from langgraph.config import get_stream_writer
from deepagents import create_deep_agent


@tool
def analyze_data(topic: str) -> str:
    """对给定主题运行数据分析。

    此工具执行实际分析并发出进度更新。
    你必须为任何分析请求调用此工具。
    """
    writer = get_stream_writer()

    writer({"status": "starting", "topic": topic, "progress": 0})
    time.sleep(0.5)

    writer({"status": "analyzing", "progress": 50})
    time.sleep(0.5)

    writer({"status": "complete", "progress": 100})
    return (
        f'Analysis of "{topic}": Customer sentiment is 85% positive, '
        "driven by product quality and support response times."
    )


agent = create_deep_agent(
    model="google_genai:gemini-3.1-pro-preview",
    system_prompt=(
        "You are a coordinator. For any analysis request, you MUST delegate "
        "to the analyst subagent using the task tool. Never try to answer directly. "
        "After receiving the result, summarize it in one sentence."
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "Performs data analysis with real-time progress tracking",
            "system_prompt": (
                "You are a data analyst. You MUST call the analyze_data tool "
                "for every analysis request. Do not use any other tools. "
                "After the analysis completes, report the result."
            ),
            "tools": [analyze_data],
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "custom":
        is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
        if is_subagent:
            subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
            print(f"[{subagent_ns}]", chunk["data"])
        else:
            print("[main]", chunk["data"])
```

```shell title="输出" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
[tools:call_abc123] {'status': 'starting', 'topic': 'customer satisfaction trends', 'progress': 0}
[tools:call_abc123] {'status': 'analyzing', 'progress': 50}
[tools:call_abc123] {'status': 'complete', 'progress': 100}
```

## 流式多种模式

组合多种流模式以获取智能体执行的完整视图：

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
# 跳过内部中间件步骤 - 只显示有意义的节点名称
INTERESTING_NODES = {"model_request", "tools"}

last_source = ""
mid_line = False  # 当我们写了没有尾换行的 Token 时为 True

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze the impact of remote work on team productivity"}]},
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
    version="v2",
):
    is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
    source = "subagent" if is_subagent else "main"

    if chunk["type"] == "updates":
        for node_name in chunk["data"]:
            if node_name not in INTERESTING_NODES:
                continue
            if mid_line:
                print()
                mid_line = False
            print(f"[{source}] 步骤: {node_name}")

    elif chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if token.content:
            # 当来源变化时打印标题
            if source != last_source:
                if mid_line:
                    print()
                    mid_line = False
                print(f"\n[{source}] ", end="")
                last_source = source
            print(token.content, end="", flush=True)
            mid_line = True

    elif chunk["type"] == "custom":
        if mid_line:
            print()
            mid_line = False
        print(f"[{source}] 自定义事件:", chunk["data"])

print()
```

## 常见模式

### 跟踪子智能体生命周期

监控子智能体何时开始、运行和完成：

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
active_subagents = {}

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research the latest AI safety developments"}]},
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "updates":
        for node_name, data in chunk["data"].items():
            # ─── 阶段 1：检测子智能体启动 ────────────────────────
            # 当主智能体的 model_request 包含 task 工具调用时，
            # 一个子智能体已被生成。
            if not chunk["ns"] and node_name == "model_request":
                for msg in data.get("messages", []):
                    for tc in getattr(msg, "tool_calls", []):
                        if tc["name"] == "task":
                            active_subagents[tc["id"]] = {
                                "type": tc["args"].get("subagent_type"),
                                "description": tc["args"].get("description", "")[:80],
                                "status": "pending",
                            }
                            print(
                                f'[lifecycle] PENDING  → 子智能体 "{tc["args"].get("subagent_type")}" '
                                f'({tc["id"]})'
                            )

            # ─── 阶段 2：检测子智能体运行 ─────────────────────────
            # 当我们从 tools:UUID 命名空间接收到事件时，该
            # 子智能体正在活跃执行。
            if chunk["ns"] and chunk["ns"][0].startswith("tools:"):
                pregel_id = chunk["ns"][0].split(":")[1]
                for sub_id, sub in active_subagents.items():
                    if sub["status"] == "pending":
                        sub["status"] = "running"
                        print(
                            f'[lifecycle] RUNNING  → 子智能体 "{sub["type"]}" '
                            f"(pregel: {pregel_id})"
                        )
                        break

            # ─── 阶段 3：检测子智能体完成 ──────────────────────
            # 当主智能体的 tools 节点返回工具消息时，
            # 子智能体已完成并返回其结果。
            if not chunk["ns"] and node_name == "tools":
                for msg in data.get("messages", []):
                    if msg.type == "tool":
                        sub = active_subagents.get(msg.tool_call_id)
                        if sub:
                            sub["status"] = "complete"
                            print(
                                f'[lifecycle] COMPLETE → 子智能体 "{sub["type"]}" '
                                f"({msg.tool_call_id})"
                            )
                            print(f"  结果预览: {str(msg.content)[:120]}...")

# 打印最终状态
print("\n--- 子智能体最终状态 ---")
for sub_id, sub in active_subagents.items():
    print(f"  {sub['type']}: {sub['status']}")
```

## v2 流式格式

<Note>
  需要 LangGraph >= 1.1。
</Note>

本页所有示例使用 v2 流式格式（`version="v2"`），这是推荐的方法。每个块是一个具有 `type`、`ns` 和 `data` 键的 `StreamPart` 字典 — 无论流模式、模式数量或子图设置如何，形状都相同。

v2 格式消除了嵌套元组解包，使在深度智能体中处理子图流式变得简单直接。比较两种格式：

<CodeGroup>
  ```python v2（推荐） theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  # 统一格式 — 无嵌套元组解包
  for chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Research quantum computing"}]},
      stream_mode=["updates", "messages", "custom"],
      subgraphs=True,
      version="v2",
  ):
      print(chunk["type"])  # "updates"、"messages" 或 "custom"
      print(chunk["ns"])    # () 为主智能体，("tools:<id>",) 为子智能体
      print(chunk["data"])  # 载荷
  ```

  ```python v1（旧版） theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  # 必须处理 (namespace, (mode, data)) 嵌套元组
  for namespace, chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Research quantum computing"}]},
      stream_mode=["updates", "messages", "custom"],
      subgraphs=True,
  ):
      mode, data = chunk[0], chunk[1]
      print(mode)       # "updates"、"messages" 或 "custom"
      print(namespace)  # () 为主智能体，("tools:<id>",) 为子智能体
      print(data)       # 载荷
  ```
</CodeGroup>

参见 [LangGraph 流式文档](/oss/python/langgraph/streaming#stream-output-format-v2)了解更多关于 v2 格式的详情，包括类型收窄和 Pydantic/dataclass 强制转换。

## 相关

* [子智能体](/oss/python/deepagents/subagents) — 使用深度智能体配置和使用子智能体
* [前端流式](/oss/python/deepagents/streaming/frontend) — 使用 `useStream` 构建深度智能体的 React UI
* [LangChain 流式概览](/oss/python/langchain/streaming/overview) — LangChain 智能体的通用流式概念

***

<div className="source-links">
  <Callout icon="terminal-2">
    [连接这些文档](/use-these-docs)到 Claude、VSCode 等，通过 MCP 获取实时答案。
  </Callout>

  <Callout icon="edit">
    [在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/deepagents/streaming.mdx)或[提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。
  </Callout>
</div>