# Sub-Agent 机制设计

> **可执行规格书**：本文档定义 Sub-Agent 架构。代码修改必须同步更新此文档。
>
> 📖 **快速开始**：查看 [快速参考指南](./sub-agents-quickref.md) 了解常用操作

---

## 概述

**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 类型定义

```python
# 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 层级关系

```python
# 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 的核心工具：

```python
# 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 - 通用型

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

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

**典型使用场景**：
- 多步骤的数据收集和分析
- 复杂的文件处理任务
- 需要多个工具协同的任务

### 2. explore - 探索型

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

```python
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 - 分析型

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

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

### 配置文件格式

```json
{
  "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"
}
```

### 加载和管理

```python
# 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 可以定义细粒度的权限：

```python
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,
    }
}
```

### 权限检查流程

```python
# 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

```python
# 主 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

```python
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 执行

```python
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 性能统计
  - 调用链可视化

- [ ] **递归控制**
  - 嵌套深度限制
  - 循环检测

---

## 集成点

### 与现有系统的集成

1. **AgentRunner** (`agent/runner.py`)
   - 扩展 `run()` 方法支持 `agent_definition` 参数
   - 根据 Agent 定义配置工具和权限

2. **ToolRegistry** (`agent/tools/registry.py`)
   - 添加 Task 工具到全局注册表
   - 支持工具过滤和权限检查

3. **TraceStore** (`agent/storage/protocols.py`)
   - 支持根据 `parent_trace_id` 查询子 Trace
   - 添加层级查询方法

4. **Events** (`agent/events.py`)
   - 添加 Sub-Agent 相关事件类型
   - `subagent_started`、`subagent_completed`

---

## 注意事项

1. **防止无限递归**
   - 默认禁止 Sub-Agent 调用 Task 工具
   - 设置最大嵌套深度（建议 3 层）

2. **性能考虑**
   - Sub-Agent 增加了额外的 LLM 调用开销
   - 合理设置 `max_iterations` 限制

3. **成本控制**
   - 跟踪每个 Sub-Agent 的 token 消耗
   - 在父 Trace 中聚合成本统计

4. **错误处理**
   - Sub-Agent 失败不应导致主任务完全失败
   - 提供降级策略（如超时、重试）

5. **可观测性**
   - 完整记录 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`