返回博客列表
最佳实践2026-06-1721 分钟阅读

角色化多Agent协作：从并行隔离到Explorer_Planner_Coder_Reviewer分工编排

Multi-Agent · 角色分工 · RoleOrchestrator · EventBus 隔离 · Token 预算控制

WeClaw_64_角色化多Agent协作：从并行隔离到Explorer_Planner_Coder_Reviewer分工编排

第四季系列文章第 3 篇（总第 64 篇） - Multi-Agent · 角色分工 · RoleOrchestrator · EventBus 隔离 · Token 预算控制

📚 专栏信息

《从零到一构建跨平台 AI 助手：WeClaw 实战指南》专栏 · 第四季

专栏定位：面向开发者和技术决策者的实战专栏，用真实案例和完整代码带你理解如何构建生产级 AI 应用

本文深入探讨 Harness Engineering 七大差距中的 G2——角色化多 Agent 协作。WeClaw 已有会话级 Agent 实例池，但当前是"并行隔离"模式——每个会话独立运行，没有角色分工。本文设计 Explorer/Planner/Coder/Reviewer 四角色协作体系，讲解 EventBus 隔离策略、SharedMemory 共享记忆、TokenBudgetController 预算控制，以及如何在不破坏现有 AgentPool 向后兼容性的前提下引入角色化。

👨‍💻 作者与项目

作者简介：翁勇刚 WENG YONGGANG 新概念龙虾-WeClaw 开发团队负责人，一群专注于跨平台 AI 应用的实践者 理念："再复杂的技术，也能用代码讲清楚"

📝 摘要

本文结构概览： 从 WeClaw 当前的"增强型单 Agent + 并行会话隔离"模式出发，分析为什么复杂任务需要角色分工——一个 Agent 同时负责分析需求、制定计划、编写代码和审查质量，就像一个人同时做产品经理、架构师、程序员和 QA。然后设计 RoleOrchestrator 编排器，引入 Explorer/Planner/Coder/Reviewer 四角色，讲解每个角色的工具权限、推理预算和协作流程。重点解决三个工程难题：EventBus 事件隔离、AgentPool 向后兼容、Token 预算控制。

核心问题： 当一个任务需要"先调研、再规划、再执行、最后审查"时，单个 Agent 需要在同一个 ReAct 循环中切换四种思维模式。这不仅是效率问题，更是认知负荷问题——模型在一次对话中难以同时保持"探索者"的开放性和"审查者"的批判性。

关键成果：

  • 设计了四角色协作体系，每个角色有独立的工具集、推理预算和 System Prompt
  • 通过独立 EventBus 实例 + SharedMemory 实现角色间安全通信
  • TokenBudgetController 硬上限 1.5x，防止成本失控

适合读者：构建多 Agent 系统的开发者、对 Agent 编排感兴趣的架构师

阅读时长：约 18 分钟

关键词：Multi-Agent、角色分工、RoleOrchestrator、EventBus 隔离、Token 预算、SharedMemory

一、为什么单个 Agent 不够？

1.1 认知负荷问题

考虑一个复杂任务："帮我分析项目中的安全漏洞，然后修复它们，最后验证修复效果"。

单个 Agent 需要在同一个 ReAct 循环中完成：

Step 1-5:   探索阶段 — 搜索代码、理解结构（需要开放性思维）
Step 6-8:   规划阶段 — 制定修复方案（需要系统性思维）
Step 9-20:  执行阶段 — 编写修复代码（需要精确性思维）
Step 21-25: 审查阶段 — 验证修复效果（需要批判性思维）

问题在于：模型在一次对话中难以同时保持"探索者"的开放性和"审查者"的批判性。 当 Agent 进入"执行者"模式后，它倾向于继续执行而非质疑——这导致审查阶段的深度不足。

1.2 上下文污染问题

更隐蔽的问题是上下文污染：

探索阶段的上下文："我发现 shell.py 有命令注入风险..."
规划阶段的上下文："修复方案是在 execute() 中添加参数过滤..."
执行阶段的上下文："已修改 shell.py 第 45 行..."
审查阶段的上下文："让我检查修改是否正确..."

到审查阶段时，Agent 已经在上下文中积累了大量"探索-规划-执行"的信息。这些信息会形成确认偏误——Agent 倾向于认为自己的执行是正确的，因为"我刚才做的"。

1.3 工具权限问题

单 Agent 模式下，所有工具对所有阶段可见。但理想情况下：

  • 探索阶段不应该有 shell.execute 权限（防止意外修改）
  • 审查阶段不应该有 file.write 权限（防止审查中修改代码）

角色分工的本质是：通过限制每个阶段的工具权限和上下文，降低认知负荷，提高每个阶段的执行质量。

二、四角色体系设计

2.1 角色定义

class AgentRole(str, Enum):
    GENERAL = "general"    # 默认角色，向后兼容
    EXPLORER = "explorer"  # 探索者
    PLANNER = "planner"     # 规划者
    CODER = "coder"         # 执行者
    REVIEWER = "reviewer"   # 审查者

@dataclass
class RoleConfig:
    """角色配置。"""
    role: AgentRole
    allowed_tools: set[str]        # 允许使用的工具集
    system_prompt_suffix: str      # 角色专属 Prompt
    max_steps: int                 # 最大推理步数
    preferred_model: str           # 首选模型
    max_tokens_budget: int         # Token 预算
    allowed_intents: list[str]     # 允许的意图类型

2.2 角色矩阵

角色职责允许的工具最大步数思维特征
Explorer分析需求、收集信息search, browser, file(只读), codebase_search10开放、发散
Planner分解任务、制定方案file(只读), codebase_search, search8系统、结构化
Coder编写代码、修改文件shell, file, screen, coding_assistant30精确、专注
Reviewer验证质量、检查错误shell, file(只读), screen, log_viewer10批判、严格

2.3 协作流程

用户输入
    │
    ▼
RoleOrchestrator.handle_task()
    │
    ├─ 1. 分析任务复杂度
    │     简单任务 → 单 Agent 快速路径（role=GENERAL）
    │     复杂任务 → 角色序列
    │
    ├─ 2. Explorer 阶段
    │     创建独立 Agent（独立 EventBus + 独立 session）
    │     执行探索 → 输出上下文摘要
    │
    ├─ 3. Planner 阶段
    │     接收 Explorer 的摘要（SharedMemory）
    │     执行规划 → 输出执行计划
    │
    ├─ 4. Coder 阶段
    │     接收 Planner 的计划
    │     执行编码 → 输出产物列表
    │
    └─ 5. Reviewer 阶段
          接收 Coder 的产物
          执行审查 → 输出审查报告
          （如发现问题 → 回到 Coder 阶段，最多 1 次返工）

三、EventBus 隔离 —— 代码级评审的关键发现

3.1 问题：EventBus 不支持子实例

V2 方案假设可以为每个角色 Agent 创建 "EventBus 子实例"。但代码级评审发现，WeClaw 的 EventBus 是一个简单的 pub-sub 实现，不支持子实例或独立通道。

# event_bus.py — 现有实现
class EventBus:
    def __init__(self):
        self._subscribers: dict[str, list[Callable]] = {}
    
    async def emit(self, event_type: str, data: Any = None) -> int:
        # 遍历该事件类型的所有订阅者并调用
        ...
    
    # ❌ 没有 create_sub_instance() 或 create_channel() 方法

如果所有角色 Agent 共享同一个 EventBus，Explorer 的事件会被 Coder 接收到——这会导致角色间的事件污染。

3.2 解决方案：独立 EventBus 实例

最简单也最安全的方案：为每个角色 Agent 创建全新的 EventBus() 实例。

class RoleOrchestrator:
    async def handle_task(self, user_input: str, parent_session_id: str) -> str:
        # Explorer 阶段
        explorer_bus = EventBus()  # ⚫ 独立实例
        explorer_agent = await self._agent_pool.get_or_create(
            session_id=f"{parent_session_id}_explorer",
            role=AgentRole.EXPLORER,
            role_config=self._role_configs[AgentRole.EXPLORER],
            event_bus=explorer_bus,  # 独立 EventBus
        )
        # ...

3.3 关键事件转发

角色完成后，RoleOrchestrator 将关键摘要事件转发到全局 EventBus：

async def handle_task(self, user_input: str, parent_session_id: str) -> str:
    # ... Explorer 执行完毕 ...
    
    # 转发关键事件到全局 EventBus
    await self._global_event_bus.emit(
        EventType.TASK_PHASE_CHANGE,
        TaskPhaseEvent(
            session_id=parent_session_id,
            phase="exploration_complete",
            role="explorer",
        )
    )
    
    # 将 Explorer 的摘要存入 SharedMemory
    shared_memory = SharedMemory(
        parent_session_summary=explorer_result,
        artifacts=[],
        decisions=[],
        errors=[],
    )
    
    # ... 进入 Planner 阶段 ...

四、SharedMemory —— 角色间的安全通信

4.1 设计

角色之间不直接交换消息，而是通过 SharedMemory 传递只读上下文摘要：

@dataclass
class SharedMemory:
    """角色间共享的只读上下文摘要。"""
    parent_session_summary: str   # 上一角色的执行摘要
    artifacts: list[str]          # 产生的文件/产物列表
    decisions: list[str]         # 关键决策记录
    errors: list[str]            # 遇到的错误

4.2 为什么不直接传递完整上下文？

直接传递完整上下文有两个问题：

  1. Token 爆炸：Explorer 阶段可能消耗 5000+ token 的上下文，全部传给 Planner 会导致预算超限
  2. 认知污染：Planner 不需要知道 Explorer 的每一步搜索细节，只需要知道"发现了什么"

SharedMemory 的本质是信息压缩——每个角色只接收上一个角色的"结论摘要"，而非"过程日志"。

4.3 传递方式

# Planner 接收 Explorer 的摘要
planner_prompt = f"""
## 上一阶段（探索）的结果摘要

{shared_memory.parent_session_summary}

## 发现的产物
{chr(10).join(f'- {a}' for a in shared_memory.artifacts)}

## 请基于以上信息制定执行计划。
"""

五、AgentPool 向后兼容 —— 不破坏现有系统

5.1 兼容性挑战

WeClaw 的 AgentPool.get_or_create() 方法已有稳定的调用方（桌面端、PWA 远程端）。任何修改都不能影响现有行为。

5.2 解决方案：默认值兼容

async def get_or_create(
    self,
    session_id: str,
    source: str = "desktop",
    metadata: Optional[dict] = None,
    # ↓ 新增参数，全部有默认值 ↓
    role: AgentRole = AgentRole.GENERAL,
    role_config: Optional[RoleConfig] = None,
    event_bus: Optional[EventBus] = None,
) -> Agent:
    # ... 现有逻辑 ...
    
    agent = Agent(
        model_registry=self._model_registry,
        tool_registry=self._tool_registry,
        event_bus=event_bus or self._event_bus,  # 角色用独立，默认用共享
        # ...
        max_steps=role_config.max_steps if role_config else self._default_max_steps,
    )
    
    # 角色工具过滤（仅 role != GENERAL 时生效）
    if role_config and role != AgentRole.GENERAL:
        agent.tool_exposure.set_role_filter(role_config.allowed_tools)
        agent.system_prompt += f"\n\n{role_config.system_prompt_suffix}"

关键设计：当 role=GENERAL（默认值）时，不传任何额外参数，行为与现有完全一致。现有调用方无需任何修改。

5.3 验证策略

# 向后兼容测试
def test_general_role_backward_compatible():
    """GENERAL 角色行为必须与无角色参数完全一致。"""
    pool = AgentPool(...)
    
    # 旧方式（不传 role 参数）
    agent_old = await pool.get_or_create("session_1", "desktop")
    
    # 新方式（显式传 role=GENERAL）
    agent_new = await pool.get_or_create("session_2", "desktop", 
                                          role=AgentRole.GENERAL)
    
    # 两者行为一致
    assert agent_old.max_steps == agent_new.max_steps
    assert agent_old.event_bus is agent_new.event_bus  # 都用共享 EventBus
    assert not hasattr(agent_old, '_role_filter')
    assert not hasattr(agent_new, '_role_filter')

六、TokenBudgetController —— 成本控制

6.1 问题：多 Agent 的成本爆炸

四角色协作意味着 4 个 Agent 各自消耗 token。如果不控制，总成本可能是单 Agent 的 4 倍。

6.2 解决方案：1.5 倍硬上限

class TokenBudgetController:
    """多 Agent 协作场景的 token 预算控制。
    总预算不超过单 Agent 的 1.5 倍。
    """
    
    def __init__(self, single_agent_budget: int, multiplier: float = 1.5):
        self._total_budget = int(single_agent_budget * multiplier)
        self._consumed = 0
    
    def consume(self, tokens: int, role: AgentRole) -> bool:
        """消耗 token 预算，返回是否还在预算内。"""
        self._consumed += tokens
        if self._consumed > self._total_budget:
            logger.warning(
                "Token 预算超限: %d/%d (role=%s)",
                self._consumed, self._total_budget, role.value
            )
            return False
        return True
    
    def get_remaining(self) -> int:
        return max(0, self._total_budget - self._consumed)

6.3 超预算降级

当 token 接近上限时，自动降级到更便宜的模型：

# 超过 80% 预算时，后续角色使用经济模型
if budget_controller.get_remaining() < total_budget * 0.2:
    model_key = "qwen-turbo"  # 经济模型
else:
    model_key = role_config.preferred_model  # 首选模型

6.4 为什么是 1.5 倍而不是 4 倍？

四角色不等于 4 倍 token，因为：

  1. SharedMemory 压缩：每个角色只接收上一个角色的摘要，不是完整上下文
  2. 工具权限收窄：角色只能用少量工具，减少了工具描述的 token 开销
  3. 步数限制：Explorer 最多 10 步、Planner 最多 8 步，远少于单 Agent 的 30 步

1.5 倍是一个经验值——足以完成复杂任务，又不会让成本失控。

七、简单任务的快速路径

7.1 不是所有任务都需要四角色

用户："今天天气怎么样？"
→ 不需要 Explorer/Planner/Coder/Reviewer
→ 走单 Agent 快速路径

7.2 复杂度判断

async def handle_task(self, user_input: str, parent_session_id: str) -> str:
    # 1. 分析任务复杂度
    complexity = await self._analyze_complexity(user_input)
    
    if complexity < self._complexity_threshold:
        # 简单任务：单 Agent 快速路径
        return await self._single_agent_path(user_input, parent_session_id)
    else:
        # 复杂任务：角色化路径
        return await self._role_based_path(user_input, parent_session_id)

复杂度判断可以基于：

  • 输入长度（>100 字符可能是复杂任务）
  • 意图分类（code_modify 类意图更可能需要角色分工）
  • 历史模式（类似任务过去的工具调用步数）

八、核心教训

8.1 EventBus 隔离不是可选项

V2 方案假设可以"在共享 EventBus 上创建子通道"，代码级评审证明这不可行。如果所有角色共享一个 EventBus，事件会跨角色泄漏——Explorer 的搜索事件被 Coder 接收，Coder 以为有新的搜索任务要执行。

教训：在 pub-sub 架构中，隔离的唯一可靠方式是独立实例。

8.2 向后兼容是设计约束，不是事后补丁

我们一开始就为 get_or_create() 的新参数设置了默认值（role=GENERAL），确保现有调用方零改动。如果等到角色化功能开发完再考虑兼容性，很可能需要重构调用方——那就不是"灰度上线"了。

8.3 Token 预算比功能更难控制

四角色的功能设计相对直观——每个角色做什么很清楚。但 token 预算控制是一个持续的工程挑战：SharedMemory 压缩到什么程度？摘要由谁生成（LLM 还是规则）？超预算时降级到什么模型？这些问题没有标准答案，需要在实际运行中持续调整。

📖 相关文章

本文是 WeClaw 专栏第四季的第 3 篇（总第 64 篇）。如果这篇文章对你有帮助，欢迎给项目点个 Star ⭐

上一篇确定性约束引擎：从Prompt引导到代码级强制的范式转变
下一篇熵管理：AI系统的垃圾回收机制