beaver_project/app-instance/backend/A2A_Multiagent_change.md

# A2A Multi-Agent 改造方案

## 1. 需求目标

当前 `spawn`/`sub-agent` 只有一种执行方式: 创建一个本地后台 subagent 去完成任务。

这次需求要改成:

1. 调用 `sub-agent` 时，不一定新建本地 subagent。
2. 先从“已添加的 Agent”里找可用目标。
3. 再从 skills 中声明的 `agent cards` 里找可用目标。
4. 通过 A2A 协议把任务发给对应 agent。
5. 支持一个任务发给多个 agent，形成 `agent group`，最后回到主 agent 汇总。
6. 保持现有 `spawn(task, label)` 兼容，不破坏已有行为。

结论先说:

- 最合适的做法不是继续把能力堆进 `SubagentManager`。
- 应该把“本地 subagent 执行”升级为“统一委派层”。
- `spawn` 工具继续保留，但语义从“创建 subagent”扩展为“委派给合适的 agent / agent group”。

## 2. 当前代码现状

### 2.1 当前触发链路

现有链路很单一:

1. `AgentLoop` 初始化 `SubagentManager`
   - 位置: `nanobot/agent/loop.py:88-114`
2. `AgentLoop._register_default_tools()` 注册 `SpawnTool`
   - 位置: `nanobot/agent/loop.py:116-138`
3. LLM 调用 `spawn(task, label)`
4. `SpawnTool.execute()` 直接转发给 `SubagentManager.spawn()`
   - 位置: `nanobot/agent/tools/spawn.py:67-76`
5. `SubagentManager.spawn()` 创建本地 asyncio 后台任务
   - 位置: `nanobot/agent/subagent.py:64-93`
6. `_run_subagent()` 用一个受限工具集运行本地子代理
   - 位置: `nanobot/agent/subagent.py:95-195`
7. `_announce_result()` 把结果包装成 `channel="system"` 的消息回投主消息总线
   - 位置: `nanobot/agent/subagent.py:197-230`
8. `AgentLoop._process_message()` 接到 `system` 消息，再整理成用户可见回复
   - 位置: `nanobot/agent/loop.py:331-347`

### 2.2 当前已经有但没接入调度链路的能力

仓库里已经有两类“候选 agent 信息”，但没有进入实际调度:

1. Plugin agents
   - `PluginLoader.find_agent()` 已能找 agent
   - 位置: `nanobot/agent/plugins.py:83-91`
   - `build_agents_summary()` 也已能汇总 agent 信息
   - 位置: `nanobot/agent/plugins.py:100-121`
   - 但当前 `AgentLoop` / `ContextBuilder` 并没有用它做调度

2. Skills
   - `SkillsLoader` 已能枚举 / 读取 skill
   - 位置: `nanobot/agent/skills.py:32-249`
   - 但 skill 目前只被当作 prompt 资源，不会暴露成“可路由 agent”

### 2.3 当前缺口

当前缺少这几层:

1. 统一的 `Agent Registry`
2. A2A `agent card` 发现与缓存
3. A2A client 调用层
4. 统一的委派器，负责在“本地 subagent / plugin agent / skill agent card / agent group”之间做路由
5. group 级别的状态管理和结果聚合

## 3. 推荐总方案

推荐采用“保留 `spawn` 工具名，重构内部执行层”的方案。

### 3.1 核心思路

把当前:

- `SpawnTool -> SubagentManager -> 本地 subagent`

改成:

- `SpawnTool -> DelegationManager -> AgentResolver -> Executor(local/plugin/a2a/group)`

也就是:

1. `spawn` 不再等价于“必须创建 subagent”。
2. `spawn` 变成“委派任务”。
3. 真正执行方式由委派层动态决定。

### 3.2 为什么这样最合适

如果直接继续扩 `SubagentManager`，很快会出现这些问题:

1. 一个类同时负责本地 LLM 运行、A2A 网络调用、agent card 发现、group 并发、结果聚合。
2. 后续要支持 plugin agent、本地 named agent、A2A streaming 时会越来越乱。
3. 当前 `SubagentManager` 的职责本来就已经比较明确: “本地后台 subagent 执行器”。

所以更合理的拆法是:

1. `SubagentManager` 保留或下沉为 `LocalSubagentExecutor`
2. 新增 `DelegationManager` 作为统一入口
3. 新增 `AgentRegistry` / `AgentResolver`
4. 新增 `A2AClient`

## 4. 推荐模块拆分

### 4.1 新增 `DelegationManager`

建议新文件:

- `nanobot/agent/delegation.py`

职责:

1. 接收 `spawn` 请求
2. 根据参数和任务内容选择目标 agent
3. 决定执行方式
4. 对 group 做并发调度
5. 统一把结果回投主消息总线

建议接口:

```python
class DelegationManager:
    async def dispatch(
        self,
        task: str,
        label: str | None = None,
        target: str | None = None,
        targets: list[str] | None = None,
        strategy: str = "auto",
        origin_channel: str = "cli",
        origin_chat_id: str = "direct",
    ) -> str: ...
```

### 4.2 保留本地执行器

当前 `nanobot/agent/subagent.py` 的 `_run_subagent()` 逻辑可以保留，但角色改为:

- `LocalSubagentExecutor`

也可以第一版不重命名文件，只把里面逻辑拆成:

1. `spawn_local()`
2. `_run_local_subagent()`
3. `_announce_local_result()`

这样可以最小改动落地。

### 4.3 新增 `AgentRegistry`

建议新文件:

- `nanobot/agent/agent_registry.py`

职责:

1. 汇总所有可调度 agent
2. 统一输出规范化 descriptor
3. 维护优先级和去重逻辑

统一后的 agent 来源:

1. workspace 中“已添加的 agent”
2. plugin agents
3. skill frontmatter 里声明的 `agent_cards`
4. 必要时 fallback 到本地 `local-subagent`

建议统一 descriptor:

```python
@dataclass
class AgentDescriptor:
    id: str
    name: str
    description: str
    source: str           # workspace | plugin | skill | builtin
    kind: str             # local_prompt | a2a_remote | local_fallback
    protocol: str | None  # a2a | None
    plugin_name: str | None = None
    skill_name: str | None = None
    model: str | None = None
    endpoint: str | None = None
    card_url: str | None = None
    tags: list[str] = field(default_factory=list)
    capabilities: dict[str, Any] = field(default_factory=dict)
    metadata: dict[str, Any] = field(default_factory=dict)
```

### 4.4 新增 A2A client 层

建议新目录:

- `nanobot/a2a/client.py`
- `nanobot/a2a/cards.py`
- `nanobot/a2a/models.py`

职责:

1. 获取 agent card
2. 解析 card 能力
3. 对远端 agent 发 JSON-RPC 请求
4. 处理同步返回 / task 轮询 / streaming 兼容

## 5. 代码插入点

## 5.1 `nanobot/agent/loop.py`

### 插入点 A: `__init__`

当前:

- `self.subagents = SubagentManager(...)`
- 位置: `nanobot/agent/loop.py:88-102`

建议改成:

1. 初始化 `PluginLoader`
2. 初始化 `AgentRegistry`
3. 初始化 `DelegationManager`
4. `DelegationManager` 内部持有 `LocalSubagentExecutor` / `A2AExecutor`

推荐形态:

```python
self.plugins = PluginLoader(workspace)
self.agent_registry = AgentRegistry(workspace, plugins=self.plugins, ...)
self.delegation = DelegationManager(
    provider=provider,
    workspace=workspace,
    bus=bus,
    registry=self.agent_registry,
    ...
)
```

### 插入点 B: `_register_default_tools`

当前:

- 注册 `SpawnTool(manager=self.subagents)`
- 位置: `nanobot/agent/loop.py:130-134`

建议改成:

```python
self.tools.register(SpawnTool(manager=self.delegation))
```

### 插入点 C: `_set_tool_context`

当前会给 `spawn` 工具写 origin context:

- 位置: `nanobot/agent/loop.py:165-192`

这里逻辑可以继续保留，不需要大改，因为 A2A / group 结果最终也要回到原会话。

## 5.2 `nanobot/agent/tools/spawn.py`

当前 `SpawnTool` 参数只有:

- `task`
- `label`

位置:

- schema: `nanobot/agent/tools/spawn.py:49-65`
- execute: `nanobot/agent/tools/spawn.py:67-76`

建议扩成:

```python
{
  "task": "string",
  "label": "string?",
  "target": "string?",
  "targets": "string[]?",
  "strategy": "auto|local|plugin|a2a|group"
}
```

兼容规则:

1. 老调用只传 `task/label` 时，等价于 `strategy="auto"`
2. `target` 表示单目标
3. `targets` 表示 group
4. `strategy="local"` 强制走本地 subagent
5. `strategy="a2a"` 强制只找 A2A 目标

## 5.3 `nanobot/agent/context.py`

当前 prompt 中只注入:

1. bootstrap
2. memory
3. skills summary

位置:

- `build_system_prompt()`: `nanobot/agent/context.py:38-76`

建议新增一段:

- `# Available Agents`

由 `AgentRegistry.build_agents_summary()` 生成，内容只放:

1. agent id / name
2. 简短 description
3. source
4. protocol
5. 是否支持 group / streaming

目标是让主 agent 知道:

1. 当前有哪些现成 agent 可用
2. 什么时候应该 `spawn(target=...)`
3. 哪些是 skill 暴露出来的 A2A agent

## 5.4 `nanobot/agent/skills.py`

这是 skill agent cards 的关键入口。

当前 skill frontmatter 已支持 `metadata` 字段，并会解析其中的 JSON:

- `_parse_nanobot_metadata()`: `nanobot/agent/skills.py:190-196`
- `_get_skill_meta()`: `nanobot/agent/skills.py:209-212`

最推荐的做法不是去扫 `SKILL.md` 正文里的自由文本，而是约定 skill frontmatter 的 `metadata.nanobot.agent_cards`。

建议新增:

```python
def list_skill_agent_cards(self) -> list[dict[str, Any]]: ...
```

推荐 skill 写法:

```md
---
name: github-research
description: GitHub research helper
metadata: '{"nanobot":{"agent_cards":[{"id":"repo-analyst","url":"https://example.com/.well-known/agent-card","tags":["github","research"],"auth_env":"REPO_AGENT_TOKEN"}]}}'
---
```

为什么推荐这样做:

1. 当前 frontmatter 解析已经存在
2. 不需要引入新的 skill 文件格式
3. 不需要解析自由文本
4. skill 打包/上传链路也不需要大改

## 5.5 `nanobot/agent/plugins.py`

当前 plugin agents 已能加载:

- `find_agent()`: `nanobot/agent/plugins.py:83-91`
- `_load_agents()`: `nanobot/agent/plugins.py:210-229`

建议:

1. `AgentRegistry` 直接复用 `PluginLoader`
2. plugin agent 作为“本地可执行 agent”来源之一

这里不建议把 plugin agent 强行转成 A2A。

更合理的处理是:

1. plugin agent 本地执行
2. skill agent cards 远程 A2A 调用
3. workspace 手动添加的 agent 也可走 A2A

## 5.6 `nanobot/config/schema.py`

当前 `ToolsConfig` 只有:

- `web`
- `exec`
- `restrict_to_workspace`
- `mcp_servers`

位置:

- `nanobot/config/schema.py:337-347`

建议新增:

```python
class A2AConfig(Base):
    enabled: bool = True
    timeout_seconds: int = 30
    poll_interval_seconds: int = 2
    card_cache_ttl_seconds: int = 300
    max_parallel_agents: int = 4
    allow_skill_cards: bool = True
    allow_workspace_agents: bool = True
    allowed_hosts: list[str] = Field(default_factory=list)
```

然后挂到:

```python
class ToolsConfig(Base):
    ...
    a2a: A2AConfig = Field(default_factory=A2AConfig)
```

## 5.7 `nanobot/web/server.py`

当前 web API 有:

- `/api/skills`
- `/api/plugins`

位置:

- skills: `nanobot/web/server.py:702-843`
- plugins: `nanobot/web/server.py:1000-1037`

建议新增:

1. `GET /api/agents`
   - 返回统一后的 agent registry
2. `POST /api/agents`
   - 添加 workspace agent card
3. `DELETE /api/agents/{id}`
   - 删除 workspace agent
4. `POST /api/agents/refresh`
   - 刷新 card cache

这样“已添加的 Agent”才有明确的持久化来源。

## 6. 推荐的数据来源优先级

为了行为稳定，推荐 resolver 按以下优先级匹配:

1. workspace 手动添加的 agent
2. plugin agents
3. skill metadata 里的 agent cards
4. fallback 到本地 subagent

原因:

1. workspace 手动添加通常是用户明确希望接入的 agent
2. plugin agent 是本地稳定能力
3. skill card 往往是外部资源，可信度和可用性最弱
4. 本地 subagent 最后兜底，保证老行为不失效

## 7. A2A 协议接入建议

## 7.1 Agent Card 发现

建议支持 3 种入口:

1. 显式 `card_url`
2. `base_url + /.well-known/agent-card`
3. fallback `base_url + /.well-known/agent.json`

这样做的原因是:

1. 当前 A2A 文档和样例在 card 路径上存在新旧写法并存
2. 兼容性会更好

## 7.2 RPC 调用兼容层

建议客户端优先尝试:

1. `tasks/send`
2. 不支持时 fallback `message/send`

后续可选支持:

1. `tasks/sendSubscribe`
2. `message/sendStream`
3. `tasks/get`
4. `tasks/cancel`

推荐第一期先做:

1. 非流式发任务
2. 如果返回 `Task` 状态不是最终态，就轮询 `tasks/get`

这样能最小代价先打通。

## 7.3 发送给远端 agent 的上下文范围

不要把主会话完整 history 直接发给远端 agent。

建议第一版只发送:

1. 任务目标
2. 必要的结构化说明
3. 主 agent 整理好的最小上下文

原因:

1. 当前本地 subagent 也不共享主会话历史
2. 外部 A2A agent 不可信时，最小化数据泄漏面
3. 避免 token 膨胀

## 8. agent group 设计

## 8.1 什么时候触发 group

建议第一版只支持两种触发:

1. 用户明确指定多个 agent
2. LLM 在工具调用里显式传 `targets=[...]`

不建议第一版做“自动拆成多个 agent 并行”的强自动化。

原因:

1. 容易失控
2. 很难解释为什么调了这些 agent
3. 对成本和网络调用不可控

## 8.2 group 执行链路

推荐链路:

1. `SpawnTool.execute()` 收到 `targets`
2. `DelegationManager.dispatch()` 创建 `group_run_id`
3. `AgentRegistry` 解析出每个 target 的 descriptor
4. 按 executor 类型并发执行
5. `asyncio.gather(..., return_exceptions=True)` 收集结果
6. 统一做 group aggregation
7. `_announce_group_result()` 回投主消息总线
8. 主 agent 再生成最终用户回复

## 8.3 group 结果聚合

建议 group 执行器输出结构化结果:

```python
@dataclass
class AgentRunResult:
    agent_id: str
    status: str          # ok | error | timeout | cancelled
    summary: str
    raw: dict[str, Any] | None = None
```

group 最终回投内容建议类似:

```text
[Agent group 'repo-check' completed]

Members:
- researcher: ok
- reviewer: ok
- planner: error

Results:
...

Summarize this naturally for the user. Mention disagreements if any.
```

这样能继续复用当前 `system -> main agent -> user` 的输出模式。

## 9. 推荐触发方式

## 9.1 用户显式触发

用户说法示例:

1. “把这个任务交给 `github-reviewer`”
2. “让 `researcher` 和 `reviewer` 一起处理”
3. “如果有现成 agent 就不要新建 subagent”

这时主 agent 应调用:

```json
{
  "task": "...",
  "target": "github-reviewer"
}
```

或者:

```json
{
  "task": "...",
  "targets": ["researcher", "reviewer"],
  "strategy": "group"
}
```

## 9.2 模型自主触发

当主 agent 判断:

1. 任务独立可并行
2. 已有 agent 专长明显更匹配
3. 任务耗时长，适合后台执行

则调用 `spawn`，但不再默认认为一定是“新建本地 subagent”。

## 9.3 自动回退

如果没有找到匹配 agent:

1. `strategy=auto` -> fallback 本地 subagent
2. `strategy=a2a` -> 直接返回未找到
3. `strategy=group` 且部分目标不存在 -> 明确报错或只跑已解析目标，建议第一版严格报错

## 10. workspace 中“已添加 agent”的建议存储

建议新增:

- `workspace/agents/registry.json`

示例:

```json
[
  {
    "id": "github-reviewer",
    "name": "GitHub Reviewer",
    "description": "Review GitHub repository changes",
    "protocol": "a2a",
    "base_url": "https://reviewer.example.com/a2a",
    "card_url": "https://reviewer.example.com/.well-known/agent-card",
    "auth_env": "GITHUB_REVIEWER_TOKEN",
    "enabled": true,
    "tags": ["github", "review"]
  }
]
```

为什么不用直接塞进 `config.json`:

1. 这是 workspace 维度资源，不是全局运行参数
2. web API 做增删改查更方便
3. 不要求用户每次改 agent 都改配置再重启

## 11. 推荐实施顺序

### Phase 1: 打通单 agent 路由

目标:

1. 引入 `AgentRegistry`
2. `spawn` 支持 `target`
3. 支持 workspace agent 和 skill agent card
4. 支持 A2A 单点调用
5. 找不到时 fallback 本地 subagent

### Phase 2: 接入 plugin agent 本地执行

目标:

1. plugin agent 进入统一 registry
2. plugin agent 可作为 `target`
3. 本地 prompt-based agent 与 A2A remote agent 共存

### Phase 3: group 并发和聚合

目标:

1. `targets=[...]`
2. 并发执行
3. group 级状态跟踪
4. 聚合后回投主 agent

### Phase 4: web 管理接口

目标:

1. `/api/agents`
2. 添加 / 删除 / 刷新 agent
3. 前端展示 unified registry

## 12. 兼容性要求

这次改造一定要保留以下兼容性:

1. 旧的 `spawn(task, label)` 调用仍然可用
2. 没有 A2A agent 时，行为和现在一致
3. skill 没写 `agent_cards` 时，skill 仍只是普通 skill
4. plugin agent 不参与调度时，现有 plugin 机制不受影响

## 13. 风险点

### 13.1 A2A 规范新旧写法并存

从当前公开文档和样例看，存在这些并行写法:

1. card 路径: `/.well-known/agent-card` 和 `/.well-known/agent.json`
2. RPC 方法: `tasks/send` 和 `message/send`

所以客户端必须做兼容适配，不能写死一种。

### 13.2 外部 agent 的安全边界

需要限制:

1. 白名单 host
2. 超时
3. card cache TTL
4. skill card 是否允许自动启用

### 13.3 远端 agent 无法直接访问本地 workspace

这意味着:

1. 不能把“去读本地文件然后处理”原样发给远端 A2A agent
2. 主 agent 需要先整理出必要上下文
3. 第一版最好只做文本级委派

## 14. 我建议的落地结论

如果要控制改动面，又要保证后续可扩展，推荐最终采用下面这个结构:

```text
AgentLoop
  -> SpawnTool
    -> DelegationManager
      -> AgentRegistry / AgentResolver
      -> LocalSubagentExecutor
      -> PluginAgentExecutor
      -> A2AExecutor
      -> AgentGroupExecutor
      -> announce_result() -> MessageBus(system) -> AgentLoop -> user
```

也就是说:

1. `spawn` 工具保留
2. `SubagentManager` 不再是唯一执行器
3. `DelegationManager` 成为真正总入口
4. skills 里的 `agent_cards` 用 frontmatter metadata 承载
5. workspace agent 单独持久化
6. group 通过并发 executor + 汇总消息实现

这是当前仓库里最稳妥、最符合现有架构的改法。

## 15. 外部参考

以下是我写这个方案时核对的 A2A 资料:

1. A2A Protocol Development Guide: https://a2aprotocol.ai/docs/guide/a2a-typescript-guide.html
2. Python A2A Tutorial: https://a2aprotocol.ai/docs/guide/python-a2a-tutorial-20250513

注意:

1. 当前公开文档里既能看到 `tasks/send`，也能看到 `message/send`
2. agent card 路径也能看到 `agent-card` 与 `agent.json` 两种写法
3. 所以实现时建议做兼容层，不要只押一种命名