外观
TIP
对于新应用，我们推荐使用事件流——这是 Deep Agents v0.6 中引入的类型化投影 API。事件流为每个投影（子智能体、消息、工具调用、值）提供独立的迭代器，因此你可以独立消费它们，而无需根据 stream_mode 分块进行分支判断。
Deep Agents 基于 LangGraph 的流式基础设施构建，并对子智能体流提供一流的支持。当深度智能体将工作委托给子智能体时，你可以独立地流式输出每个子智能体的更新——实时跟踪进度、LLM token 和工具调用。
深度智能体流式输出可以实现的功能：
- 流式子智能体进度——跟踪每个子智能体并行运行时的执行情况。
- 流式输出 LLM token——从主智能体和每个子智能体流式输出 token。
- 流式输出工具调用——查看子智能体执行过程中的工具调用和结果。
- 流式输出自定义更新——从子智能体节点内部发出用户定义的信号。
启用子图流式输出
Deep Agents 使用 LangGraph 的子图流式输出来呈现子智能体执行的事件。要接收子智能体事件，请在流式输出时启用 stream_subgraphs。
python
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"])python
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"])python
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"])python
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"])python
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"])python
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"])python
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"])ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "google-genai:gemini-3.6-flash",
systemPrompt: "You are a helpful research assistant",
subagents: [
{
name: "researcher",
description: "Researches a topic in depth",
systemPrompt: "You are a thorough researcher.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Research quantum computing advances" },
],
},
{
streamMode: "updates",
subgraphs: true,
},
)) {
if (namespace.length > 0) {
// 子智能体事件——命名空间标识来源
console.log(`[subagent: ${namespace.join("|")}]`);
} else {
// 主智能体事件
console.log("[main agent]");
}
console.log(chunk);
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "openai:gpt-5.5",
systemPrompt: "You are a helpful research assistant",
subagents: [
{
name: "researcher",
description: "Researches a topic in depth",
systemPrompt: "You are a thorough researcher.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Research quantum computing advances" },
],
},
{
streamMode: "updates",
subgraphs: true,
},
)) {
if (namespace.length > 0) {
// 子智能体事件——命名空间标识来源
console.log(`[subagent: ${namespace.join("|")}]`);
} else {
// 主智能体事件
console.log("[main agent]");
}
console.log(chunk);
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "anthropic:claude-sonnet-4-6",
systemPrompt: "You are a helpful research assistant",
subagents: [
{
name: "researcher",
description: "Researches a topic in depth",
systemPrompt: "You are a thorough researcher.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Research quantum computing advances" },
],
},
{
streamMode: "updates",
subgraphs: true,
},
)) {
if (namespace.length > 0) {
// 子智能体事件——命名空间标识来源
console.log(`[subagent: ${namespace.join("|")}]`);
} else {
// 主智能体事件
console.log("[main agent]");
}
console.log(chunk);
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "openrouter:openrouter:z-ai/glm-5.2",
systemPrompt: "You are a helpful research assistant",
subagents: [
{
name: "researcher",
description: "Researches a topic in depth",
systemPrompt: "You are a thorough researcher.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Research quantum computing advances" },
],
},
{
streamMode: "updates",
subgraphs: true,
},
)) {
if (namespace.length > 0) {
// 子智能体事件——命名空间标识来源
console.log(`[subagent: ${namespace.join("|")}]`);
} else {
// 主智能体事件
console.log("[main agent]");
}
console.log(chunk);
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "fireworks:accounts/fireworks/models/glm-5p2",
systemPrompt: "You are a helpful research assistant",
subagents: [
{
name: "researcher",
description: "Researches a topic in depth",
systemPrompt: "You are a thorough researcher.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Research quantum computing advances" },
],
},
{
streamMode: "updates",
subgraphs: true,
},
)) {
if (namespace.length > 0) {
// 子智能体事件——命名空间标识来源
console.log(`[subagent: ${namespace.join("|")}]`);
} else {
// 主智能体事件
console.log("[main agent]");
}
console.log(chunk);
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "baseten:zai-org/GLM-5.2",
systemPrompt: "You are a helpful research assistant",
subagents: [
{
name: "researcher",
description: "Researches a topic in depth",
systemPrompt: "You are a thorough researcher.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Research quantum computing advances" },
],
},
{
streamMode: "updates",
subgraphs: true,
},
)) {
if (namespace.length > 0) {
// 子智能体事件——命名空间标识来源
console.log(`[subagent: ${namespace.join("|")}]`);
} else {
// 主智能体事件
console.log("[main agent]");
}
console.log(chunk);
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "ollama:north-mini-code-1.0",
systemPrompt: "You are a helpful research assistant",
subagents: [
{
name: "researcher",
description: "Researches a topic in depth",
systemPrompt: "You are a thorough researcher.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Research quantum computing advances" },
],
},
{
streamMode: "updates",
subgraphs: true,
},
)) {
if (namespace.length > 0) {
// 子智能体事件——命名空间标识来源
console.log(`[subagent: ${namespace.join("|")}]`);
} else {
// 主智能体事件
console.log("[main agent]");
}
console.log(chunk);
}命名空间
当启用 subgraphs 时，每个流式事件都包含一个命名空间，用于标识产生该事件的智能体。命名空间是代表智能体层级的节点名称和任务 ID 的路径。
| 命名空间 | 来源 |
|---|---|
() （空） | 主智能体 |
("tools:abc123",) | 由主智能体的 task 工具调用 abc123 派生的子智能体 |
("tools:abc123", "model_request:def456") | 子智能体内的模型请求节点 |
使用命名空间将事件路由到正确的 UI 组件：
python
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']}")ts
for await (const [namespace, chunk] of await agent.stream(
{ messages: [{ role: "user", content: "Plan my vacation" }] },
{ streamMode: "updates", subgraphs: true },
)) {
// 检查此事件是否来自子智能体
const isSubagent = namespace.some((segment: string) =>
segment.startsWith("tools:"),
);
if (isSubagent) {
// 从命名空间中提取工具调用 ID
const toolCallId = namespace
.find((s: string) => s.startsWith("tools:"))
?.split(":")[1];
console.log(`Subagent ${toolCallId}:`, chunk);
} else {
console.log("Main agent:", chunk);
}
}子智能体进度
使用 stream_mode="updates" 在每个步骤完成时跟踪子智能体进度。这对于显示哪些子智能体处于活动状态以及它们完成了哪些工作非常有用。
python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
system_prompt=(
"You are a project coordinator with no research knowledge. "
"For every user request, you must call the task() tool with "
"subagent_type set to researcher. Never answer research questions yourself. "
"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"\nSubagent complete: {msg.name}")
print(f" Result: {str(msg.content)[:200]}...")
else:
print(f"[main agent] step: {node_name}")
# 子智能体更新（非空命名空间）
else:
for node_name, data in chunk["data"].items():
print(f" [{chunk['ns'][0]}] step: {node_name}")python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="openai:gpt-5.5",
system_prompt=(
"You are a project coordinator with no research knowledge. "
"For every user request, you must call the task() tool with "
"subagent_type set to researcher. Never answer research questions yourself. "
"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"\nSubagent complete: {msg.name}")
print(f" Result: {str(msg.content)[:200]}...")
else:
print(f"[main agent] step: {node_name}")
# 子智能体更新（非空命名空间）
else:
for node_name, data in chunk["data"].items():
print(f" [{chunk['ns'][0]}] step: {node_name}")python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
system_prompt=(
"You are a project coordinator with no research knowledge. "
"For every user request, you must call the task() tool with "
"subagent_type set to researcher. Never answer research questions yourself. "
"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"\nSubagent complete: {msg.name}")
print(f" Result: {str(msg.content)[:200]}...")
else:
print(f"[main agent] step: {node_name}")
# 子智能体更新（非空命名空间）
else:
for node_name, data in chunk["data"].items():
print(f" [{chunk['ns'][0]}] step: {node_name}")python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
system_prompt=(
"You are a project coordinator with no research knowledge. "
"For every user request, you must call the task() tool with "
"subagent_type set to researcher. Never answer research questions yourself. "
"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"\nSubagent complete: {msg.name}")
print(f" Result: {str(msg.content)[:200]}...")
else:
print(f"[main agent] step: {node_name}")
# 子智能体更新（非空命名空间）
else:
for node_name, data in chunk["data"].items():
print(f" [{chunk['ns'][0]}] step: {node_name}")python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
system_prompt=(
"You are a project coordinator with no research knowledge. "
"For every user request, you must call the task() tool with "
"subagent_type set to researcher. Never answer research questions yourself. "
"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"\nSubagent complete: {msg.name}")
print(f" Result: {str(msg.content)[:200]}...")
else:
print(f"[main agent] step: {node_name}")
# 子智能体更新（非空命名空间）
else:
for node_name, data in chunk["data"].items():
print(f" [{chunk['ns'][0]}] step: {node_name}")python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
system_prompt=(
"You are a project coordinator with no research knowledge. "
"For every user request, you must call the task() tool with "
"subagent_type set to researcher. Never answer research questions yourself. "
"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"\nSubagent complete: {msg.name}")
print(f" Result: {str(msg.content)[:200]}...")
else:
print(f"[main agent] step: {node_name}")
# 子智能体更新（非空命名空间）
else:
for node_name, data in chunk["data"].items():
print(f" [{chunk['ns'][0]}] step: {node_name}")python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
system_prompt=(
"You are a project coordinator with no research knowledge. "
"For every user request, you must call the task() tool with "
"subagent_type set to researcher. Never answer research questions yourself. "
"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"\nSubagent complete: {msg.name}")
print(f" Result: {str(msg.content)[:200]}...")
else:
print(f"[main agent] step: {node_name}")
# 子智能体更新（非空命名空间）
else:
for node_name, data in chunk["data"].items():
print(f" [{chunk['ns'][0]}] step: {node_name}")bash
[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_requestts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "google-genai:gemini-3.6-flash",
systemPrompt:
"You are a project coordinator with no research knowledge. " +
"For every user request, you must call the task() tool with " +
"subagent_type set to researcher. Never answer research questions yourself. " +
"Keep your final response to one sentence.",
subagents: [
{
name: "researcher",
description: "Researches topics thoroughly",
systemPrompt:
"You are a thorough researcher. Research the given topic " +
"and provide a concise summary in 2-3 sentences.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Write a short summary about AI safety" },
],
},
{ streamMode: "updates", subgraphs: true },
)) {
// 主智能体更新（空命名空间）
if (namespace.length === 0) {
for (const [nodeName, data] of Object.entries(chunk)) {
if (nodeName === "tools") {
// 返回给主智能体的子智能体结果
for (const msg of (data as any).messages ?? []) {
if (msg.type === "tool") {
console.log(`\nSubagent complete: ${msg.name}`);
console.log(` Result: ${String(msg.content).slice(0, 200)}...`);
}
}
} else {
console.log(`[main agent] step: ${nodeName}`);
}
}
}
// 子智能体更新（非空命名空间）
else {
for (const [nodeName] of Object.entries(chunk)) {
console.log(` [${namespace[0]}] step: ${nodeName}`);
}
}
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "openai:gpt-5.5",
systemPrompt:
"You are a project coordinator with no research knowledge. " +
"For every user request, you must call the task() tool with " +
"subagent_type set to researcher. Never answer research questions yourself. " +
"Keep your final response to one sentence.",
subagents: [
{
name: "researcher",
description: "Researches topics thoroughly",
systemPrompt:
"You are a thorough researcher. Research the given topic " +
"and provide a concise summary in 2-3 sentences.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Write a short summary about AI safety" },
],
},
{ streamMode: "updates", subgraphs: true },
)) {
// 主智能体更新（空命名空间）
if (namespace.length === 0) {
for (const [nodeName, data] of Object.entries(chunk)) {
if (nodeName === "tools") {
// 返回给主智能体的子智能体结果
for (const msg of (data as any).messages ?? []) {
if (msg.type === "tool") {
console.log(`\nSubagent complete: ${msg.name}`);
console.log(` Result: ${String(msg.content).slice(0, 200)}...`);
}
}
} else {
console.log(`[main agent] step: ${nodeName}`);
}
}
}
// 子智能体更新（非空命名空间）
else {
for (const [nodeName] of Object.entries(chunk)) {
console.log(` [${namespace[0]}] step: ${nodeName}`);
}
}
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "anthropic:claude-sonnet-4-6",
systemPrompt:
"You are a project coordinator with no research knowledge. " +
"For every user request, you must call the task() tool with " +
"subagent_type set to researcher. Never answer research questions yourself. " +
"Keep your final response to one sentence.",
subagents: [
{
name: "researcher",
description: "Researches topics thoroughly",
systemPrompt:
"You are a thorough researcher. Research the given topic " +
"and provide a concise summary in 2-3 sentences.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Write a short summary about AI safety" },
],
},
{ streamMode: "updates", subgraphs: true },
)) {
// 主智能体更新（空命名空间）
if (namespace.length === 0) {
for (const [nodeName, data] of Object.entries(chunk)) {
if (nodeName === "tools") {
// 返回给主智能体的子智能体结果
for (const msg of (data as any).messages ?? []) {
if (msg.type === "tool") {
console.log(`\nSubagent complete: ${msg.name}`);
console.log(` Result: ${String(msg.content).slice(0, 200)}...`);
}
}
} else {
console.log(`[main agent] step: ${nodeName}`);
}
}
}
// 子智能体更新（非空命名空间）
else {
for (const [nodeName] of Object.entries(chunk)) {
console.log(` [${namespace[0]}] step: ${nodeName}`);
}
}
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "openrouter:openrouter:z-ai/glm-5.2",
systemPrompt:
"You are a project coordinator with no research knowledge. " +
"For every user request, you must call the task() tool with " +
"subagent_type set to researcher. Never answer research questions yourself. " +
"Keep your final response to one sentence.",
subagents: [
{
name: "researcher",
description: "Researches topics thoroughly",
systemPrompt:
"You are a thorough researcher. Research the given topic " +
"and provide a concise summary in 2-3 sentences.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Write a short summary about AI safety" },
],
},
{ streamMode: "updates", subgraphs: true },
)) {
// 主智能体更新（空命名空间）
if (namespace.length === 0) {
for (const [nodeName, data] of Object.entries(chunk)) {
if (nodeName === "tools") {
// 返回给主智能体的子智能体结果
for (const msg of (data as any).messages ?? []) {
if (msg.type === "tool") {
console.log(`\nSubagent complete: ${msg.name}`);
console.log(` Result: ${String(msg.content).slice(0, 200)}...`);
}
}
} else {
console.log(`[main agent] step: ${nodeName}`);
}
}
}
// 子智能体更新（非空命名空间）
else {
for (const [nodeName] of Object.entries(chunk)) {
console.log(` [${namespace[0]}] step: ${nodeName}`);
}
}
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "fireworks:accounts/fireworks/models/glm-5p2",
systemPrompt:
"You are a project coordinator with no research knowledge. " +
"For every user request, you must call the task() tool with " +
"subagent_type set to researcher. Never answer research questions yourself. " +
"Keep your final response to one sentence.",
subagents: [
{
name: "researcher",
description: "Researches topics thoroughly",
systemPrompt:
"You are a thorough researcher. Research the given topic " +
"and provide a concise summary in 2-3 sentences.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Write a short summary about AI safety" },
],
},
{ streamMode: "updates", subgraphs: true },
)) {
// 主智能体更新（空命名空间）
if (namespace.length === 0) {
for (const [nodeName, data] of Object.entries(chunk)) {
if (nodeName === "tools") {
// 返回给主智能体的子智能体结果
for (const msg of (data as any).messages ?? []) {
if (msg.type === "tool") {
console.log(`\nSubagent complete: ${msg.name}`);
console.log(` Result: ${String(msg.content).slice(0, 200)}...`);
}
}
} else {
console.log(`[main agent] step: ${nodeName}`);
}
}
}
// 子智能体更新（非空命名空间）
else {
for (const [nodeName] of Object.entries(chunk)) {
console.log(` [${namespace[0]}] step: ${nodeName}`);
}
}
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "baseten:zai-org/GLM-5.2",
systemPrompt:
"You are a project coordinator with no research knowledge. " +
"For every user request, you must call the task() tool with " +
"subagent_type set to researcher. Never answer research questions yourself. " +
"Keep your final response to one sentence.",
subagents: [
{
name: "researcher",
description: "Researches topics thoroughly",
systemPrompt:
"You are a thorough researcher. Research the given topic " +
"and provide a concise summary in 2-3 sentences.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Write a short summary about AI safety" },
],
},
{ streamMode: "updates", subgraphs: true },
)) {
// 主智能体更新（空命名空间）
if (namespace.length === 0) {
for (const [nodeName, data] of Object.entries(chunk)) {
if (nodeName === "tools") {
// 返回给主智能体的子智能体结果
for (const msg of (data as any).messages ?? []) {
if (msg.type === "tool") {
console.log(`\nSubagent complete: ${msg.name}`);
console.log(` Result: ${String(msg.content).slice(0, 200)}...`);
}
}
} else {
console.log(`[main agent] step: ${nodeName}`);
}
}
}
// 子智能体更新（非空命名空间）
else {
for (const [nodeName] of Object.entries(chunk)) {
console.log(` [${namespace[0]}] step: ${nodeName}`);
}
}
}ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "ollama:north-mini-code-1.0",
systemPrompt:
"You are a project coordinator with no research knowledge. " +
"For every user request, you must call the task() tool with " +
"subagent_type set to researcher. Never answer research questions yourself. " +
"Keep your final response to one sentence.",
subagents: [
{
name: "researcher",
description: "Researches topics thoroughly",
systemPrompt:
"You are a thorough researcher. Research the given topic " +
"and provide a concise summary in 2-3 sentences.",
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Write a short summary about AI safety" },
],
},
{ streamMode: "updates", subgraphs: true },
)) {
// 主智能体更新（空命名空间）
if (namespace.length === 0) {
for (const [nodeName, data] of Object.entries(chunk)) {
if (nodeName === "tools") {
// 返回给主智能体的子智能体结果
for (const msg of (data as any).messages ?? []) {
if (msg.type === "tool") {
console.log(`\nSubagent complete: ${msg.name}`);
console.log(` Result: ${String(msg.content).slice(0, 200)}...`);
}
}
} else {
console.log(`[main agent] step: ${nodeName}`);
}
}
}
// 子智能体更新（非空命名空间）
else {
for (const [nodeName] of Object.entries(chunk)) {
console.log(` [${namespace[0]}] step: ${nodeName}`);
}
}
}bash
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
[tools:call_def456] step: model_request
[tools:call_def456] step: model_request
Subagent complete: task
Result: # Comprehensive Report on AI Safety...
Main agent step: model_requestLLM token
使用 stream_mode="messages" 从主智能体和子智能体两者流式输出单独的 token。每个消息事件都包含标识来源智能体的元数据。
python
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: {subagent_ns}] ---")
current_source = subagent_ns
if token.content:
print(token.content, end="", flush=True)
else:
# 来自主智能体的 token
if "main" != current_source:
print("\n\n--- [main agent] ---")
current_source = "main"
if token.content:
print(token.content, end="", flush=True)
print()ts
let currentSource = "";
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{
role: "user",
content: "Research quantum computing advances",
},
],
},
{ streamMode: "messages", subgraphs: true },
)) {
const [message] = chunk;
// 检查此事件是否来自子智能体（命名空间包含 "tools:"）
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
if (isSubagent) {
// 来自子智能体的 token
const subagentNs = namespace.find((s: string) => s.startsWith("tools:"))!;
if (subagentNs !== currentSource) {
process.stdout.write(`\n\n--- [subagent: ${subagentNs}] ---\n`);
currentSource = subagentNs;
}
if (message.text) {
process.stdout.write(message.text);
}
} else {
// 来自主智能体的 token
if ("main" !== currentSource) {
process.stdout.write(`\n\n--- [main agent] ---\n`);
currentSource = "main";
}
if (message.text) {
process.stdout.write(message.text);
}
}
}
process.stdout.write("\n");工具调用
当子智能体使用工具时，你可以流式输出工具调用事件来显示每个子智能体正在做什么。工具调用分块出现在 messages 流式模式中。
python
from langchain.messages import AIMessageChunk, ToolMessage
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 isinstance(token, AIMessageChunk) and token.tool_call_chunks:
for tc in token.tool_call_chunks:
if tc.get("name"):
print(f"\n[{source}] Tool call: {tc['name']}")
# 参数以分块流式传输——增量写入
if tc.get("args"):
print(tc["args"], end="", flush=True)
# 工具结果
if isinstance(token, ToolMessage):
print(f"\n[{source}] Tool result [{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()ts
import { AIMessageChunk, ToolMessage } from "langchain";
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{
role: "user",
content: "Research recent quantum computing advances",
},
],
},
{ streamMode: "messages", subgraphs: true },
)) {
const [message] = chunk;
// 识别来源："main" 或子智能体命名空间段
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
const source = isSubagent
? namespace.find((s: string) => s.startsWith("tools:"))!
: "main";
// 工具调用分块（流式工具调用）
if (AIMessageChunk.isInstance(message) && message.tool_call_chunks?.length) {
for (const tc of message.tool_call_chunks) {
if (tc.name) {
console.log(`\n[${source}] Tool call: ${tc.name}`);
}
// 参数以分块流式传输——增量写入
if (tc.args) {
process.stdout.write(tc.args);
}
}
}
// 工具结果
if (ToolMessage.isInstance(message)) {
console.log(
`\n[${source}] Tool result [${message.name}]: ${message.text?.slice(0, 150)}`,
);
}
// 常规 AI 内容（跳过工具调用消息）
if (
AIMessageChunk.isInstance(message) &&
message.text &&
!message.tool_call_chunks?.length
) {
process.stdout.write(message.text);
}
}
process.stdout.write("\n");自定义更新
在子智能体工具内部使用 get_stream_writer 来发出自定义进度事件：
在子智能体工具内部使用 config.writer 来发出自定义进度事件：
python
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:
"""Run a data analysis on a given topic.
This tool performs the actual analysis and emits progress updates.
You MUST call this tool for any analysis request.
"""
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.6-flash",
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],
},
],
)
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"])python
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:
"""Run a data analysis on a given topic.
This tool performs the actual analysis and emits progress updates.
You MUST call this tool for any analysis request.
"""
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="openai:gpt-5.5",
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],
},
],
)
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"])python
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:
"""Run a data analysis on a given topic.
This tool performs the actual analysis and emits progress updates.
You MUST call this tool for any analysis request.
"""
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="anthropic:claude-sonnet-4-6",
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],
},
],
)
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"])python
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:
"""Run a data analysis on a given topic.
This tool performs the actual analysis and emits progress updates.
You MUST call this tool for any analysis request.
"""
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="openrouter:z-ai/glm-5.2",
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],
},
],
)
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"])python
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:
"""Run a data analysis on a given topic.
This tool performs the actual analysis and emits progress updates.
You MUST call this tool for any analysis request.
"""
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="fireworks:accounts/fireworks/models/glm-5p2",
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],
},
],
)
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"])python
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:
"""Run a data analysis on a given topic.
This tool performs the actual analysis and emits progress updates.
You MUST call this tool for any analysis request.
"""
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="baseten:zai-org/GLM-5.2",
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],
},
],
)
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"])python
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:
"""Run a data analysis on a given topic.
This tool performs the actual analysis and emits progress updates.
You MUST call this tool for any analysis request.
"""
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="ollama:north-mini-code-1.0",
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],
},
],
)
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"])bash
[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}ts
import { createDeepAgent } from "deepagents";
import { tool, type ToolRuntime } from "langchain";
import { z } from "zod";
/**
* 一个通过 config.writer 发出自定义进度事件的工具。
* writer 会将数据发送到 "custom" 流式模式。
*/
const analyzeData = tool(
async ({ topic }: { topic: string }, config: ToolRuntime) => {
const writer = config.writer;
writer?.({ status: "starting", topic, progress: 0 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "analyzing", progress: 50 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "complete", progress: 100 });
return `Analysis of "${topic}": Customer sentiment is 85% positive, driven by product quality and support response times.`;
},
{
name: "analyze_data",
description:
"Run a data analysis on a given topic. " +
"This tool performs the actual analysis and emits progress updates. " +
"You MUST call this tool for any analysis request.",
schema: z.object({
topic: z.string().describe("The topic or subject to analyze"),
}),
},
);
const agent = createDeepAgent({
model: "google-genai:gemini-3.6-flash",
systemPrompt:
"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",
systemPrompt:
"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: [analyzeData],
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{
role: "user",
content: "Analyze customer satisfaction trends",
},
],
},
{ streamMode: "custom", subgraphs: true },
)) {
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
if (isSubagent) {
const subagentNs = namespace.find((s: string) => s.startsWith("tools:"))!;
console.log(`[${subagentNs}]`, chunk);
} else {
console.log("[main]", chunk);
}
}ts
import { createDeepAgent } from "deepagents";
import { tool, type ToolRuntime } from "langchain";
import { z } from "zod";
/**
* 一个通过 config.writer 发出自定义进度事件的工具。
* writer 会将数据发送到 "custom" 流式模式。
*/
const analyzeData = tool(
async ({ topic }: { topic: string }, config: ToolRuntime) => {
const writer = config.writer;
writer?.({ status: "starting", topic, progress: 0 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "analyzing", progress: 50 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "complete", progress: 100 });
return `Analysis of "${topic}": Customer sentiment is 85% positive, driven by product quality and support response times.`;
},
{
name: "analyze_data",
description:
"Run a data analysis on a given topic. " +
"This tool performs the actual analysis and emits progress updates. " +
"You MUST call this tool for any analysis request.",
schema: z.object({
topic: z.string().describe("The topic or subject to analyze"),
}),
},
);
const agent = createDeepAgent({
model: "openai:gpt-5.5",
systemPrompt:
"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",
systemPrompt:
"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: [analyzeData],
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{
role: "user",
content: "Analyze customer satisfaction trends",
},
],
},
{ streamMode: "custom", subgraphs: true },
)) {
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
if (isSubagent) {
const subagentNs = namespace.find((s: string) => s.startsWith("tools:"))!;
console.log(`[${subagentNs}]`, chunk);
} else {
console.log("[main]", chunk);
}
}ts
import { createDeepAgent } from "deepagents";
import { tool, type ToolRuntime } from "langchain";
import { z } from "zod";
/**
* 一个通过 config.writer 发出自定义进度事件的工具。
* writer 会将数据发送到 "custom" 流式模式。
*/
const analyzeData = tool(
async ({ topic }: { topic: string }, config: ToolRuntime) => {
const writer = config.writer;
writer?.({ status: "starting", topic, progress: 0 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "analyzing", progress: 50 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "complete", progress: 100 });
return `Analysis of "${topic}": Customer sentiment is 85% positive, driven by product quality and support response times.`;
},
{
name: "analyze_data",
description:
"Run a data analysis on a given topic. " +
"This tool performs the actual analysis and emits progress updates. " +
"You MUST call this tool for any analysis request.",
schema: z.object({
topic: z.string().describe("The topic or subject to analyze"),
}),
},
);
const agent = createDeepAgent({
model: "anthropic:claude-sonnet-4-6",
systemPrompt:
"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",
systemPrompt:
"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: [analyzeData],
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{
role: "user",
content: "Analyze customer satisfaction trends",
},
],
},
{ streamMode: "custom", subgraphs: true },
)) {
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
if (isSubagent) {
const subagentNs = namespace.find((s: string) => s.startsWith("tools:"))!;
console.log(`[${subagentNs}]`, chunk);
} else {
console.log("[main]", chunk);
}
}ts
import { createDeepAgent } from "deepagents";
import { tool, type ToolRuntime } from "langchain";
import { z } from "zod";
/**
* 一个通过 config.writer 发出自定义进度事件的工具。
* writer 会将数据发送到 "custom" 流式模式。
*/
const analyzeData = tool(
async ({ topic }: { topic: string }, config: ToolRuntime) => {
const writer = config.writer;
writer?.({ status: "starting", topic, progress: 0 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "analyzing", progress: 50 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "complete", progress: 100 });
return `Analysis of "${topic}": Customer sentiment is 85% positive, driven by product quality and support response times.`;
},
{
name: "analyze_data",
description:
"Run a data analysis on a given topic. " +
"This tool performs the actual analysis and emits progress updates. " +
"You MUST call this tool for any analysis request.",
schema: z.object({
topic: z.string().describe("The topic or subject to analyze"),
}),
},
);
const agent = createDeepAgent({
model: "openrouter:openrouter:z-ai/glm-5.2",
systemPrompt:
"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",
systemPrompt:
"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: [analyzeData],
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{
role: "user",
content: "Analyze customer satisfaction trends",
},
],
},
{ streamMode: "custom", subgraphs: true },
)) {
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
if (isSubagent) {
const subagentNs = namespace.find((s: string) => s.startsWith("tools:"))!;
console.log(`[${subagentNs}]`, chunk);
} else {
console.log("[main]", chunk);
}
}ts
import { createDeepAgent } from "deepagents";
import { tool, type ToolRuntime } from "langchain";
import { z } from "zod";
/**
* 一个通过 config.writer 发出自定义进度事件的工具。
* writer 会将数据发送到 "custom" 流式模式。
*/
const analyzeData = tool(
async ({ topic }: { topic: string }, config: ToolRuntime) => {
const writer = config.writer;
writer?.({ status: "starting", topic, progress: 0 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "analyzing", progress: 50 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "complete", progress: 100 });
return `Analysis of "${topic}": Customer sentiment is 85% positive, driven by product quality and support response times.`;
},
{
name: "analyze_data",
description:
"Run a data analysis on a given topic. " +
"This tool performs the actual analysis and emits progress updates. " +
"You MUST call this tool for any analysis request.",
schema: z.object({
topic: z.string().describe("The topic or subject to analyze"),
}),
},
);
const agent = createDeepAgent({
model: "fireworks:accounts/fireworks/models/glm-5p2",
systemPrompt:
"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",
systemPrompt:
"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: [analyzeData],
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{
role: "user",
content: "Analyze customer satisfaction trends",
},
],
},
{ streamMode: "custom", subgraphs: true },
)) {
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
if (isSubagent) {
const subagentNs = namespace.find((s: string) => s.startsWith("tools:"))!;
console.log(`[${subagentNs}]`, chunk);
} else {
console.log("[main]", chunk);
}
}ts
import { createDeepAgent } from "deepagents";
import { tool, type ToolRuntime } from "langchain";
import { z } from "zod";
/**
* 一个通过 config.writer 发出自定义进度事件的工具。
* writer 会将数据发送到 "custom" 流式模式。
*/
const analyzeData = tool(
async ({ topic }: { topic: string }, config: ToolRuntime) => {
const writer = config.writer;
writer?.({ status: "starting", topic, progress: 0 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "analyzing", progress: 50 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "complete", progress: 100 });
return `Analysis of "${topic}": Customer sentiment is 85% positive, driven by product quality and support response times.`;
},
{
name: "analyze_data",
description:
"Run a data analysis on a given topic. " +
"This tool performs the actual analysis and emits progress updates. " +
"You MUST call this tool for any analysis request.",
schema: z.object({
topic: z.string().describe("The topic or subject to analyze"),
}),
},
);
const agent = createDeepAgent({
model: "baseten:zai-org/GLM-5.2",
systemPrompt:
"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",
systemPrompt:
"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: [analyzeData],
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{
role: "user",
content: "Analyze customer satisfaction trends",
},
],
},
{ streamMode: "custom", subgraphs: true },
)) {
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
if (isSubagent) {
const subagentNs = namespace.find((s: string) => s.startsWith("tools:"))!;
console.log(`[${subagentNs}]`, chunk);
} else {
console.log("[main]", chunk);
}
}ts
import { createDeepAgent } from "deepagents";
import { tool, type ToolRuntime } from "langchain";
import { z } from "zod";
/**
* 一个通过 config.writer 发出自定义进度事件的工具。
* writer 会将数据发送到 "custom" 流式模式。
*/
const analyzeData = tool(
async ({ topic }: { topic: string }, config: ToolRuntime) => {
const writer = config.writer;
writer?.({ status: "starting", topic, progress: 0 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "analyzing", progress: 50 });
await new Promise((r) => setTimeout(r, 500));
writer?.({ status: "complete", progress: 100 });
return `Analysis of "${topic}": Customer sentiment is 85% positive, driven by product quality and support response times.`;
},
{
name: "analyze_data",
description:
"Run a data analysis on a given topic. " +
"This tool performs the actual analysis and emits progress updates. " +
"You MUST call this tool for any analysis request.",
schema: z.object({
topic: z.string().describe("The topic or subject to analyze"),
}),
},
);
const agent = createDeepAgent({
model: "ollama:north-mini-code-1.0",
systemPrompt:
"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",
systemPrompt:
"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: [analyzeData],
},
],
});
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{
role: "user",
content: "Analyze customer satisfaction trends",
},
],
},
{ streamMode: "custom", subgraphs: true },
)) {
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
if (isSubagent) {
const subagentNs = namespace.find((s: string) => s.startsWith("tools:"))!;
console.log(`[${subagentNs}]`, chunk);
} else {
console.log("[main]", chunk);
}
}bash
[tools:call_abc123] { status: 'fetching', progress: 0 }
[tools:call_abc123] { status: 'analyzing', progress: 50 }
[tools:call_abc123] { status: 'complete', progress: 100 }流式输出多种模式
组合多种流式模式以全面了解智能体执行情况：
python
# 跳过内部中间件步骤——只显示有意义的节点名
INTERESTING_NODES = {"model", "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}] step: {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}] custom event:", chunk["data"])
print()ts
// 跳过内部中间件步骤——只显示有意义的节点名
const INTERESTING_NODES = new Set(["model", "tools"]);
let lastSource = "";
let midLine = false; // 当我们已写入 token 但还没有换行时为 true
for await (const [namespace, mode, data] of await agent.stream(
{
messages: [
{
role: "user",
content: "Analyze the impact of remote work on team productivity",
},
],
},
{ streamMode: ["updates", "messages", "custom"], subgraphs: true },
)) {
const isSubagent = namespace.some((s: string) => s.startsWith("tools:"));
const source = isSubagent ? "subagent" : "main";
if (mode === "updates") {
for (const nodeName of Object.keys(data)) {
if (!INTERESTING_NODES.has(nodeName)) continue;
if (midLine) {
process.stdout.write("\n");
midLine = false;
}
console.log(`[${source}] step: ${nodeName}`);
}
} else if (mode === "messages") {
const [message] = data;
if (message.text) {
// 当来源变化时打印一个标题
if (source !== lastSource) {
if (midLine) {
process.stdout.write("\n");
midLine = false;
}
process.stdout.write(`\n[${source}] `);
lastSource = source;
}
process.stdout.write(message.text);
midLine = true;
}
} else if (mode === "custom") {
if (midLine) {
process.stdout.write("\n");
midLine = false;
}
console.log(`[${source}] custom event:`, data);
}
}
process.stdout.write("\n");常见模式
跟踪子智能体生命周期
监控子智能体何时启动、运行和完成：
python
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 节点包含 task 工具调用时，
# 说明已生成一个子智能体。
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] PENDING → subagent "{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]
# 检查是否有待处理（pending）的子智能体需要标记为运行中。
# 注意：pregel 任务 ID 与 tool_call_id 不同，
# 因此我们会在第一个子智能体事件时将待处理子智能体标记为运行中。
for sub_id, sub in active_subagents.items():
if sub["status"] == "pending":
sub["status"] = "running"
print(
f'[lifecycle] RUNNING → subagent "{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 → subagent "{sub["type"]}" '
f"({msg.tool_call_id})"
)
print(f" Result preview: {str(msg.content)[:120]}...")
# 打印最终状态
print("\n--- Final subagent states ---")
for sub_id, sub in active_subagents.items():
print(f" {sub['type']}: {sub['status']}")ts
function getToolCalls(message: unknown): Array<{
id?: string;
name?: string;
args?: Record<string, unknown>;
}> {
if (!message || typeof message !== "object") {
return [];
}
const record = message as Record<string, unknown>;
const toolCalls = record.tool_calls ?? record.toolCalls;
return Array.isArray(toolCalls)
? (toolCalls as Array<{
id?: string;
name?: string;
args?: Record<string, unknown>;
}>)
: [];
}
const activeSubagents = new Map<
string,
{ type?: string; description?: string; status: string }
>();
for await (const [namespace, chunk] of await agent.stream(
{
messages: [
{ role: "user", content: "Research the latest AI safety developments" },
],
},
{ streamMode: "updates", subgraphs: true },
)) {
for (const [nodeName, data] of Object.entries(chunk)) {
// ─── 阶段 1：检测子智能体启动 ────────────────────────
// 当主智能体发出 task 工具调用时，已生成一个子智能体。
if (namespace.length === 0) {
for (const msg of (data as { messages?: unknown[] }).messages ?? []) {
for (const tc of getToolCalls(msg)) {
if (tc.name === "task" && tc.id) {
activeSubagents.set(tc.id, {
type: tc.args?.subagent_type as string | undefined,
description: String(tc.args?.description ?? "").slice(0, 80),
status: "pending",
});
console.log(
`[lifecycle] PENDING → subagent "${tc.args?.subagent_type}" (${tc.id})`,
);
}
}
}
}
// ─── 阶段 2：检测子智能体运行 ─────────────────────────
// 当我们从 tools:UUID 命名空间收到事件时，
// 说明该子智能体正在执行。
if (namespace.length > 0 && namespace[0].startsWith("tools:")) {
const pregelId = namespace[0].split(":")[1];
// 检查是否有待处理（pending）的子智能体需要标记为运行中。
// 注意：pregel 任务 ID 与 tool_call_id 不同，
// 因此我们会在第一个子智能体事件时将待处理子智能体标记为运行中。
let markedRunning = false;
for (const [, sub] of activeSubagents) {
if (sub.status === "pending") {
sub.status = "running";
markedRunning = true;
console.log(
`[lifecycle] RUNNING → subagent "${sub.type}" (pregel: ${pregelId})`,
);
break;
}
}
if (!markedRunning && activeSubagents.size === 0) {
activeSubagents.set(pregelId, {
type: "researcher",
status: "running",
});
console.log(
`[lifecycle] RUNNING → subagent "researcher" (pregel: ${pregelId})`,
);
}
}
// ─── 阶段 3：检测子智能体完成 ──────────────────────
// 当主智能体的 tools 节点返回工具消息时，
// 说明子智能体已完成并返回其结果。
if (namespace.length === 0 && nodeName === "tools") {
for (const msg of (data as { messages?: Array<Record<string, unknown>> })
.messages ?? []) {
if (msg.type === "tool") {
const toolCallId = String(msg.tool_call_id ?? msg.toolCallId ?? "");
const subagent = activeSubagents.get(toolCallId);
if (subagent) {
subagent.status = "complete";
console.log(
`[lifecycle] COMPLETE → subagent "${subagent.type}" (${toolCallId})`,
);
console.log(
` Result preview: ${String(msg.content).slice(0, 120)}...`,
);
}
}
}
}
}
}
// 打印最终状态
console.log("\n--- Final subagent states ---");
for (const [id, sub] of activeSubagents) {
console.log(` ${sub.type}: ${sub.status}`);
}v2 流式格式
INFO
需要 LangGraph >= 1.1。
本页的所有示例都使用 v2 流式格式（version="v2"），这是推荐的做法。每个分块都是一个带有 type、ns 和 data 键的 StreamPart 字典——无论流式模式、模式数量或子图设置如何，其形状都相同。
v2 格式消除了嵌套元组解包，使在 Deep Agents 中处理子图流式输出变得简单直接。比较这两种格式：
python
# 统一的格式——无需嵌套元组解包
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
# 必须处理 (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) # 负载数据有关 v2 格式的更多详细信息（包括类型收窄以及 Pydantic/dataclass 强制转换），请参阅 LangGraph 流式输出文档。
相关
- 子智能体——配置并在 Deep Agents 中使用子智能体
- 前端流式输出——为 Deep Agents 使用
useStream构建 React UI - LangChain 事件流——LangChain 智能体的一般流式概念