Sub-Agent 机制设计

可执行规格书：本文档定义 Sub-Agent 架构。代码修改必须同步更新此文档。

📖 快速开始：查看快速参考指南了解常用操作

概述

Sub-Agent 是在独立 Trace 中运行的专门化 Agent，用于处理复杂的子任务。

核心特性

任务隔离：每个 Sub-Agent 在独立的 Trace 中运行，有完整的执行记录
权限控制：Sub-Agent 有独立的权限配置，限制其行为范围
层级关系：通过 parent_trace_id 建立父子关系，支持多层嵌套
专门化：不同类型的 Sub-Agent 专注于特定领域（如代码探索、研究分析）
防止递归：默认禁止 Sub-Agent 再启动其他 Sub-Agent，避免无限嵌套

与主 Agent 的区别

特性	主 Agent (Primary)	Sub-Agent
触发方式	用户直接调用	主 Agent 通过 Task 工具调用
运行环境	独立 Trace	独立 Trace（有父 Trace）
权限范围	完整权限	受限权限（可配置）
工具访问	所有工具	受限工具集
可见性	用户可见	主 Agent 可见，用户可选可见

架构设计

1. Agent 类型定义

# agent/models/agent.py

@dataclass
class AgentDefinition:
    """Agent 定义"""
    name: str
    description: Optional[str] = None
    mode: Literal["primary", "subagent", "all"] = "all"

    # 权限配置
    permissions: Dict[str, Any] = field(default_factory=dict)

    # 工具限制
    allowed_tools: Optional[List[str]] = None
    denied_tools: Optional[List[str]] = None

    # 模型配置
    model: Optional[str] = None
    temperature: Optional[float] = None
    max_iterations: Optional[int] = None

    # 自定义 System Prompt
    system_prompt: Optional[str] = None

    # 是否可以调用其他 Sub-Agent
    can_spawn_subagent: bool = False

实现位置：agent/models/agent.py:AgentDefinition

2. Trace 层级关系

# agent/models/trace.py (扩展现有模型)

@dataclass
class Trace:
    trace_id: str
    mode: Literal["call", "agent"]

    # 新增：Sub-Agent 支持
    parent_trace_id: Optional[str] = None  # 父 Trace ID
    agent_definition: Optional[str] = None  # Agent 类型名称
    spawned_by_tool: Optional[str] = None   # 启动此 Sub-Agent 的工具调用 ID

    # 原有字段...
    task: Optional[str] = None
    agent_type: Optional[str] = None
    status: Literal["running", "completed", "failed"] = "running"
    # ...

实现位置：agent/models/trace.py:Trace（扩展现有类）

3. Task Tool 实现

Task Tool 是启动 Sub-Agent 的核心工具：

# agent/tools/builtin/task.py

@tool(
    name="task",
    description="启动sub-agent处理复杂的子任务",
    requires_confirmation=False
)
async def task_tool(
    subagent_type: str,  # Sub-Agent 类型
    description: str,     # 任务简短描述（3-5词）
    prompt: str,          # 详细任务描述
    ctx: ToolContext
) -> ToolResult:
    """
    启动一个 Sub-Agent 执行子任务

    Args:
        subagent_type: Sub-Agent 类型（如 "explore", "general"）
        description: 任务简短描述
        prompt: 完整的任务描述
        ctx: 工具上下文

    Returns:
        Sub-Agent 的执行结果
    """
    # 1. 验证 Sub-Agent 类型
    agent_def = await get_agent_definition(subagent_type)
    if not agent_def or agent_def.mode == "primary":
        raise ValueError(f"Invalid subagent type: {subagent_type}")

    # 2. 检查权限
    if not ctx.current_agent.can_spawn_subagent:
        raise PermissionError("Current agent cannot spawn sub-agents")

    # 3. 创建子 Trace
    sub_trace = Trace.create(
        mode="agent",
        parent_trace_id=ctx.trace_id,
        agent_definition=subagent_type,
        spawned_by_tool=ctx.step_id,
        task=prompt,
        agent_type=subagent_type,
        context={
            "parent_task": ctx.trace.task,
            "description": description,
        }
    )
    sub_trace_id = await ctx.trace_store.create_trace(sub_trace)

    # 4. 配置 Sub-Agent Runner
    sub_config = AgentConfig(
        agent_type=subagent_type,
        max_iterations=agent_def.max_iterations or 10,
        # 继承父 Agent 的某些配置
        skills_dir=ctx.runner.config.skills_dir,
    )

    sub_runner = AgentRunner(
        trace_store=ctx.trace_store,
        memory_store=ctx.memory_store,
        state_store=ctx.state_store,
        tool_registry=_build_restricted_registry(agent_def),
        llm_call=ctx.runner.llm_call,
        config=sub_config,
    )

    # 5. 运行 Sub-Agent（收集所有事件）
    events = []
    async for event in sub_runner.run(
        task=prompt,
        model=agent_def.model or ctx.model,
        trace_id=sub_trace_id,
    ):
        events.append(event)
        # 可选：将事件转发给父 Agent

    # 6. 提取结果
    conclusion_events = [e for e in events if e.type == "conclusion"]
    final_result = conclusion_events[-1].data["content"] if conclusion_events else ""

    # 7. 生成摘要
    tool_summary = _summarize_tool_calls(events)

    output = f"{final_result}\n\n<task_metadata>\n"
    output += f"sub_trace_id: {sub_trace_id}\n"
    output += f"total_steps: {len(events)}\n"
    output += f"tool_calls: {tool_summary}\n"
    output += "</task_metadata>"

    return ToolResult(
        title=description,
        output=output,
        metadata={
            "sub_trace_id": sub_trace_id,
            "subagent_type": subagent_type,
            "tool_summary": tool_summary,
        }
    )


def _build_restricted_registry(agent_def: AgentDefinition) -> ToolRegistry:
    """根据 Agent 定义构建受限的工具注册表"""
    registry = ToolRegistry()
    global_registry = get_tool_registry()

    for tool_name, tool_def in global_registry.tools.items():
        # 检查是否在允许列表中
        if agent_def.allowed_tools and tool_name not in agent_def.allowed_tools:
            continue

        # 检查是否在拒绝列表中
        if agent_def.denied_tools and tool_name in agent_def.denied_tools:
            continue

        registry.register_tool(tool_def)

    # 默认禁止 task 工具（防止递归）
    if not agent_def.can_spawn_subagent:
        registry.tools.pop("task", None)

    return registry


def _summarize_tool_calls(events: List[AgentEvent]) -> str:
    """总结工具调用情况"""
    tool_calls = [e for e in events if e.type == "tool_call"]
    summary = {}
    for event in tool_calls:
        tool = event.data.get("tool", "unknown")
        summary[tool] = summary.get(tool, 0) + 1

    return ", ".join(f"{tool}×{count}" for tool, count in summary.items())

实现位置：agent/tools/builtin/task.py:task_tool

内置 Sub-Agent 类型

1. general - 通用型

用途：执行复杂的多步骤任务和研究分析

GENERAL_AGENT = AgentDefinition(
    name="general",
    description="通用型 Sub-Agent，用于执行复杂的多步骤任务和研究分析",
    mode="subagent",
    allowed_tools=None,  # 允许所有工具
    denied_tools=["task"],  # 禁止启动其他 Sub-Agent
    max_iterations=20,
)

典型使用场景：

多步骤的数据收集和分析
复杂的文件处理任务
需要多个工具协同的任务

2. explore - 探索型

用途：快速探索代码库，查找文件和代码

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,
    system_prompt="""你是一个代码探索专家。专注于：
1. 快速定位相关文件和代码
2. 理解代码结构和依赖关系
3. 总结关键信息

注意：
- 你只能读取和搜索，不能编辑或执行代码
- 优先使用搜索工具，而不是逐个读取文件
- 提供清晰的文件路径和行号引用
""",
)

典型使用场景：

"找出所有使用了 Redis 的代码"
"分析 API 路由的实现方式"
"查找配置文件的加载逻辑"

3. analyst - 分析型

用途：深度分析和报告生成

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 配置系统

配置文件格式

{
  "agents": {
    "custom-reviewer": {
      "description": "代码审查专家",
      "mode": "subagent",
      "allowed_tools": ["read_file", "search_code", "list_files"],
      "denied_tools": ["write_file", "edit_file", "execute_bash"],
      "max_iterations": 10,
      "temperature": 0.2,
      "system_prompt": "你是一个代码审查专家...",
      "can_spawn_subagent": false
    },
    "my-primary": {
      "description": "自定义主 Agent",
      "mode": "primary",
      "can_spawn_subagent": true
    }
  },
  "default_agent": "my-primary"
}

加载和管理

# agent/agent_registry.py

class AgentRegistry:
    """Agent 注册表"""

    def __init__(self):
        self.agents: Dict[str, AgentDefinition] = {}
        self._load_builtin_agents()

    def _load_builtin_agents(self):
        """加载内置 Agent"""
        self.register(GENERAL_AGENT)
        self.register(EXPLORE_AGENT)
        self.register(ANALYST_AGENT)

    def load_from_config(self, config_path: str):
        """从配置文件加载自定义 Agent"""
        import json
        with open(config_path) as f:
            config = json.load(f)

        for name, cfg in config.get("agents", {}).items():
            agent_def = AgentDefinition(
                name=name,
                **cfg
            )
            self.register(agent_def)

    def register(self, agent: AgentDefinition):
        """注册 Agent"""
        self.agents[agent.name] = agent

    def get(self, name: str) -> Optional[AgentDefinition]:
        """获取 Agent 定义"""
        return self.agents.get(name)

    def list_subagents(self) -> List[AgentDefinition]:
        """列出所有可用的 Sub-Agent"""
        return [
            agent for agent in self.agents.values()
            if agent.mode in ("subagent", "all")
        ]


# 全局注册表
_agent_registry = AgentRegistry()

def get_agent_registry() -> AgentRegistry:
    """获取全局 Agent 注册表"""
    return _agent_registry

async def get_agent_definition(name: str) -> Optional[AgentDefinition]:
    """获取 Agent 定义"""
    return _agent_registry.get(name)

实现位置：agent/agent_registry.py:AgentRegistry

权限控制

权限模型

每个 Agent 可以定义细粒度的权限：

PERMISSIONS_EXAMPLE = {
    # 工具级别权限
    "tools": {
        "write_file": "deny",
        "read_file": "allow",
        "execute_bash": "deny",
    },

    # 路径级别权限
    "paths": {
        "/etc": "deny",
        "/tmp": "allow",
        "~/.ssh": "deny",
    },

    # 网络级别权限
    "network": {
        "allowed_domains": ["*.example.com", "api.github.com"],
        "blocked_domains": ["*.evil.com"],
    },

    # 资源限制
    "limits": {
        "max_file_size": 10_000_000,  # 10MB
        "max_iterations": 15,
        "timeout_seconds": 300,
    }
}

权限检查流程

# agent/permission.py

class PermissionChecker:
    """权限检查器"""

    def __init__(self, agent_def: AgentDefinition):
        self.agent_def = agent_def
        self.permissions = agent_def.permissions

    def check_tool_access(self, tool_name: str) -> bool:
        """检查工具访问权限"""
        # 检查拒绝列表
        if self.agent_def.denied_tools and tool_name in self.agent_def.denied_tools:
            return False

        # 检查允许列表
        if self.agent_def.allowed_tools and tool_name not in self.agent_def.allowed_tools:
            return False

        # 检查工具权限配置
        tool_perms = self.permissions.get("tools", {})
        if tool_perms.get(tool_name) == "deny":
            return False

        return True

    def check_path_access(self, path: str, mode: str = "read") -> bool:
        """检查路径访问权限"""
        path_perms = self.permissions.get("paths", {})

        for pattern, action in path_perms.items():
            if self._match_path(path, pattern):
                return action == "allow"

        # 默认允许读取，拒绝写入
        return mode == "read"

    def _match_path(self, path: str, pattern: str) -> bool:
        """路径匹配（支持通配符）"""
        import fnmatch
        return fnmatch.fnmatch(path, pattern)

实现位置：agent/permission.py:PermissionChecker

使用示例

1. 主 Agent 调用 Sub-Agent

# 主 Agent 的 System Prompt 中会包含 Task 工具说明
system_prompt = """
你是一个智能助手。当遇到复杂任务时，可以使用 task 工具启动专门的 Sub-Agent。

可用的 Sub-Agent：
- explore: 探索代码库，查找文件和代码
- general: 执行复杂的多步骤任务
- analyst: 深度分析和报告生成

示例：
用户："这个项目用了哪些数据库？"
你：使用 task(subagent_type="explore", description="查找数据库使用", prompt="...")
"""

# 主 Agent 执行时
async for event in runner.run(task="分析这个项目的架构"):
    if event.type == "tool_call" and event.data["tool"] == "task":
        # Sub-Agent 被启动
        print(f"启动 Sub-Agent: {event.data['args']['subagent_type']}")

2. 自定义 Sub-Agent

from agent import AgentRunner, AgentConfig
from agent.agent_registry import get_agent_registry
from agent.models.agent import AgentDefinition

# 1. 定义自定义 Sub-Agent
custom_agent = AgentDefinition(
    name="security-scanner",
    description="安全扫描专家",
    mode="subagent",
    allowed_tools=["read_file", "search_code", "list_files"],
    system_prompt="""你是一个安全扫描专家。专注于：
1. 查找常见安全漏洞（SQL注入、XSS、CSRF等）
2. 检查敏感信息泄露（密钥、密码等）
3. 分析依赖项的安全问题

输出格式：
- 漏洞类型
- 影响范围
- 修复建议
""",
    max_iterations=20,
)

# 2. 注册到全局注册表
get_agent_registry().register(custom_agent)

# 3. 主 Agent 可以调用它
# 在 System Prompt 中会自动列出这个新的 Sub-Agent

3. 监控 Sub-Agent 执行

from agent.storage import TraceStore

async def analyze_subagent_performance(trace_id: str, trace_store: TraceStore):
    """分析 Sub-Agent 性能"""
    trace = await trace_store.get_trace(trace_id)
    steps = await trace_store.get_steps(trace_id)

    # 找出所有 Sub-Agent 调用
    subagent_calls = []
    for step in steps:
        if step.step_type == "tool_call" and step.data.get("tool") == "task":
            sub_trace_id = step.data.get("result", {}).get("metadata", {}).get("sub_trace_id")
            if sub_trace_id:
                sub_trace = await trace_store.get_trace(sub_trace_id)
                subagent_calls.append({
                    "type": sub_trace.agent_type,
                    "task": sub_trace.task,
                    "steps": sub_trace.total_steps,
                    "tokens": sub_trace.total_tokens,
                    "cost": sub_trace.total_cost,
                })

    # 生成报告
    print(f"主任务: {trace.task}")
    print(f"Sub-Agent 调用次数: {len(subagent_calls)}")
    for i, call in enumerate(subagent_calls, 1):
        print(f"\n{i}. {call['type']}")
        print(f"   任务: {call['task'][:50]}...")
        print(f"   步骤: {call['steps']}, Token: {call['tokens']}, 成本: ${call['cost']:.4f}")

实现计划

Phase 1: 基础架构（MVP）

[ ] AgentDefinition 模型 (agent/models/agent.py)
- 定义 Agent 类型、权限、配置
[ ] Trace 扩展 (agent/models/trace.py)
- 添加 parent_trace_id、agent_definition 等字段
[ ] AgentRegistry (agent/agent_registry.py)
- Agent 注册和管理
- 加载内置和自定义 Agent
[ ] Task Tool (agent/tools/builtin/task.py)
- 实现 Sub-Agent 启动逻辑
- Trace 层级管理
- 结果聚合
[ ] 内置 Sub-Agent (agent/builtin_agents.py)
- general: 通用型
- explore: 探索型

Phase 2: 权限控制

[ ] PermissionChecker (agent/permission.py)
- 工具级别权限检查
- 路径级别权限检查
- 资源限制检查
[ ] 受限 ToolRegistry (agent/tools/builtin/task.py)
- 根据 Agent 定义过滤工具

Phase 3: 高级特性

[ ] 配置系统
- JSON 配置文件支持
- 动态加载自定义 Agent
[ ] 监控和分析
- Sub-Agent 性能统计
- 调用链可视化
[ ] 递归控制
- 嵌套深度限制
- 循环检测

集成点

与现有系统的集成

AgentRunner (agent/runner.py)
- 扩展 run() 方法支持 agent_definition 参数
- 根据 Agent 定义配置工具和权限
ToolRegistry (agent/tools/registry.py)
- 添加 Task 工具到全局注册表
- 支持工具过滤和权限检查
TraceStore (agent/storage/protocols.py)
- 支持根据 parent_trace_id 查询子 Trace
- 添加层级查询方法
Events (agent/events.py)
- 添加 Sub-Agent 相关事件类型
- subagent_started、subagent_completed

注意事项

防止无限递归
- 默认禁止 Sub-Agent 调用 Task 工具
- 设置最大嵌套深度（建议 3 层）
性能考虑
- Sub-Agent 增加了额外的 LLM 调用开销
- 合理设置 max_iterations 限制
成本控制
- 跟踪每个 Sub-Agent 的 token 消耗
- 在父 Trace 中聚合成本统计
错误处理
- Sub-Agent 失败不应导致主任务完全失败
- 提供降级策略（如超时、重试）
可观测性
- 完整记录 Sub-Agent 的执行过程
- 支持调用链追踪和分析

参考资料

OpenCode 实现：/Users/sunlit/Code/opencode/packages/opencode/src/tool/task.ts
OpenCode Agent 定义：/Users/sunlit/Code/opencode/packages/opencode/src/agent/agent.ts
当前项目架构：docs/README.md

sub-agents.md 19 KB Историја Датотека

Sub-Agent 机制设计

概述

核心特性

与主 Agent 的区别

架构设计

1. Agent 类型定义

2. Trace 层级关系

3. Task Tool 实现

内置 Sub-Agent 类型

1. general - 通用型

2. explore - 探索型

3. analyst - 分析型

Agent 配置系统

配置文件格式

加载和管理

权限控制

权限模型

权限检查流程

使用示例

1. 主 Agent 调用 Sub-Agent

2. 自定义 Sub-Agent

3. 监控 Sub-Agent 执行

实现计划

Phase 1: 基础架构（MVP）

Phase 2: 权限控制

Phase 3: 高级特性

集成点

与现有系统的集成

注意事项

参考资料

sub-agents.md 19 KB

Историја Датотека