可执行规格书:Sub-Agent 实现细节。代码修改必须同步更新此文档。
# agent/models/agent.py
@dataclass
class AgentDefinition:
name: str
description: Optional[str] = None
mode: Literal["primary", "subagent", "all"] = "all"
# 权限配置
allowed_tools: Optional[List[str]] = None # 白名单
denied_tools: Optional[List[str]] = None # 黑名单
permissions: Dict[str, Any] = field(default_factory=dict)
# 模型配置
model: Optional[str] = None
temperature: Optional[float] = None
max_iterations: Optional[int] = None
system_prompt: Optional[str] = None
# 是否可以调用其他 Sub-Agent
can_spawn_subagent: bool = False
实现位置:agent/models/agent.py:AgentDefinition(待实现)
# agent/models/trace.py (扩展现有模型)
@dataclass
class Trace:
# ... 原有字段
# Sub-Agent 支持(新增)
parent_trace_id: Optional[str] = None # 父 Trace ID
agent_definition: Optional[str] = None # Agent 类型名称
spawned_by_tool: Optional[str] = None # 启动此 Sub-Agent 的 Step ID
实现位置:agent/models/trace.py:Trace
GENERAL_AGENT = AgentDefinition(
name="general",
description="通用型 Sub-Agent,执行复杂的多步骤任务",
mode="subagent",
denied_tools=["task"],
max_iterations=20,
)
用途:多步骤任务、数据收集、复杂分析
EXPLORE_AGENT = AgentDefinition(
name="explore",
description="探索型 Sub-Agent,快速查找文件和代码",
mode="subagent",
allowed_tools=["read_file", "list_files", "search_code", "search_files"],
denied_tools=["write_file", "edit_file", "execute_bash", "task"],
max_iterations=15,
)
用途:代码库探索、文件查找、结构分析
ANALYST_AGENT = AgentDefinition(
name="analyst",
description="分析型 Sub-Agent,深度分析和报告生成",
mode="subagent",
allowed_tools=["read_file", "list_files", "search_code", "web_search", "fetch_url"],
denied_tools=["task", "write_file", "edit_file"],
max_iterations=25,
temperature=0.3,
)
用途:技术栈分析、性能分析、安全审计
实现位置:agent/builtin_agents.py(待实现)
# agent/tools/builtin/task.py
@tool(name="task")
async def task_tool(
subagent_type: str, # Sub-Agent 类型
description: str, # 任务简短描述(3-5词)
prompt: str, # 详细任务描述
ctx: ToolContext
) -> ToolResult:
"""启动 Sub-Agent 处理子任务"""
实现位置:agent/tools/builtin/task.py:task_tool(待实现)
# agent/agent_registry.py
class AgentRegistry:
def register(self, agent: AgentDefinition) -> None:
"""注册 Agent"""
def get(self, name: str) -> Optional[AgentDefinition]:
"""获取 Agent 定义"""
def list_subagents(self) -> List[AgentDefinition]:
"""列出所有可用的 Sub-Agent"""
def load_from_config(self, config_path: str) -> None:
"""从配置文件加载"""
# 全局注册表
def get_agent_registry() -> AgentRegistry:
"""获取全局 Agent 注册表"""
实现位置:agent/agent_registry.py:AgentRegistry(待实现)
{
"agents": {
"code-reviewer": {
"description": "代码审查专家",
"mode": "subagent",
"allowed_tools": ["read_file", "search_code", "list_files"],
"max_iterations": 15,
"temperature": 0.2,
"system_prompt": "你是代码审查专家...",
"can_spawn_subagent": false
}
}
}
框架预设:agent/subagents/default.json
项目配置(可选):./subagents/custom.json
from agent.agent_registry import get_agent_registry
# 加载框架预设
registry = get_agent_registry()
registry.load_from_config("agent/subagents/default.json")
# 加载项目自定义配置(可选,覆盖框架预设)
from pathlib import Path
if Path("./subagents/custom.json").exists():
registry.load_from_config("./subagents/custom.json")
from agent import AgentRunner, AgentConfig
runner = AgentRunner(
llm_call=your_llm,
config=AgentConfig(max_iterations=20),
)
# 主 Agent 会自动使用 task 工具启动 Sub-Agent
async for event in runner.run(
task="使用 explore sub-agent 查找所有配置文件"
):
if event.type == "conclusion":
print(event.data["content"])
from agent.models.agent import AgentDefinition
from agent.agent_registry import get_agent_registry
# 定义
custom = AgentDefinition(
name="security-scanner",
description="安全扫描专家",
mode="subagent",
allowed_tools=["read_file", "search_code"],
)
# 注册
get_agent_registry().register(custom)
完整示例:examples/subagent_example.py