跳转到内容

Tool 注册表与推荐系统

学习目标

理解 Tool Registry 如何管理工具注册,BM25 推荐系统如何根据上下文筛选工具,以及工具执行映射的实现。


项目实践

工具注册表

ToolRegistry 是一个单例,管理所有注册的工具:

class ToolRegistry:
tools: dict[str, Tool] = {}
def register_tool(self, tool: Tool):
self.tools[tool.name] = tool
def get_tool(self, name: str) -> Tool: ...
def get_all_tools(self) -> list[Tool]: ...

注册装饰器:

def register_tool(include_functions: list[str] = None):
def decorator(cls):
# 1. 提取类的 docstring 作为工具描述
# 2. 提取 include_functions 中指定的方法 schema
# 3. 创建 Tool 对象并注册到 ToolRegistry
tool = Tool(name=cls.__name__, description=..., schemas=...)
ToolRegistry().register_tool(tool)
return cls
return decorator

使用方式:

@register_tool(include_functions=["run"])
class WritePRD(Action):
async def run(self, ...) -> ...: ...

这会将 WritePRD.run 方法注册为一个可用工具。

工具数据结构

class ToolParam(BaseModel):
name: str # 参数名
description: str # 参数描述
type: str # 参数类型
required: bool # 是否必填
class ToolSchema(BaseModel):
name: str # 方法名
description: str # 方法描述
parameters: list[ToolParam] # 参数列表
return_type: str # 返回类型
class Tool(BaseModel):
name: str # 工具名
description: str # 工具描述
schemas: dict[str, ToolSchema] # 方法 schema 映射

BM25ToolRecommender

BM25 是信息检索中的经典算法,根据查询词与文档的相关性排序。在 MetaGPT 中:

class BM25ToolRecommender(ToolRecommender):
def __init__(self, tools: list[str], force: bool = False):
# tools: 可用工具列表
# force: True = 推荐所有工具,False = 按相关性筛选
self.bm25 = BM25Okapi(corpus=[tool.description for tool in tools])
async def recommend_tools(self, query: str = None) -> list[Tool]:
# 1. 使用 query 或 working memory 最新内容作为查询
# 2. 用 BM25 计算每个工具的相关性
# 3. 返回相关性高于阈值的工具
if self.force:
return self.tools
scores = self.bm25.get_scores(query)
return [tool for tool, score in zip(self.tools, scores) if score > threshold]

推荐时机

在 RoleZero 的 _think() 中:

tools = await self.tool_recommender.recommend_tools()
tool_info = json.dumps({tool.name: tool.schemas for tool in tools})

这比将所有工具 schema 注入 prompt 节省大量 token。

工具执行映射

RoleZero 维护 tool_execution_map:字符串命令到 callable 的映射:

self.tool_execution_map = {
"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,
}

子类可以扩展:

# Architect 添加终端命令执行
def _update_tool_execution(self):
self.tool_execution_map.update({"Terminal.run_command": self.terminal.run_command})
# ProductManager 添加 PRD 生成
def _update_tool_execution(self):
wp = WritePRD()
self.tool_execution_map.update(tool2name(WritePRD, ["run"], wp.run))

内置工具的自动注册:

# Browser 工具
self.tool_execution_map.update({
f"Browser.{i}": getattr(self.browser, i)
for i in ["click", "close_tab", "go_back", "goto", "hover", "press", "scroll", ...]
})
# Editor 工具
self.tool_execution_map.update({
f"Editor.{i}": getattr(self.editor, i)
for i in ["append_file", "create_file", "edit_file_by_replace", "open_file", "read", ...]
})

工具推荐信息传递

在 DataInterpreter 的 _write_and_exec_code() 中:

if self.tool_recommender:
context = self.working_memory.get()[-1].content # _think 阶段的 thoughts
plan = self.planner.plan if self.use_plan else None
tool_info = await self.tool_recommender.get_recommended_tool_info(context, plan)
else:
tool_info = ""
# 传递给 WriteAnalysisCode
code = await WriteAnalysisCode().run(
user_requirement=self.user_requirement,
plan_status=plan_status,
tool_info=tool_info,
...
)

get_recommended_tool_info() 将推荐工具的 schema 格式化为 LLM 可读的文本。

工具声明格式

Role 通过 tools 列表声明需要的工具:

class ProductManager(RoleZero):
tools: list[str] = [
"RoleZero",
"Browser",
"Editor",
"SearchEnhancedQA",
]
class Architect(RoleZero):
tools: list[str] = [
"Editor:write,read,similarity_search", # 细粒度控制
"RoleZero",
"Terminal:run_command",
]
  • "ToolName":注册整个工具的所有方法
  • "ToolName:method1,method2":只注册指定方法
  • "<all>":注册所有可用工具(特殊通配符)

问题与规避

BM25 推荐不准确

  • BM25 基于关键词匹配,不理解语义
  • 如果查询与工具描述关键词不匹配,可能遗漏相关工具
  • 对策:确保工具 docstring 包含丰富的关键词,或使用语义 embedding 替代 BM25

force=True 导致上下文膨胀

  • force=True 时推荐所有工具,不管相关性
  • 对于工具数量多的场景会导致 prompt 过长
  • 对策:只在工具较少时使用 force=True,或增加工具描述的精简度

工具注册冲突

  • 如果两个类注册同名工具,后注册的会覆盖先注册的
  • 对策:使用类名作为工具名(cls.__name__),避免冲突

设计取舍

BM25 vs Embedding 推荐

  • BM25:轻量、无需 LLM、基于关键词匹配
  • Embedding:语义理解更好,但需要向量模型
  • MetaGPT 选择 BM25 作为默认,因为工具描述通常短小,关键词匹配效果好

声明式 vs 自动发现

  • 声明式:Role 通过 tools 列表声明需要的工具
  • 自动发现:运行时扫描所有注册工具
  • 声明式的好处:可控、可预测的 token 消耗、避免推荐不相关工具

参考来源

  • 源码验证: metagpt/tools/tool_registry.py — ToolRegistry 和 register_tool 装饰器
  • 源码验证: metagpt/tools/tool_recommend.py — BM25ToolRecommender
  • 源码验证: metagpt/tools/tool_data_type.py — Tool/ToolSchema/ToolParam 数据结构
  • 源码验证: metagpt/roles/di/role_zero.py:119set_tool_execution() 默认映射
  • 源码验证: metagpt/roles/di/role_zero.py:218_think() 中的工具推荐