RoleZero：动态思考与工具执行

学习目标

理解 RoleZero 作为新一代通用 Agent 基类的设计，包括三层意图路由、工具执行、经验检索和快速思考机制。

---

项目实践

RoleZero 概览

RoleZero 是 ProductManager、Architect 等角色的新基类，相比传统 Role 增加了：

• 动态思考-行动循环：不依赖预定义 action 列表，由 LLM 动态决定命令
• 工具推荐：BM25 推荐相关工具，减少 LLM 上下文
• 经验检索：从经验池中检索相似上下文，注入 prompt
• 三层意图路由：Quick Think → Think → Act 渐进式处理
• 工具执行映射：字符串命令到 callable 的映射表

class RoleZero(Role):
    name: str = "Zero"
    react_mode: Literal["react"] = "react"   # 固定为 react
    max_react_loop: int = 50                  # 较大循环上限
    tools: list[str] = []                     # 可用工具
    tool_execution_map: dict[str, Callable] = {}  # 命令映射
    experience_retriever: ExpRetriever = DummyExpRetriever()
    memory_k: int = 200                       # 历史消息数量
    use_fixed_sop: bool = False               # 是否使用固定 SOP

三层意图路由

RoleZero 的 _react() 实现了三层路由：

async def _react(self) -> Message:
    # 1. 设置 todo，允许处理新消息
    self._set_state(0)

    # 2. 快速思考：尝试用简单方式回答
    quick_rsp, _ = await self._quick_think()
    if quick_rsp:
        return quick_rsp

    # 3. 正式 think-act 循环
    actions_taken = 0
    while actions_taken < self.rc.max_react_loop:
        await self._observe()
        has_todo = await self._think()
        if not has_todo:
            break
        rsp = await self._act()
        actions_taken += 1

        # 超过 10 次时询问人类是否继续
        if actions_taken >= self.rc.max_react_loop:
            human_rsp = await self.ask_human("...")
            if "yes" in human_rsp.lower():
                actions_taken = 0
    return rsp

Quick Think：意图分类

_quick_think() 先做意图分类，避免不必要的复杂处理：

async def _quick_think(self) -> Tuple[Message, str]:
    # 1. 格式化最近的消息为上下文
    memory = self.get_memories(k=self.memory_k)
    context = self.llm.format_msg(memory + [UserMessage(content=QUICK_THINK_PROMPT)])

    # 2. 意图分类
    intent_result = await self.llm.aask(context, system_msgs=[...])

    if "QUICK" in intent_result or "AMBIGUOUS" in intent_result:
        # 直接回答简单问题
        answer = await self.llm.aask(...)
    elif "SEARCH" in intent_result:
        # 调用搜索引擎
        answer = await SearchEnhancedQA().run(query)

    if answer:
        self.rc.memory.add(AIMessage(content=answer, cause_by=QUICK_THINK_TAG))
        await self.reply_to_human(content=answer)
        return rsp_msg, intent_result
    return None, intent_result

意图类型：

意图	处理方式	示例
QUICK	直接 LLM 回答	”你好”、“现在几点”
AMBIGUOUS	直接 LLM 回答	模糊的问题
SEARCH	调用搜索引擎	”今天天气怎么样”
TASK	进入正式 think-act 循环	”创建一个 snake game”

Think：动态命令生成

_think() 是 RoleZero 的核心决策步骤：

async def _think(self) -> bool:
    # 0. 准备工作
    if not self.planner.plan.goal:
        self.planner.plan.goal = self.get_memories()[-1].content
        self.respond_language = await self.llm.aask(DETECT_LANGUAGE_PROMPT)

    # 1. 经验检索
    example = self._retrieve_experience()

    # 2. 计划状态
    plan_status, current_task = get_plan_status(planner=self.planner)

    # 3. 工具信息
    tools = await self.tool_recommender.recommend_tools()
    tool_info = json.dumps({tool.name: tool.schemas for tool in tools})

    # 4. 构建 system prompt
    system_prompt = self.system_prompt.format(
        role_info=self._get_prefix(),
        available_commands=tool_info,
        example=example,
        instruction=self.instruction.strip(),
    )

    # 5. 构建 user prompt
    prompt = self.cmd_prompt.format(
        plan_status=plan_status,
        current_task=current_task,
        respond_language=self.respond_language,
    )

    # 6. 处理历史消息（解析浏览器/编辑器结果、图片）
    memory = self.rc.memory.get(self.memory_k)
    memory = await parse_browser_actions(memory, browser=self.browser)
    memory = await parse_editor_result(memory)
    memory = await parse_images(memory, llm=self.llm)

    # 7. 调用 LLM 生成命令
    self.command_rsp = await self.llm_cached_aask(req=req, system_msgs=[system_prompt])
    return True

Act：命令解析与执行

_act() 解析 LLM 生成的命令字符串并执行：

async def _act(self) -> Message:
    # 1. 解析命令
    commands, ok, self.command_rsp = await parse_commands(
        command_rsp=self.command_rsp,
        llm=self.llm,
        exclusive_tool_commands=self.exclusive_tool_commands,
    )
    self.rc.memory.add(AIMessage(content=self.command_rsp))

    if not ok:
        error_msg = commands
        self.rc.memory.add(UserMessage(content=error_msg, cause_by=RunCommand))
        return error_msg

    # 2. 执行命令
    outputs = await self._run_commands(commands)

    # 3. 结果写入 memory
    self.rc.memory.add(UserMessage(content=outputs, cause_by=RunCommand))
    return AIMessage(content=f"... Outputs: {outputs}", cause_by=RunCommand)

_run_commands() 遍历命令并执行：

async def _run_commands(self, commands) -> str:
    outputs = []
    for cmd in commands:
        if self._is_special_command(cmd):
            output = await self._run_special_command(cmd)
            continue
        if cmd["command_name"] in self.tool_execution_map:
            tool_obj = self.tool_execution_map[cmd["command_name"]]
            tool_output = await tool_obj(**cmd["args"])
            outputs.append(f"{cmd['command_name']}: {tool_output}")
        else:
            outputs.append(f"Command {cmd['command_name']} not found.")
            break
    return "\n\n".join(outputs)

默认工具映射

RoleZero 默认注册以下工具：

工具	方法	用途
Plan.append_task	self.planner.plan.append_task	添加任务
Plan.reset_task	self.planner.plan.reset_task	重置任务
Plan.replace_task	self.planner.plan.replace_task	替换任务
RoleZero.ask_human	self.ask_human	询问人类
RoleZero.reply_to_human	self.reply_to_human	回复人类
Browser.*	self.browser.*	浏览器操作（click/scroll/type 等）
Editor.*	self.editor.*	编辑器操作（open/read/write 等）
SearchEnhancedQA.run	SearchEnhancedQA().run	搜索（如果启用）

Special Commands

特殊命令需要额外处理：

命令	处理
Plan.finish_current_task	标记当前任务完成，切换到下一个
end	结束执行，生成摘要，回复人类
RoleZero.ask_human	向人类提问，收到回复后继续或停止
Terminal.run_command	终端命令执行，输出格式化

Experience Retrieval

def _retrieve_experience(self) -> str:
    """Default implementation of experience retrieval."""
    context = [str(msg) for msg in self.rc.memory.get(self.memory_k)]
    context = "\n\n".join(context)
    example = self.experience_retriever.retrieve(context=context)
    return example

子类可以覆盖 _retrieve_experience() 来注入特定经验：

# Architect 注入架构设计示例
def _retrieve_experience(self) -> str:
    return ARCHITECT_EXAMPLE

LLM 缓存与经验池

llm_cached_aask() 使用 exp_cache 装饰器自动管理经验：

@exp_cache(context_builder=RoleZeroContextBuilder(), serializer=RoleZeroSerializer())
async def llm_cached_aask(self, *, req: list[dict], system_msgs: list[str], **kwargs) -> str:
    return await self.llm.aask(req, system_msgs=system_msgs)

• RoleZeroContextBuilder：尝试将相似经验注入 prompt
• RoleZeroSerializer：从请求中提取关键部分用于经验池
• 自动裁剪过长条目，保留必要上下文

---

问题与规避

exclusive_tool_commands 去重

• 编辑操作（如 Editor.edit_file_by_replace）如果重复执行会导致文件内容混乱
• exclusive_tool_commands 列表确保只保留第一次出现
• 注意：这只在命令解析层去重，不防止 LLM 生成重复命令

max_react_loop 人类确认

• max_react_loop >= 10 时，达到上限后会询问人类是否继续
• 在 MGXEnv 中通过 ask_human() 实现
• 如果不在 MGXEnv 中，ask_human() 返回固定字符串

语言检测

• DETECT_LANGUAGE_PROMPT 让 LLM 判断用户需求语言，确保回复使用相同语言
• 如果用户需求是中文，respond_language 会被设为 “Chinese”

---

设计取舍

RoleZero vs 传统 Role

• 传统 Role：预定义 actions 列表，by_order 模式按顺序执行，适合固定 SOP
• RoleZero：动态生成命令，适合灵活场景
• ProductManager/Architect 虽然继承 RoleZero，但设置 use_fixed_sop=True 回退到 by_order 模式
• 这保留了 RoleZero 的工具系统和经验检索，同时维持了固定 SOP 的确定性

动态命令 vs 预定义 Action

• 动态命令由 LLM 生成，灵活性高但不可预测
• 预定义 Action 由代码定义，可控但缺乏灵活性
• RoleZero 折中：命令格式由 parse_commands() 约束，执行逻辑由 tool_execution_map 定义

---

参考来源

• 源码验证: metagpt/roles/di/role_zero.py:55 — RoleZero 类定义
• 源码验证: metagpt/roles/di/role_zero.py:303 — _react() 三层路由
• 源码验证: metagpt/roles/di/role_zero.py:342 — _quick_think() 意图分类
• 源码验证: metagpt/roles/di/role_zero.py:198 — _think() 动态思考
• 源码验证: metagpt/roles/di/role_zero.py:280 — _act() 命令执行
• 源码验证: metagpt/roles/di/role_zero.py:119 — set_tool_execution() 默认映射
• 源码验证: metagpt/roles/di/role_zero.py:267 — llm_cached_aask() 经验池