Sub-Agent 是在独立 Trace 中运行的专门化 Agent,用于处理特定类型的子任务。
| 名称 | 用途 | 允许工具 | 典型场景 |
|---|---|---|---|
general |
通用多步骤任务 | 所有工具(除 task) | 数据收集、复杂分析 |
explore |
代码探索 | read, search, list | 查找代码、理解结构 |
analyst |
深度分析 | read, search, web | 技术栈分析、报告 |
from agent import AgentRunner, AgentConfig
# 主 Agent 会自动知道如何使用 Sub-Agent
runner = AgentRunner(llm_call=your_llm, config=AgentConfig())
async for event in runner.run(
task="使用 explore sub-agent 分析这个项目的架构"
):
# 处理事件
pass
from agent.models.agent import AgentDefinition
from agent.agent_registry import get_agent_registry
# 定义
custom_agent = AgentDefinition(
name="code-reviewer",
description="代码审查专家",
mode="subagent",
allowed_tools=["read_file", "search_code"],
system_prompt="你是代码审查专家...",
)
# 注册
get_agent_registry().register(custom_agent)
创建 sub_agents.json:
{
"agents": {
"my-agent": {
"description": "我的自定义 Agent",
"mode": "subagent",
"allowed_tools": ["read_file", "search_code"],
"system_prompt": "..."
}
}
}
加载配置:
get_agent_registry().load_from_config("sub_agents.json")
primary: 主 Agent,可以被用户直接调用,可以启动 Sub-Agentsubagent: Sub-Agent,只能被其他 Agent 通过 Task 工具调用all: 两者皆可AgentDefinition(
allowed_tools=["read_file", "search_code"], # 白名单
denied_tools=["write_file", "execute_bash"], # 黑名单
)
AgentDefinition(
permissions={
"paths": {
"/etc": "deny",
"~/.ssh": "deny",
"/tmp": "allow",
}
}
)
AgentDefinition(
can_spawn_subagent=False, # 默认为 False
denied_tools=["task"], # 显式禁止 Task 工具
)
async for event in runner.run(task="..."):
if event.type == "tool_call" and event.data["tool"] == "task":
# Sub-Agent 启动
print(f"启动: {event.data['args']['subagent_type']}")
elif event.type == "tool_result" and event.data["tool"] == "task":
# Sub-Agent 完成
metadata = event.data["metadata"]
print(f"完成: {metadata['sub_trace_id']}")
from agent.storage import TraceStore
# 获取主 Trace
main_trace = await trace_store.get_trace(trace_id)
# 查询所有子 Trace
sub_traces = await trace_store.query_traces(
parent_trace_id=trace_id
)
# 获取 Sub-Agent 的详细步骤
for sub_trace in sub_traces:
steps = await trace_store.get_steps(sub_trace.trace_id)
print(f"{sub_trace.agent_type}: {len(steps)} steps")
exploregeneral# ✅ 好的实践:最小权限原则
AgentDefinition(
allowed_tools=["read_file", "search_code"], # 只给必要的工具
can_spawn_subagent=False, # 禁止嵌套
)
# ❌ 避免:过于宽松的权限
AgentDefinition(
allowed_tools=None, # 所有工具都可用
can_spawn_subagent=True, # 允许无限嵌套
)
AgentDefinition(
max_iterations=15, # 探索型任务:10-20 次
# max_iterations=30, # 复杂分析任务:20-40 次
)
system_prompt = """你是 XXX 专家。
专注于:
1. [核心职责 1]
2. [核心职责 2]
注意事项:
- [限制条件 1]
- [限制条件 2]
输出格式:
[期望的输出格式]
"""
# 在 Trace 中追踪每个 Sub-Agent 的成本
trace = await trace_store.get_trace(trace_id)
print(f"总成本: ${trace.total_cost:.4f}")
print(f"Token 消耗: {trace.total_tokens}")
# 查看子 Trace 的成本
for sub_trace in await trace_store.query_traces(parent_trace_id=trace_id):
print(f" {sub_trace.agent_type}: ${sub_trace.total_cost:.4f}")
A: 默认不可以。可以通过设置 can_spawn_subagent=True 允许,但要注意:
A: 可以通过以下方式:
AgentDefinition(
max_iterations=10, # 限制迭代次数
permissions={
"limits": {
"timeout_seconds": 120 # 2 分钟超时
}
}
)
A: 可以通过 Task Tool 传递必要的上下文:
# 主 Agent
prompt = f"""
分析用户 {user_id} 的项目 {project_id}。
项目路径: {project_path}
关注点: {focus_areas}
"""
task_tool(
subagent_type="explore",
description="分析项目",
prompt=prompt, # 完整的上下文信息
ctx=ctx
)
A: