流式传输

对于新应用，我们推荐使用事件流—这是 Deep Agents v0.6 中引入的类型化投影 API。事件流为每个投影（子代理、消息、工具调用、值）提供独立的迭代器，因此您可以独立消费它们，而不是基于 stream_mode 块进行分支处理。

Deep Agents 建立在 LangGraph 的流式基础设施之上，并对子代理流提供一流的支持。当深度代理将工作委派给子代理时，您可以独立地从每个子代理流式获取更新—实时跟踪进度、LLM 令牌和工具调用。深度代理流式传输可以实现：

流式获取子代理进度—在子代理并行运行时跟踪其执行过程。
流式获取 LLM 令牌—从主代理和每个子代理流式获取令牌。
流式获取工具调用—查看子代理执行过程中的工具调用和结果。
流式获取自定义更新—从子代理节点内部发出用户定义的信号。

启用子图流式传输

Deep Agents 使用 LangGraph 的子图流式传输来呈现子代理执行过程中的事件。要接收子代理事件，请在流式传输时启用 stream_subgraphs。

from deepagents import create_deep_agent
    
agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    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,
    version="v2",
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            # 子代理事件 - 命名空间标识来源
            print(f"[subagent: {chunk['ns']}]")
        else:
            # 主代理事件
            print("[main agent]")
        print(chunk["data"])

from deepagents import create_deep_agent
    
agent = create_deep_agent(
    model="openai:gpt-5.5",
    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,
    version="v2",
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            # 子代理事件 - 命名空间标识来源
            print(f"[subagent: {chunk['ns']}]")
        else:
            # 主代理事件
            print("[main agent]")
        print(chunk["data"])

from deepagents import create_deep_agent
    
agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    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,
    version="v2",
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            # 子代理事件 - 命名空间标识来源
            print(f"[subagent: {chunk['ns']}]")
        else:
            # 主代理事件
            print("[main agent]")
        print(chunk["data"])

from deepagents import create_deep_agent
    
agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    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,
    version="v2",
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            # 子代理事件 - 命名空间标识来源
            print(f"[subagent: {chunk['ns']}]")
        else:
            # 主代理事件
            print("[main agent]")
        print(chunk["data"])

from deepagents import create_deep_agent
    
agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    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,
    version="v2",
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            # 子代理事件 - 命名空间标识来源
            print(f"[subagent: {chunk['ns']}]")
        else:
            # 主代理事件
            print("[main agent]")
        print(chunk["data"])

from deepagents import create_deep_agent
    
agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    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,
    version="v2",
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            # 子代理事件 - 命名空间标识来源
            print(f"[subagent: {chunk['ns']}]")
        else:
            # 主代理事件
            print("[main agent]")
        print(chunk["data"])

from deepagents import create_deep_agent
    
agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    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,
    version="v2",
):
    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 组件：

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"子代理 {tool_call_id}: {chunk['data']}")
        else:
            print(f"主代理: {chunk['data']}")

子代理进度

使用 stream_mode="updates" 在每一步完成时跟踪子代理进度。这对于显示哪些子代理处于活动状态以及它们已完成的工作非常有用。

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    system_prompt=(
        "你是一个没有研究知识的项目协调员。"
        "对于每个用户请求，你必须使用 subagent_type 设置为 researcher 来调用 task() 工具。"
        "切勿自己回答研究问题。你的最终回复请保持为一句话。"
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "深入研究主题",
            "system_prompt": (
                "你是一个深入的研究员。研究给定主题，"
                "并用2-3句话提供一个简洁的摘要。"
            ),
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "写一段关于人工智能安全的简短摘要"}]},
    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}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-5.5",
    system_prompt=(
        "你是一个没有研究知识的项目协调员。"
        "对于每个用户请求，你必须使用 subagent_type 设置为 researcher 来调用 task() 工具。"
        "切勿自己回答研究问题。你的最终回复请保持为一句话。"
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "深入研究主题",
            "system_prompt": (
                "你是一个深入的研究员。研究给定主题，"
                "并用2-3句话提供一个简洁的摘要。"
            ),
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "写一段关于人工智能安全的简短摘要"}]},
    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}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    system_prompt=(
        "你是一个没有研究知识的项目协调员。"
        "对于每个用户请求，你必须使用 subagent_type 设置为 researcher 来调用 task() 工具。"
        "切勿自己回答研究问题。你的最终回复请保持为一句话。"
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "深入研究主题",
            "system_prompt": (
                "你是一个深入的研究员。研究给定主题，"
                "并用2-3句话提供一个简洁的摘要。"
            ),
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "写一段关于人工智能安全的简短摘要"}]},
    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}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    system_prompt=(
        "你是一个没有研究知识的项目协调员。"
        "对于每个用户请求，你必须使用 subagent_type 设置为 researcher 来调用 task() 工具。"
        "切勿自己回答研究问题。你的最终回复请保持为一句话。"
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "深入研究主题",
            "system_prompt": (
                "你是一个深入的研究员。研究给定主题，"
                "并用2-3句话提供一个简洁的摘要。"
            ),
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "写一段关于人工智能安全的简短摘要"}]},
    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}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    system_prompt=(
        "你是一个没有研究知识的项目协调员。"
        "对于每个用户请求，你必须使用 subagent_type 设置为 researcher 来调用 task() 工具。"
        "切勿自己回答研究问题。你的最终回复请保持为一句话。"
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "深入研究主题",
            "system_prompt": (
                "你是一个深入的研究员。研究给定主题，"
                "并用2-3句话提供一个简洁的摘要。"
            ),
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "写一段关于人工智能安全的简短摘要"}]},
    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}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    system_prompt=(
        "你是一个没有研究知识的项目协调员。"
        "对于每个用户请求，你必须使用 subagent_type 设置为 researcher 来调用 task() 工具。"
        "切勿自己回答研究问题。你的最终回复请保持为一句话。"
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "深入研究主题",
            "system_prompt": (
                "你是一个深入的研究员。研究给定主题，"
                "并用2-3句话提供一个简洁的摘要。"
            ),
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "写一段关于人工智能安全的简短摘要"}]},
    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}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    system_prompt=(
        "你是一个没有研究知识的项目协调员。"
        "对于每个用户请求，你必须使用 subagent_type 设置为 researcher 来调用 task() 工具。"
        "切勿自己回答研究问题。你的最终回复请保持为一句话。"
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "深入研究主题",
            "system_prompt": (
                "你是一个深入的研究员。研究给定主题，"
                "并用2-3句话提供一个简洁的摘要。"
            ),
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "写一段关于人工智能安全的简短摘要"}]},
    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}")

Output

[main agent] step: model_request
  [tools:call_abc123] step: model_request
  [tools:call_abc123] step: tools
  [tools:call_abc123] step: model_request

Subagent complete: task
  Result: ## AI Safety Report...
[main agent] step: model_request

LLM 令牌

使用 stream_mode="messages" 流式传输来自主代理和各个子代理的单独令牌。每个消息事件都包含标识源代理的元数据。

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:
            # 来自子代理的令牌
            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:
            # 来自主代理的令牌
            if "main" != current_source:
                print("\n\n--- [主代理] ---")
                current_source = "main"
            if token.content:
                print(token.content, end="", flush=True)

print()

工具调用

当子代理使用工具时，您可以流式传输工具调用事件以显示每个子代理正在执行的操作。工具调用块会出现在 messages 流模式中。

from langchain.messages import AIMessageChunk, ToolMessage

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "研究最新的量子计算进展"}]},
    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 isinstance(token, AIMessageChunk) and 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 isinstance(token, ToolMessage):
            print(f"\n[{source}] 工具结果 [{token.name}]: {str(token.content)[:150]}")

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

print()

自定义更新

在子代理工具内部使用 get_stream_writer 来发出自定义进度事件：

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'对“{topic}”的分析结果：客户情绪为85%正面，'
        "这主要归因于产品质量和支持响应时间。"
    )


agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    system_prompt=(
        "你是一个协调员。对于任何分析请求，你**必须**"
        "使用任务工具将其委派给分析师子代理。切勿尝试直接回答。"
        "收到结果后，请用一句话进行总结。"
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "执行带有实时进度跟踪的数据分析",
            "system_prompt": (
                "你是一名数据分析师。对于每个分析请求，你**必须**调用 analyze_data 工具。"
                "不要使用任何其他工具。"
                "分析完成后，请报告结果。"
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "custom":
        custom_event_count += 1
        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"])

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'对“{topic}”的分析结果：客户情绪为85%正面，'
        "这主要归因于产品质量和支持响应时间。"
    )


agent = create_deep_agent(
    model="openai:gpt-5.5",
    system_prompt=(
        "你是一个协调员。对于任何分析请求，你**必须**"
        "使用任务工具将其委派给分析师子代理。切勿尝试直接回答。"
        "收到结果后，请用一句话进行总结。"
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "执行带有实时进度跟踪的数据分析",
            "system_prompt": (
                "你是一名数据分析师。对于每个分析请求，你**必须**调用 analyze_data 工具。"
                "不要使用任何其他工具。"
                "分析完成后，请报告结果。"
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "custom":
        custom_event_count += 1
        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"])

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'对“{topic}”的分析结果：客户情绪为85%正面，'
        "这主要归因于产品质量和支持响应时间。"
    )


agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    system_prompt=(
        "你是一个协调员。对于任何分析请求，你**必须**"
        "使用任务工具将其委派给分析师子代理。切勿尝试直接回答。"
        "收到结果后，请用一句话进行总结。"
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "执行带有实时进度跟踪的数据分析",
            "system_prompt": (
                "你是一名数据分析师。对于每个分析请求，你**必须**调用 analyze_data 工具。"
                "不要使用任何其他工具。"
                "分析完成后，请报告结果。"
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "custom":
        custom_event_count += 1
        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"])

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'对“{topic}”的分析结果：客户情绪为85%正面，'
        "这主要归因于产品质量和支持响应时间。"
    )


agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    system_prompt=(
        "你是一个协调员。对于任何分析请求，你**必须**"
        "使用任务工具将其委派给分析师子代理。切勿尝试直接回答。"
        "收到结果后，请用一句话进行总结。"
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "执行带有实时进度跟踪的数据分析",
            "system_prompt": (
                "你是一名数据分析师。对于每个分析请求，你**必须**调用 analyze_data 工具。"
                "不要使用任何其他工具。"
                "分析完成后，请报告结果。"
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "custom":
        custom_event_count += 1
        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"])

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'对“{topic}”的分析结果：客户情绪为85%正面，'
        "这主要归因于产品质量和支持响应时间。"
    )


agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    system_prompt=(
        "你是一个协调员。对于任何分析请求，你**必须**"
        "使用任务工具将其委派给分析师子代理。切勿尝试直接回答。"
        "收到结果后，请用一句话进行总结。"
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "执行带有实时进度跟踪的数据分析",
            "system_prompt": (
                "你是一名数据分析师。对于每个分析请求，你**必须**调用 analyze_data 工具。"
                "不要使用任何其他工具。"
                "分析完成后，请报告结果。"
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "custom":
        custom_event_count += 1
        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"])

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'对“{topic}”的分析结果：客户情绪为85%正面，'
        "这主要归因于产品质量和支持响应时间。"
    )


agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    system_prompt=(
        "你是一个协调员。对于任何分析请求，你**必须**"
        "使用任务工具将其委派给分析师子代理。切勿尝试直接回答。"
        "收到结果后，请用一句话进行总结。"
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "执行带有实时进度跟踪的数据分析",
            "system_prompt": (
                "你是一名数据分析师。对于每个分析请求，你**必须**调用 analyze_data 工具。"
                "不要使用任何其他工具。"
                "分析完成后，请报告结果。"
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "custom":
        custom_event_count += 1
        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"])

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'对“{topic}”的分析结果：客户情绪为85%正面，'
        "这主要归因于产品质量和支持响应时间。"
    )


agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    system_prompt=(
        "你是一个协调员。对于任何分析请求，你**必须**"
        "使用任务工具将其委派给分析师子代理。切勿尝试直接回答。"
        "收到结果后，请用一句话进行总结。"
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "执行带有实时进度跟踪的数据分析",
            "system_prompt": (
                "你是一名数据分析师。对于每个分析请求，你**必须**调用 analyze_data 工具。"
                "不要使用任何其他工具。"
                "分析完成后，请报告结果。"
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "custom":
        custom_event_count += 1
        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"])

Output

[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}

流式传输多种模式

组合多种流模式以获得代理执行的完整视图：

# 跳过内部中间件步骤 - 仅显示有意义的节点名称
INTERESTING_NODES = {"model", "tools"}

last_source = ""
mid_line = False  # 当我们写入令牌但没有尾随换行时为真

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "分析远程工作对团队生产力的影响"}]},
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
    version="v2",
):
    is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
    source = "子代理" if is_subagent else "主"

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

常见模式

跟踪子代理生命周期

监控子代理何时启动、运行和完成：

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：检测子智能体启动 ────────────────────────
            # 当主智能体的模型节点包含任务工具调用时，
            # 说明已生成一个子智能体。
            if not chunk["ns"] and node_name == "model":
                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] 挂起 → 子智能体 "{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]
                # 检查是否有待处理的子智能体需要标记为运行中。
                # 注意：pregel 任务 ID 与 tool_call_id 不同，
                # 因此我们在收到第一个子智能体事件时，就将任何待处理的子智能体标记为运行中。
                for sub_id, sub in active_subagents.items():
                    if sub["status"] == "pending":
                        sub["status"] = "running"
                        print(
                            f'[lifecycle] 运行中 → 子智能体 "{sub["type"]}" '
                            f"(pregel: {pregel_id})"
                        )
                        break

            # ─── 阶段 3：检测子智能体完成 ──────────────────────
            # 当主智能体的工具节点返回工具消息时，
            # 表明子智能体已完成并返回了结果。
            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] 完成 → 子智能体 "{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 流式格式

需要 LangGraph >= 1.1。

本页的所有示例都使用 v2 流式格式 (version="v2")，这是推荐的做法。每个块都是一个包含 type、ns 和 data 键的 StreamPart 字典 —— 其结构相同，无论流模式、模式数量或子图设置如何。 v2 格式消除了嵌套元组解包，使得在 Deep Agents 中处理子图流式传输变得简单直接。比较两种格式：

# 统一格式 — 无需嵌套元组解包
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", or "custom"
    print(chunk["ns"])    # () for main agent, ("tools:<id>",) for subagent
    print(chunk["data"])  # payload

# 必须处理 (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", or "custom"
    print(namespace)  # () for main agent, ("tools:<id>",) for subagent
    print(data)       # payload

关于 v2 格式的更多细节，包括类型窄化和 Pydantic/dataclass 强制转换，请参见 LangGraph 流式文档。

Get started

Deployment

Execution environment

Context management

Delegation

Steering

Frontend

Protocols

启用子图流式传输

命名空间

子代理进度

LLM 令牌

工具调用

自定义更新

流式传输多种模式

常见模式

跟踪子代理生命周期

v2 流式格式

相关内容

​启用子图流式传输

​命名空间

​子代理进度

​LLM 令牌

​工具调用

​自定义更新

​流式传输多种模式

​常见模式

​跟踪子代理生命周期

​v2 流式格式

​相关内容

启用子图流式传输

命名空间

子代理进度

LLM 令牌

工具调用

自定义更新

流式传输多种模式

常见模式

跟踪子代理生命周期

v2 流式格式

相关内容