外观
异步子智能体让监督者智能体启动立即返回的后台任务,因此监督者可以在子智能体并发工作的同时继续与用户交互。监督者可以随时检查进度、发送后续指令或取消任务。
这是建立在子智能体之上的,后者同步运行并阻塞监督者直到完成。当任务长时间运行、可并行化或需要中途控制时,请使用异步子智能体。
INFO
异步子智能体是 deepagents 0.5.0 中提供的一项预览功能。预览功能正在积极开发中,API 可能会变化。
异步子智能体是 deepagents 1.9.0 中提供的一项预览功能。预览功能正在积极开发中,API 可能会变化。
INFO
异步子智能体与任何实现 Agent Protocol 的服务器通信。你可以使用 LangSmith Deployments,或自托管任何兼容 Agent Protocol 的服务器。每个子智能体独立于监督者运行,监督者通过 SDK 控制它们进行启动、检查、更新和取消。
何时使用异步子智能体
| 维度 | 同步子智能体 | 异步子智能体 |
|---|---|---|
| 执行模型 | 监督者阻塞直到子智能体完成 | 立即返回作业 ID;监督者继续 |
| 并发 | 并行但阻塞 | 并行且非阻塞 |
| 任务中途更新 | 不可能 | 通过 update_async_task 发送后续指令 |
| 取消 | 不可能 | 通过 cancel_async_task 取消运行中的任务 |
| 有状态性 | 无状态——调用之间没有持久状态 | 有状态——在其自身的线程上跨交互保持状态 |
| 最适合 | 智能体应在继续之前等待结果的任务 | 在聊天中以交互方式管理的长时间运行、复杂任务 |
配置异步子智能体
将异步子智能体定义为 AsyncSubAgent 规范列表,每个规范指向一个 Agent Protocol 服务器:
python
from deepagents import AsyncSubAgent, create_deep_agent
async_subagents = [
AsyncSubAgent(
name="researcher",
description="Research agent for information gathering and synthesis",
graph_id="researcher",
# 无 url → ASGI 传输(在同一部署中协同部署)
),
AsyncSubAgent(
name="coder",
description="Coding agent for code generation and review",
graph_id="coder",
# url="https://coder-deployment.langsmith.dev" # 可选:用于远程的 HTTP 传输
),
]
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
subagents=async_subagents,
)| 字段 | 类型 | 描述 |
|---|---|---|
name | str | 必需。唯一标识符。监督者启动任务时使用它。 |
description | str | 必需。该子智能体做什么。监督者使用它来决定委派给哪个智能体。 |
graph_id | str | 必需。Agent Protocol 服务器上的图 ID(或 assistant ID)。对于基于 LangGraph 的部署,它必须匹配在 langgraph.json 中注册的图。 |
url | str | 可选。省略时使用 ASGI 传输(进程内)。设置时使用 HTTP 传输连接远程 Agent Protocol 服务器。 |
headers | dict[str, str] | 可选。发送到远程服务器的请求的附加标头。用于与自托管 Agent Protocol 服务器进行自定义身份验证。 |
ts
import { createDeepAgent, type AsyncSubAgent } from "deepagents";
const asyncSubagents: AsyncSubAgent[] = [
{
name: "researcher",
description: "Research agent for information gathering and synthesis",
graphId: "researcher",
// 无 url → ASGI 传输(在同一部署中协同部署)
},
{
name: "coder",
description: "Coding agent for code generation and review",
graphId: "coder",
// url: "https://coder-deployment.langsmith.dev" // 可选:用于远程的 HTTP 传输
},
];
const agent = createDeepAgent({
model: "google-genai:gemini-3.6-flash",
subagents: [...asyncSubagents],
});ts
import { createDeepAgent, type AsyncSubAgent } from "deepagents";
const asyncSubagents: AsyncSubAgent[] = [
{
name: "researcher",
description: "Research agent for information gathering and synthesis",
graphId: "researcher",
// 无 url → ASGI 传输(在同一部署中协同部署)
},
{
name: "coder",
description: "Coding agent for code generation and review",
graphId: "coder",
// url: "https://coder-deployment.langsmith.dev" // 可选:用于远程的 HTTP 传输
},
];
const agent = createDeepAgent({
model: "openai:gpt-5.5",
subagents: [...asyncSubagents],
});ts
import { createDeepAgent, type AsyncSubAgent } from "deepagents";
const asyncSubagents: AsyncSubAgent[] = [
{
name: "researcher",
description: "Research agent for information gathering and synthesis",
graphId: "researcher",
// 无 url → ASGI 传输(在同一部署中协同部署)
},
{
name: "coder",
description: "Coding agent for code generation and review",
graphId: "coder",
// url: "https://coder-deployment.langsmith.dev" // 可选:用于远程的 HTTP 传输
},
];
const agent = createDeepAgent({
model: "anthropic:claude-sonnet-4-6",
subagents: [...asyncSubagents],
});ts
import { createDeepAgent, type AsyncSubAgent } from "deepagents";
const asyncSubagents: AsyncSubAgent[] = [
{
name: "researcher",
description: "Research agent for information gathering and synthesis",
graphId: "researcher",
// 无 url → ASGI 传输(在同一部署中协同部署)
},
{
name: "coder",
description: "Coding agent for code generation and review",
graphId: "coder",
// url: "https://coder-deployment.langsmith.dev" // 可选:用于远程的 HTTP 传输
},
];
const agent = createDeepAgent({
model: "openrouter:openrouter:z-ai/glm-5.2",
subagents: [...asyncSubagents],
});ts
import { createDeepAgent, type AsyncSubAgent } from "deepagents";
const asyncSubagents: AsyncSubAgent[] = [
{
name: "researcher",
description: "Research agent for information gathering and synthesis",
graphId: "researcher",
// 无 url → ASGI 传输(在同一部署中协同部署)
},
{
name: "coder",
description: "Coding agent for code generation and review",
graphId: "coder",
// url: "https://coder-deployment.langsmith.dev" // 可选:用于远程的 HTTP 传输
},
];
const agent = createDeepAgent({
model: "fireworks:accounts/fireworks/models/glm-5p2",
subagents: [...asyncSubagents],
});ts
import { createDeepAgent, type AsyncSubAgent } from "deepagents";
const asyncSubagents: AsyncSubAgent[] = [
{
name: "researcher",
description: "Research agent for information gathering and synthesis",
graphId: "researcher",
// 无 url → ASGI 传输(在同一部署中协同部署)
},
{
name: "coder",
description: "Coding agent for code generation and review",
graphId: "coder",
// url: "https://coder-deployment.langsmith.dev" // 可选:用于远程的 HTTP 传输
},
];
const agent = createDeepAgent({
model: "baseten:zai-org/GLM-5.2",
subagents: [...asyncSubagents],
});ts
import { createDeepAgent, type AsyncSubAgent } from "deepagents";
const asyncSubagents: AsyncSubAgent[] = [
{
name: "researcher",
description: "Research agent for information gathering and synthesis",
graphId: "researcher",
// 无 url → ASGI 传输(在同一部署中协同部署)
},
{
name: "coder",
description: "Coding agent for code generation and review",
graphId: "coder",
// url: "https://coder-deployment.langsmith.dev" // 可选:用于远程的 HTTP 传输
},
];
const agent = createDeepAgent({
model: "ollama:north-mini-code-1.0",
subagents: [...asyncSubagents],
});| 字段 | 类型 | 描述 |
|---|---|---|
name | string | 必需。唯一标识符。监督者启动任务时使用它。 |
description | string | 必需。该子智能体做什么。监督者使用它来决定委派给哪个智能体。 |
graphId | string | 必需。Agent Protocol 服务器上的图 ID(或 assistant ID)。对于基于 LangGraph 的部署,它必须匹配在 langgraph.json 中注册的图。 |
url | string | 可选。省略时使用 ASGI 传输(进程内)。设置时使用 HTTP 传输连接远程 Agent Protocol 服务器。 |
headers | Record<string, string> | 可选。发送到远程服务器的请求的附加标头。用于与自托管 Agent Protocol 服务器进行自定义身份验证。 |
对于基于 LangGraph 的部署,请在同一个 langgraph.json 中注册所有图以进行协同部署:
json
{
"graphs": {
"supervisor": "./src/supervisor.py:graph",
"researcher": "./src/researcher.py:graph",
"coder": "./src/coder.py:graph"
}
}使用异步子智能体工具
当配置了异步子智能体时,AsyncSubAgentMiddleware 会包含在默认中间件栈中,它给监督者提供五个工具:
| 工具 | 用途 | 返回 |
|---|---|---|
start_async_task | 启动一个新的后台任务 | 任务 ID(立即) |
check_async_task | 获取任务的当前状态和结果 | 状态 + 结果(如果已完成) |
update_async_task | 向运行中的任务发送新指令 | 确认 + 更新后的状态 |
cancel_async_task | 停止运行中的任务 | 确认 |
list_async_tasks | 列出所有被跟踪的任务及实时状态 | 所有任务的摘要 |
监督者的 LLM 像调用任何其他工具一样调用这些工具。中间件自动处理线程创建、运行管理和状态持久化。
理解生命周期
一次典型交互遵循以下顺序:
- 启动(Launch) 在服务器上创建新线程,以任务描述作为输入开始一次运行,并将线程 ID 作为任务 ID 返回。监督者向用户报告此 ID,并且不轮询完成状态。
- 检查(Check) 获取当前运行状态。如果运行成功,它检索线程状态以提取子智能体的最终输出。如果仍在运行,则向用户报告。
- 更新(Update) 使用中断多重任务策略在同一个线程上创建一次新运行。之前的运行被中断,子智能体使用完整对话历史加上新指令重新开始。任务 ID 保持不变。
- 取消(Cancel) 在服务器上调用
runs.cancel(),并将任务标记为"cancelled"。 - 列出(List) 遍历所有被跟踪的任务。对于非终态任务,它并行地从服务器获取实时状态。终态(
success、error、cancelled)从缓存返回。
理解状态管理
任务元数据存储在监督者图上的专用状态通道(async_tasks)中,与消息历史分开。这一点很关键,因为深度智能体会在上下文窗口填满时压缩其消息历史。如果任务 ID 只存在于工具消息中,它们会在压缩期间丢失。专用通道确保监督者即使经过多轮摘要也能始终通过 list_async_tasks 回忆起其任务。
每个被跟踪的任务记录任务 ID、智能体名称、线程 ID、运行 ID、状态和时间戳(created_at、last_checked_at、last_updated_at)。
任务元数据存储在监督者图上的专用状态通道(asyncTasks)中,与消息历史分开。这一点很关键,因为深度智能体会在上下文窗口填满时压缩其消息历史。如果任务 ID 只存在于工具消息中,它们会在压缩期间丢失。专用通道确保监督者即使经过多轮摘要也能始终通过 list_async_tasks 回忆起其任务。
每个被跟踪的任务记录任务 ID、智能体名称、线程 ID、运行 ID、状态和时间戳(createdAt、checkedAt、updatedAt)。
选择传输方式
ASGI 传输(协同部署)
当子智能体规范省略 url 字段时,LangGraph SDK 使用 ASGI 传输——SDK 调用通过进程内函数调用而非 HTTP 路由。对于基于 LangGraph 的部署,这要求两个图都在同一个 langgraph.json 中注册。
ASGI 传输消除了网络延迟,且不需要额外的身份验证配置。子智能体仍然作为具有自己状态的独立线程运行。这是推荐的默认方式。
HTTP 传输(远程)
添加 url 字段以切换到 HTTP 传输,此时 SDK 调用通过网络传输到远程 Agent Protocol 服务器:
python
from deepagents import AsyncSubAgent
AsyncSubAgent(
name="researcher",
description="Research agent",
graph_id="researcher",
url="https://my-research-deployment.langsmith.dev",
)typescript
{
name: "researcher",
description: "Research agent",
graphId: "researcher",
url: "https://my-research-deployment.langsmith.dev",
}对于 LangGraph 部署,身份验证由 LangGraph SDK 使用环境变量 LANGSMITH_API_KEY(或 LANGGRAPH_API_KEY)处理。自托管 Agent Protocol 服务器可能使用不同的身份验证机制。
当子智能体需要独立扩展、不同的资源配置档案,或由不同团队维护时,请使用 HTTP 传输。
选择部署拓扑
单一部署
单一部署意味着所有智能体通过 ASGI 传输协同部署在同一台服务器上。对于基于 LangGraph 的部署,在一个 langgraph.json 中注册所有图。这是推荐的起点——管理一台服务器,智能体之间零网络延迟。
拆分部署
监督者在一台服务器上,子智能体在另一台服务器上并通过 HTTP 传输。当子智能体需要不同的计算配置档案或独立扩展时使用。
混合
在混合部署中,一些子智能体通过 ASGI 协同部署,另一些通过 HTTP 远程部署:
python
from deepagents import AsyncSubAgent
async_subagents = [
AsyncSubAgent(
name="researcher",
description="Research agent",
graph_id="researcher",
# 无 url → ASGI(协同部署)
),
AsyncSubAgent(
name="coder",
description="Coding agent",
graph_id="coder",
url="https://coder-deployment.langsmith.dev",
# 有 url → HTTP(远程)
),
]ts
import type { AsyncSubAgent } from "deepagents";
const asyncSubagents: AsyncSubAgent[] = [
{
name: "researcher",
description: "Research agent",
graphId: "researcher",
// 无 url → ASGI(协同部署)
},
{
name: "coder",
description: "Coding agent",
graphId: "coder",
url: "https://coder-deployment.langsmith.dev",
// 有 url → HTTP(远程)
},
];最佳实践
为本地开发调整工作线程池大小
使用 langgraph dev 在本地运行时,请增加工作线程池以容纳并发的子智能体运行。每个活动运行占用一个工作线程槽位。一个有 3 个并发子智能体任务的监督者需要 4 个槽位(1 个监督者 + 3 个子智能体)。资源配置不足会导致启动排队。
bash
langgraph dev --n-jobs-per-worker 10编写清晰的子智能体描述
监督者使用描述来决定启动哪个子智能体。要具体且以行动为导向:
python
from deepagents import AsyncSubAgent
AsyncSubAgent(
name="researcher",
description="Conducts in-depth research using web search. Use for questions requiring multiple searches and synthesis.",
graph_id="researcher",
)python
from deepagents import AsyncSubAgent
AsyncSubAgent(
name="helper",
description="helps with stuff",
graph_id="helper",
)typescript
// 好的示例
{
name: "researcher",
description: "Conducts in-depth research using web search. Use for questions requiring multiple searches and synthesis.",
graphId: "researcher",
}typescript
// 差的示例
{
name: "helper",
description: "helps with stuff",
graphId: "helper",
}使用线程 ID 进行追踪
使用基于 LangGraph 的部署时,每次异步子智能体运行都是一次标准 LangGraph 运行,在 LangSmith 中完全可见。监督者的追踪显示 launch、check、update、cancel 和 list 的工具调用。每个子智能体运行都作为单独追踪出现,并通过线程 ID 关联。使用线程 ID(任务 ID)将监督者编排追踪与子智能体执行追踪关联起来。
故障排查
监督者在启动后立即轮询
问题:监督者在启动后立即循环调用 check,将异步执行变成阻塞。
解决方案:中间件会注入系统提示词规则来防止这种情况。如果轮询仍然存在,请在监督者的系统提示词中强化该行为:
python
from deepagents import create_deep_agent
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
system_prompt="""...your instructions...
After launching an async subagent, ALWAYS return control to the user.
Never call check_async_task immediately after launch.""",
subagents=async_subagents,
)ts
import { createDeepAgent } from "deepagents";
const agent = createDeepAgent({
model: "google_genai:gemini-3.6-flash",
systemPrompt: `...your instructions...
After launching an async subagent, ALWAYS return control to the user.
Never call check_async_task immediately after launch.`,
subagents: [...asyncSubagents],
});监督者报告过期的状态
问题:监督者引用对话历史中较早的任务状态,而不是进行新的 check 调用。
解决方案:中间件提示词指示模型“对话历史中的任务状态始终是过期的”。如果仍然发生这种情况,请添加明确指令,在报告状态之前始终调用 check 或 list。
任务 ID 查找失败
问题:监督者截断或重新格式化任务 ID,导致 check 或 cancel 失败。
解决方案:中间件提示词指示模型始终使用完整的任务 ID。如果截断仍然存在,这通常是模型特定问题——尝试不同的模型,或在系统提示词中添加“始终显示完整的 task_id,绝不截断或缩写它”。
子智能体启动排队而不是运行
问题:启动子智能体挂起或需要很长时间才能开始。
解决方案:工作线程池可能已耗尽。使用 --n-jobs-per-worker 增加池大小。请参阅为本地开发调整工作线程池大小。
参考实现
async-deep-agents 仓库包含 Python 和 TypeScript 两种语言的工作示例,可部署到 LangSmith Deployments。它演示了一个监督者,其研究员和编码员子智能体作为后台任务运行。