Back to Blog
Tech Tutorials2026-03-268 min read

Tool Registration System Evolution: From Manual Mapping to Config-Driven Auto-Discovery

The complete evolution of the tool registration system from hardcoded mappings to tools.json config-driven auto-discovery.

WeClaw 工具注册系统演进:从手动映射到配置驱动自动发现的架构之路

系列文章第 24 篇 - 从"每加一个工具改五处代码"到"配置一行自动接入"的架构演进之旅


📚 专栏信息

《从零到一构建跨平台 AI 助手:WeClaw 实战指南》专栏

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

本系列共 25 篇,分为八大模块

📖 模块一【通讯架构设计】(3 篇):混合通讯、设备绑定、请求路由
🔧 模块二【核心技术实现】(4 篇):WebSocket 路由、心跳重连、离线队列
🛡️ 模块三【安全与治理】(3 篇):密钥管理、Token 吊销、速率限制
🔍 模块四【调试与监控】(2 篇):全链路追踪、日志分析
💡 模块五【问题诊断实战】(3 篇):典型问题排查与修复
⚙️ 模块六【性能优化】(1 篇):启动速度、内存优化
🤖 模块七【主动陪伴系统】(3 篇):决策引擎、防骚扰机制、渐进式建档
🛠️ 模块八【工具扩展架构】(2 篇):工具注册系统、可选依赖处理

  • 模块定位:工具扩展架构 · 第 1 篇(共 2 篇)
  • 前置知识:Python 基础、动态导入(importlib)、JSON 配置
  • 关联文章:第 25 篇(可选依赖的优雅处理)

👨‍💻 作者与项目

作者简介:翁勇刚 WENG YONGGANG
新概念龙虾-WeClaw 开发团队负责人,一群专注于跨平台 AI 应用的实践者
理念:"让工具扩展像添加配置一样简单,让开发者专注于业务逻辑"


📝 摘要

本文结构概览: 本文首先从"每新增一个工具要改五个文件"的真实痛点出发,分析为什么需要配置驱动的工具注册系统;然后用"餐厅菜单管理"比喻讲解 BaseTool 抽象基类和 ToolRegistry 注册中心的协作原理;接着通过 469 行核心代码详解五个关键方法(load_config、auto_discover、_preload_tool_metadata、_build_init_kwargs、get_all_schemas)的实现;随后还原一次"某工具 import 失败导致启动报错"的问题排查过程;最后给出 Schema 缓存和错误容错等性能优化策略。

背景:在 WeClaw 从 7 个工具扩展到 38+ 工具的过程中,我们遇到了严重的可维护性问题:每新增一个工具,开发者需要修改 5-6 个文件,容易遗漏、容易出错。

核心问题:如何设计一套工具注册系统,让新增工具只需"声明配置"而非"到处改代码"?如何在保证类型安全的同时实现动态发现?

解决方案:采用配置驱动 + 自动发现的三层架构。tools.json 声明工具元信息,ToolRegistry 动态导入并实例化,BaseTool 定义统一接口契约。

关键成果

  • 新增工具从"改 5 个文件"简化为"改 2 个文件"(工具类 + tools.json)
  • 支持 38+ 工具、170+ Actions 的规模化管理
  • Schema 缓存使重复查询耗时降低 95%
  • 单个工具加载失败不影响其他工具

适合读者:有 Python 基础,对插件架构、依赖注入、配置驱动设计感兴趣的开发者

阅读时长:约 20 分钟

关键词ToolRegistryBaseTooltools.jsonauto_discover懒加载Schema 缓存动态导入


一、为什么需要工具注册系统?——从"改五个文件"的痛苦说起

1.1 场景重现:Phase 6 的扩展噩梦

想象这个开发场景:

产品经理说:"我们需要新增 16 个工具,包括合同生成、财务报告、简历生成、思维导图..."

作为开发者,你的第一反应是什么?

在 WeClaw 早期版本,新增一个工具意味着:

┌─────────────────────────────────────────────────────────────────────┐
│  新增一个工具的"五步曲"(噩梦版)                                      │
├─────────────────────────────────────────────────────────────────────┤
│  步骤 1:创建 src/tools/{name}.py                   ✅ 必须的        │
│  步骤 2:在 registry.py 手动 import 并注册           ❌ 容易遗漏      │
│  步骤 3:在 registry.py _build_init_kwargs 添加参数  ❌ 容易遗漏      │
│  步骤 4:在 prompts.py 添加意图关键词映射            ⚠️ 容易出错      │
│  步骤 5:在 tool_exposure.py 添加工具暴露配置        ⚠️ 容易出错      │
│  步骤 6:更新文档说明                               😅 经常忘记      │
└─────────────────────────────────────────────────────────────────────┘

真实数据:Phase 6 需要新增 16 个工具,按旧流程意味着:

  • 修改 registry.py 32 次(import + register)
  • 修改 prompts.py 16 次
  • 人工检查 80+ 处代码改动

结果

  • 花了 3 天时间才把 16 个工具全部接入
  • 上线后发现 2 个工具忘记注册,1 个工具参数映射错误
  • 团队士气低落:"以后谁还敢加工具?"

1.2 方案对比:四种工具管理策略

让我们对比四种工具管理方案:

方案像什么?(比喻)优点缺点适用场景
硬编码注册手抄菜单,新菜手写上去简单直接,一眼看清改动多,容易遗漏工具 < 5 个
if-else 路由服务员背菜单,点啥找啥逻辑清晰代码膨胀,N 个工具 N 个分支工具 < 10 个
配置驱动菜单印在本子上,看本点菜改配置不改代码需要解析器工具 10~50 个
自动发现 + 懒加载智能菜单系统,扫码即可零改动扩展,按需加载架构复杂工具 50+ 个
# ❌ 错误示范 1:硬编码注册
class BadHardcodedRegistry:
    def __init__(self):
        # 问题:每新增一个工具,这里就要加一行
        self.tools = {
            "shell": ShellTool(),
            "file": FileTool(),
            "screen": ScreenTool(),
            # ... 想象这里有 38 行 ...
        }
# ❌ 错误示范 2:if-else 路由
def bad_call_tool(tool_name: str, action: str, params: dict):
    # 问题:N 个工具就有 N 个 elif 分支
    if tool_name == "shell":
        return ShellTool().execute(action, params)
    elif tool_name == "file":
        return FileTool().execute(action, params)
    elif tool_name == "screen":
        return ScreenTool().execute(action, params)
    # ... 38 个 elif 分支 ...
    else:
        raise ValueError(f"未知工具: {tool_name}")
# ✅ 正确做法:配置驱动 + 自动发现
class GoodConfigDrivenRegistry:
    def __init__(self):
        self._tools: dict[str, BaseTool] = {}
    
    def load_config(self, config_path: Path) -> None:
        """从 JSON 配置加载工具定义"""
        with open(config_path, "r", encoding="utf-8") as f:
            config = json.load(f)
        self._tool_configs = config.get("tools", {})
    
    def auto_discover(self) -> None:
        """根据配置自动发现并注册工具"""
        for tool_name, cfg in self._tool_configs.items():
            if not cfg.get("enabled", True):
                continue
            # 动态导入 + 实例化
            mod = importlib.import_module(cfg["module"])
            cls = getattr(mod, cfg["class"])
            self._tools[tool_name] = cls()

1.3 核心挑战:如何让新增工具只需"声明"?

要实现真正的"配置即扩展",我们需要解决三个核心问题:

挑战 1:统一接口契约

不同工具有不同的 action(shell 有 run,file 有 read/write/edit),如何让 Registry 统一调用?

挑战 2:构造参数差异

shell 需要 timeout 和 blacklist,file 需要 max_read_size,如何处理不同工具的初始化参数?

挑战 3:错误隔离

如果 browser 工具依赖的 Playwright 未安装,不能导致整个系统启动失败。

这就引出了我们的 BaseTool + ToolRegistry 架构——用抽象基类定义契约,用注册中心统一调度。


二、理解 BaseTool + ToolRegistry 架构

2.1 什么是 BaseTool 抽象基类?

官方定义

BaseTool 是 WeClaw 工具系统的抽象基类,定义了所有工具必须实现的统一接口:get_actions() 返回支持的动作列表,execute() 执行具体动作。

大白话解释: 想象你开一家连锁餐厅,每家分店的菜品不同,但点餐流程必须统一:

┌─────────────────────────────────────────────────────────────────────┐
│                      餐厅菜单系统比喻                                 │
├─────────────────────────────────────────────────────────────────────┤
│  BaseTool         = 点餐标准流程     "看菜单 → 下单 → 上菜"           │
│  get_actions()    = 菜单列表         "本店有哪些菜?"                  │
│  execute()        = 厨师做菜         "收到订单,开始烹饪"              │
│  get_schema()     = 菜品详情卡       "这道菜的配料、价格、口味..."     │
│  ToolRegistry     = 餐饮管理公司     "管理所有分店,统一调度"          │
│  tools.json       = 分店花名册       "哪些分店开业、地址在哪..."       │
└─────────────────────────────────────────────────────────────────────┘

2.2 BaseTool 的核心接口

来看 BaseTool 的关键代码:

# src/tools/base.py

class BaseTool(ABC):
    """工具基类。所有工具必须继承此类。"""

    name: str = ""                    # 工具唯一标识
    emoji: str = "🔧"                  # 显示图标
    title: str = ""                   # 显示名称
    description: str = ""             # 工具描述
    timeout: float = DEFAULT_TOOL_TIMEOUT  # 超时时间

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """子类初始化钩子:自动生成 name"""
        super().__init_subclass__(**kwargs)
        if not cls.name:
            # ✅ 自动命名:ShellTool → shell
            cls.name = cls.__name__.lower().replace("tool", "")

    @abstractmethod
    def get_actions(self) -> list[ActionDef]:
        """返回此工具支持的所有动作定义。
        
        Returns:
            ActionDef 列表,每个 ActionDef 包含:
            - name: 动作名称
            - description: 动作描述
            - parameters: JSON Schema 格式的参数定义
            - required_params: 必填参数列表
        """
        ...

    @abstractmethod
    async def execute(self, action: str, params: dict[str, Any]) -> ToolResult:
        """执行指定动作。
        
        Args:
            action: 动作名称(必须是 get_actions() 返回的某个动作)
            params: 动作参数(符合 ActionDef.parameters 定义)
            
        Returns:
            ToolResult 执行结果
        """
        ...

设计亮点

  1. 自动命名__init_subclass__ 钩子自动将 ShellTool 转为 shell
  2. 双抽象方法get_actions() + execute() 形成"声明-执行"分离
  3. 类型安全:使用 ActionDef 数据类定义参数 Schema

2.3 ToolRegistry 的三层职责

ToolRegistry 承担三层职责,让我们用架构图理解:

                           ┌─────────────────────────────────┐
                           │         tools.json              │
                           │  ┌─────────────────────────┐    │
                           │  │  "shell": {              │    │
                           │  │    "module": "src..."    │    │
                           │  │    "class": "ShellTool"  │    │
                           │  │    "enabled": true       │    │
                           │  │  }                       │    │
                           │  └─────────────────────────┘    │
                           └───────────────┬─────────────────┘
                                           │
                    ┌──────────────────────┼──────────────────────┐
                    │                      ▼                      │
                    │   ┌─────────────────────────────────────┐   │
                    │   │      第一层:配置加载                 │   │
                    │   │      load_config()                   │   │
                    │   │  • 读取 tools.json                   │   │
                    │   │  • 解析 global_settings              │   │
                    │   │  • 缓存 tool_configs                 │   │
                    │   └─────────────────┬───────────────────┘   │
                    │                     │                       │
                    │                     ▼                       │
                    │   ┌─────────────────────────────────────┐   │
                    │   │      第二层:自动发现                 │   │
                    │   │      auto_discover()                 │   │
                    │   │  • 遍历 tool_configs                 │   │
                    │   │  • importlib.import_module()         │   │
                    │   │  • getattr(mod, class_name)          │   │
                    │   │  • _build_init_kwargs() 构建参数     │   │
                    │   │  • 实例化并注册                       │   │
                    │   └─────────────────┬───────────────────┘   │
                    │                     │                       │
                    │                     ▼                       │
                    │   ┌─────────────────────────────────────┐   │
                    │   │      第三层:元信息缓存               │   │
                    │   │      _preload_tool_metadata()        │   │
                    │   │  • 预加载 actions 定义               │   │
                    │   │  • 构建 func_map 映射                │   │
                    │   │  • Schema 缓存                       │   │
                    │   └─────────────────────────────────────┘   │
                    │                                             │
                    │               ToolRegistry                  │
                    └─────────────────────────────────────────────┘
                                           │
                                           ▼
                    ┌─────────────────────────────────────────────┐
                    │              已注册工具池                     │
                    │  ┌─────────┐ ┌─────────┐ ┌─────────┐       │
                    │  │  shell  │ │  file   │ │ screen  │  ...  │
                    │  └─────────┘ └─────────┘ └─────────┘       │
                    └─────────────────────────────────────────────┘

2.4 对比表:手动注册 vs 配置驱动

维度手动注册配置驱动区别说明
新增工具改 5 个文件改 2 个文件tools.json + 工具类文件
启用/禁用注释代码enabled: false零代码改动
参数配置硬编码在代码里写在 JSON 配置里支持热更新
错误隔离一个失败全挂单个失败不影响其他try-except 包裹
类型检查编译时检查运行时检查需要额外验证

为什么选择配置驱动?

因为 WeClaw 的工具数量已经超过 38 个,且还在持续增长。配置驱动让:

  • 开发者:只需关注工具本身的业务逻辑
  • 运维人员:可以通过配置启用/禁用工具,无需重新部署
  • 测试人员:可以针对单个工具进行隔离测试

三、代码详解——五个关键方法

3.1 load_config():从 JSON 到内存的第一步

# src/tools/registry.py

def load_config(self, config_path: Path | None = None) -> None:
    """从 JSON 配置文件加载工具定义。

    Args:
        config_path: 配置文件路径,默认 config/tools.json
    """
    path = config_path or _DEFAULT_TOOLS_JSON
    
    # ✅ 防御性编程:配置文件不存在时优雅降级
    if not path.exists():
        logger.warning("工具配置文件不存在: %s", path)
        return

    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except Exception as e:
        # ✅ 错误容错:解析失败不崩溃
        logger.error("加载工具配置失败: %s", e)
        return

    # 解析三大配置块
    self._global_settings = data.get("global_settings", {})
    self._categories = data.get("categories", {})

    tools_section = data.get("tools", {})
    for tool_name, tool_cfg in tools_section.items():
        self._tool_configs[tool_name] = tool_cfg
    
    logger.info("从配置加载了 %d 个工具定义", len(tools_section))

配置文件结构(tools.json 关键片段):

{
  "version": "1.0",
  "description": "Weclaw 工具配置",
  "onboarding_checklist": [
    "1. 创建 src/tools/{name}.py 继承 BaseTool",
    "2. 在本文件 tools 段添加工具配置",
    "3. 在 registry.py _build_init_kwargs 添加参数映射",
    "..."
  ],
  "tools": {
    "shell": {
      "enabled": true,
      "module": "src.tools.shell",
      "class": "ShellTool",
      "display": { "name": "命令行", "emoji": "💻", "category": "system" },
      "config": { "timeout": 30, "max_output_length": 10000 },
      "security": { "risk_level": "high", "blacklist": ["rm -rf", "..."] }
    }
  },
  "categories": { "system": { "name": "系统操作", "emoji": "⚙️" } },
  "global_settings": { "max_concurrent_tools": 3, "default_timeout": 30 }
}

设计要点

  1. onboarding_checklist:新工具接入检查清单,降低遗漏风险
  2. 三层配置分离:display(展示)/ config(参数)/ security(安全)
  3. categories:工具分类,支持按类别查询

3.2 auto_discover():动态导入的魔法

# src/tools/registry.py

def auto_discover(self, lazy: bool = True) -> None:
    """根据已加载的配置自动发现并注册工具实例。

    Args:
        lazy: 是否启用懒加载模式。默认 True。
            - True: 只记录工具元信息,首次使用时才加载(推荐)
            - False: 立即加载所有工具实例
    """
    for tool_name, cfg in self._tool_configs.items():
        # ✅ 步骤 1:检查是否启用
        if not cfg.get("enabled", True):
            logger.info("工具 '%s' 已禁用,跳过", tool_name)
            continue

        module_path = cfg.get("module", "")
        class_name = cfg.get("class", "")
        
        # ✅ 步骤 2:配置完整性检查
        if not module_path or not class_name:
            logger.warning("工具 '%s' 缺少 module/class 配置", tool_name)
            continue

        # ✅ 步骤 3:构建初始化参数
        init_kwargs = self._build_init_kwargs(tool_name, cfg)

        if lazy:
            # 懒加载模式:只记录元信息
            self._lazy_tools[tool_name] = (module_path, class_name, init_kwargs)
            self._preload_tool_metadata(tool_name, module_path, class_name)
        else:
            # 立即加载模式
            try:
                # ✅ 关键:动态导入
                mod = importlib.import_module(module_path)
                cls = getattr(mod, class_name)
                tool_instance = cls(**init_kwargs)
                self.register(tool_instance)
            except Exception as e:
                # ✅ 错误隔离:单个工具失败不影响其他
                logger.error("自动发现工具 '%s' 失败: %s", tool_name, e)

动态导入原理

# 动态导入等价于以下静态代码:
# from src.tools.shell import ShellTool

mod = importlib.import_module("src.tools.shell")  # 相当于 import src.tools.shell
cls = getattr(mod, "ShellTool")                   # 相当于访问 shell.ShellTool
tool_instance = cls(timeout=30, blacklist=[...])  # 实例化

3.3 _preload_tool_metadata():预加载元信息

# src/tools/registry.py

def _preload_tool_metadata(self, tool_name: str, module_path: str, class_name: str) -> None:
    """预加载工具元数据(不实例化工具)。

    用于在懒加载模式下获取工具的 action 定义,以便生成 schema。
    """
    try:
        mod = importlib.import_module(module_path)
        cls = getattr(mod, class_name)
        
        # ✅ 检查类是否符合 BaseTool 接口
        if hasattr(cls, "name") and hasattr(cls, "get_actions"):
            # 获取懒加载的初始化参数
            init_kwargs = self._lazy_tools.get(tool_name, (None, None, {}))[2]
            tool_instance = cls(**init_kwargs)
            
            # ✅ 关键:注册函数映射(func_name → (tool_name, action_name))
            for action in tool_instance.get_actions():
                func_name = f"{tool_name}_{action.name}"
                self._func_map[func_name] = (tool_name, action.name)
            
            # 立即注册到 _tools(因为大多数工具初始化很快)
            self._tools[tool_name] = tool_instance
            logger.debug("预加载工具元数据: %s", tool_name)
    except Exception as e:
        # ⚠️ 预加载失败只记录警告,不阻塞
        logger.warning("预加载工具 '%s' 元数据失败: %s", tool_name, e)

func_map 的作用

┌────────────────────────────────────────────────────────────────┐
│  AI 模型调用: shell_run                                         │
│                  │                                              │
│                  ▼                                              │
│  func_map["shell_run"] → ("shell", "run")                      │
│                  │                                              │
│                  ▼                                              │
│  registry.get_tool("shell").execute("run", params)             │
└────────────────────────────────────────────────────────────────┘

3.4 _build_init_kwargs():16+ 工具的构造参数分发

这是最长的方法,处理不同工具的差异化初始化参数:

# src/tools/registry.py

def _build_init_kwargs(self, tool_name: str, cfg: dict) -> dict[str, Any]:
    """从工具配置中提取构造参数。"""
    kwargs: dict[str, Any] = {}
    tool_config = cfg.get("config", {})
    security = cfg.get("security", {})

    # ✅ 按工具名分发参数
    if tool_name == "shell":
        kwargs["timeout"] = tool_config.get("timeout", 30)
        kwargs["max_output_length"] = tool_config.get("max_output_length", 10000)
        kwargs["working_directory"] = tool_config.get("working_directory", "")
        kwargs["env_vars"] = tool_config.get("env_vars", {})
        kwargs["blacklist"] = security.get("blacklist", [])
        kwargs["whitelist"] = security.get("whitelist", [])
        kwargs["whitelist_mode"] = security.get("whitelist_mode", False)
    
    elif tool_name == "file":
        kwargs["max_read_size"] = tool_config.get("max_read_size", 1_048_576)
        kwargs["max_lines_per_page"] = tool_config.get("max_lines_per_page", 200)
        kwargs["denied_extensions"] = tool_config.get("denied_extensions", [])
    
    elif tool_name == "browser":
        kwargs["headless"] = tool_config.get("headless", False)
        kwargs["timeout"] = tool_config.get("timeout", 30000)
        kwargs["viewport_width"] = tool_config.get("viewport_width", 1280)
        kwargs["viewport_height"] = tool_config.get("viewport_height", 720)
    
    # ... 更多工具的参数映射(共 38+ 个工具)
    
    elif tool_name == "mind_map":
        kwargs["output_dir"] = tool_config.get("output_dir", "generated")
    
    elif tool_name == "speech_to_text":
        kwargs["output_dir"] = tool_config.get("output_dir", "generated")
    
    elif tool_name == "coding_assistant":
        kwargs["output_dir"] = tool_config.get("output_dir", "generated")
            
    return kwargs

代码行数统计_build_init_kwargs 共 98 行,覆盖 38+ 个工具的参数映射。

⚠️ 注意:这是目前架构的一个"代码味道"——每新增工具仍需在此方法添加 elif 分支。后续可优化为反射注入或装饰器声明。

3.5 get_all_schemas():Schema 缓存优化

# src/tools/registry.py

def get_all_schemas(self, use_cache: bool = True) -> list[dict[str, Any]]:
    """获取所有工具的 function calling schema 列表。

    Args:
        use_cache: 是否使用缓存。默认 True。
            首次调用时生成 schema 并缓存,后续直接返回缓存。
    """
    # ✅ 缓存命中:直接返回
    if use_cache and self._schema_cache is not None:
        return self._schema_cache

    schemas = []
    for tool in self._tools.values():
        # 调用每个工具的 get_schema() 方法
        schemas.extend(tool.get_schema())

    # ✅ 写入缓存
    if use_cache:
        self._schema_cache = schemas

    return schemas

Schema 生成原理(BaseTool.get_schema):

# src/tools/base.py

def get_schema(self) -> list[dict[str, Any]]:
    """生成 OpenAI function calling 兼容的 tools schema 列表。"""
    schemas = []
    for action in self.get_actions():
        func_name = f"{self.name}_{action.name}"
        schema: dict[str, Any] = {
            "type": "function",
            "function": {
                "name": func_name,
                "description": f"[{self.title}] {action.description}",
                "parameters": {
                    "type": "object",
                    "properties": action.parameters,
                    "required": action.required_params,
                },
            },
        }
        schemas.append(schema)
    return schemas

性能数据

场景耗时备注
首次生成 Schema(38 工具)12.3ms遍历所有工具
缓存命中0.02ms直接返回列表引用
提升比例99.8%缓存效果显著

四、问题诊断——工具加载失败排查

4.1 真实案例:BrowserTool 导入失败

用户报告

"启动 WeClaw 时报错 ModuleNotFoundError: No module named 'playwright',然后整个程序崩溃了!"

错误日志

2026-03-22 09:00:01 | registry | ERROR | 自动发现工具 'browser' 失败: No module named 'playwright'
Traceback (most recent call last):
  ...
ModuleNotFoundError: No module named 'playwright'

问题:Playwright 是 BrowserTool 的可选依赖,未安装不应导致整体启动失败。

4.2 排查步骤

1️⃣ 检查 auto_discover 的错误处理

# 旧代码(有问题)
def auto_discover_old(self) -> None:
    for tool_name, cfg in self._tool_configs.items():
        # ❌ 问题:没有 try-except 包裹,一个失败全挂
        mod = importlib.import_module(cfg["module"])
        cls = getattr(mod, cfg["class"])
        self.register(cls())

2️⃣ 定位失败位置

┌─────────────────────────────────────────────────────────────────────┐
│  加载顺序                                                            │
├─────────────────────────────────────────────────────────────────────┤
│  ✅ shell     → 成功                                                 │
│  ✅ file      → 成功                                                 │
│  ✅ screen    → 成功                                                 │
│  ❌ browser   → ModuleNotFoundError('playwright')                   │
│  ⏭️ app_control → 未执行(因为 browser 失败导致整体中断)             │
│  ⏭️ ...       → 未执行                                               │
└─────────────────────────────────────────────────────────────────────┘

3️⃣ 根因分析

                    ┌─────────────────────────┐
                    │   auto_discover()       │
                    └───────────┬─────────────┘
                                │
                    ┌───────────▼─────────────┐
                    │  for tool in configs:   │
                    │    import_module()      │
                    │         │               │
                    │     browser 导入失败     │
                    │         │               │
                    │   ❌ 抛出异常,循环中断   │
                    └─────────────────────────┘
                                │
                    ┌───────────▼─────────────┐
                    │  整个程序崩溃           │
                    │  后续工具未注册         │
                    └─────────────────────────┘

4.3 修复方案

# ✅ 修复后:每个工具独立 try-except
def auto_discover(self, lazy: bool = True) -> None:
    for tool_name, cfg in self._tool_configs.items():
        if not cfg.get("enabled", True):
            continue

        module_path = cfg.get("module", "")
        class_name = cfg.get("class", "")
        
        if not module_path or not class_name:
            continue

        init_kwargs = self._build_init_kwargs(tool_name, cfg)

        if lazy:
            self._lazy_tools[tool_name] = (module_path, class_name, init_kwargs)
            self._preload_tool_metadata(tool_name, module_path, class_name)
        else:
            try:
                mod = importlib.import_module(module_path)
                cls = getattr(mod, class_name)
                tool_instance = cls(**init_kwargs)
                self.register(tool_instance)
            except Exception as e:
                # ✅ 关键:单个失败只记录日志,继续下一个
                logger.error("自动发现工具 '%s' 失败: %s", tool_name, e)

4.4 验证结果

2026-03-22 09:00:01 | registry | INFO  | 已注册工具: 💻 命令行 (shell)
2026-03-22 09:00:01 | registry | INFO  | 已注册工具: 📄 文件操作 (file)
2026-03-22 09:00:01 | registry | INFO  | 已注册工具: 📸 屏幕截图 (screen)
2026-03-22 09:00:01 | registry | WARNING | 预加载工具 'browser' 元数据失败: No module named 'playwright'
2026-03-22 09:00:01 | registry | INFO  | 已注册工具: 🪟 应用控制 (app_control)
2026-03-22 09:00:01 | registry | INFO  | 已注册工具: 📋 剪贴板 (clipboard)
...
2026-03-22 09:00:02 | registry | INFO  | 从配置加载了 38 个工具定义
2026-03-22 09:00:02 | registry | INFO  | 成功注册 37 个工具,1 个失败

关键改进

  • browser 失败只打印 WARNING,不阻塞
  • 后续 35 个工具正常加载
  • 启动日志明确显示"成功 37 / 失败 1"

4.5 经验总结:工具加载排查 Checklist

  • 检查 tools.json 中 module/class 路径是否正确
  • 检查工具依赖是否已安装(pip list | grep xxx
  • 检查 _build_init_kwargs 是否覆盖该工具
  • 查看日志中的具体错误信息
  • 单独 import 该工具模块测试:python -c "from src.tools.xxx import XxxTool"

五、性能与扩展性优化

5.1 Schema 缓存:避免重复生成

问题:每次 AI 对话都需要获取 tools schema,38 个工具 × 每个工具平均 4 个 action = 152 个 function 定义。

优化前:每次调用都重新遍历生成

# ❌ 优化前:每次都重新生成
def get_all_schemas_slow(self) -> list[dict[str, Any]]:
    schemas = []
    for tool in self._tools.values():
        schemas.extend(tool.get_schema())  # 每次都调用
    return schemas

优化后:首次生成后缓存

# ✅ 优化后:带缓存
def get_all_schemas(self, use_cache: bool = True) -> list[dict[str, Any]]:
    if use_cache and self._schema_cache is not None:
        return self._schema_cache  # 缓存命中
    
    schemas = []
    for tool in self._tools.values():
        schemas.extend(tool.get_schema())
    
    if use_cache:
        self._schema_cache = schemas
    
    return schemas

缓存失效时机

def register(self, tool: BaseTool) -> None:
    """注册工具时清除缓存"""
    self._tools[tool.name] = tool
    self._schema_cache = None  # ✅ 关键:注册新工具时清缓存

def unregister(self, tool_name: str) -> bool:
    """注销工具时清除缓存"""
    if tool_name in self._tools:
        del self._tools[tool_name]
        self._schema_cache = None  # ✅ 关键:注销工具时清缓存
        return True
    return False

5.2 错误容错:单个失败不阻塞整体

设计原则:可选依赖的工具失败,不应阻塞核心工具的加载。

# src/tools/registry.py

def _preload_tool_metadata(self, tool_name: str, module_path: str, class_name: str) -> None:
    try:
        mod = importlib.import_module(module_path)
        cls = getattr(mod, class_name)
        # ... 注册逻辑 ...
    except ImportError as e:
        # ✅ 可选依赖缺失:只记录警告
        logger.warning("工具 '%s' 依赖缺失,已跳过: %s", tool_name, e)
    except AttributeError as e:
        # ✅ 类名错误:记录错误
        logger.error("工具 '%s' 类名错误: %s", tool_name, e)
    except Exception as e:
        # ✅ 其他异常:通用处理
        logger.warning("预加载工具 '%s' 失败: %s", tool_name, e)

5.3 扩展指南:新增一个工具只需 3 步

Step 1:创建工具类

# src/tools/my_tool.py

from src.tools.base import BaseTool, ActionDef, ToolResult, ToolResultStatus

class MyTool(BaseTool):
    name = "my_tool"
    emoji = "🔧"
    title = "我的工具"
    description = "这是一个示例工具"

    def get_actions(self) -> list[ActionDef]:
        return [
            ActionDef(
                name="do_something",
                description="执行某个操作",
                parameters={
                    "input": {"type": "string", "description": "输入参数"}
                },
                required_params=["input"]
            )
        ]

    async def execute(self, action: str, params: dict) -> ToolResult:
        if action == "do_something":
            return ToolResult(
                status=ToolResultStatus.SUCCESS,
                output=f"处理结果: {params.get('input')}"
            )
        return ToolResult(
            status=ToolResultStatus.ERROR,
            error=f"未知动作: {action}"
        )

Step 2:添加配置到 tools.json

{
  "tools": {
    "my_tool": {
      "enabled": true,
      "module": "src.tools.my_tool",
      "class": "MyTool",
      "display": {
        "name": "我的工具",
        "emoji": "🔧",
        "description": "这是一个示例工具",
        "category": "utility"
      },
      "config": {},
      "security": {
        "risk_level": "low",
        "require_confirmation": false
      },
      "actions": ["do_something"]
    }
  }
}

Step 3:(可选)添加参数映射

如果工具需要特殊的初始化参数,在 _build_init_kwargs 中添加:

elif tool_name == "my_tool":
    kwargs["some_param"] = tool_config.get("some_param", "default")

5.4 Do's / Don'ts 清单

Do's(推荐做法)

  • ✅ 使用 enabled: false 禁用工具,而不是删除配置
  • ✅ 为新工具编写 get_actions() 时,参数 description 要详细
  • ✅ 在 tools.json 的 onboarding_checklist 中记录接入步骤
  • ✅ 使用 get_all_schemas(use_cache=True) 获取 Schema
  • ✅ 单元测试时用 get_all_schemas(use_cache=False) 避免缓存干扰

Don'ts(避免做法)

  • ❌ 在 auto_discover 中直接 raise 异常
  • ❌ 忘记在 register/unregister 时清除 schema 缓存
  • ❌ 在工具的 __init__ 中执行重操作(如启动浏览器)
  • ❌ 硬编码工具参数,应该从 tools.json 读取
  • ❌ 忽略 _build_init_kwargs 的 elif 分支覆盖

六、总结与展望

6.1 核心要点回顾

本文讲解了 WeClaw 工具注册系统的演进之路:

3 个核心组件

  1. BaseTool(工具基类):定义 get_actions() + execute() 统一接口
  2. ToolRegistry(注册中心):配置加载 → 自动发现 → Schema 缓存
  3. tools.json(配置文件):声明式定义工具元信息

1 个核心公式

可扩展工具系统 = 配置声明(tools.json) 
              + 自动发现(importlib) 
              + 错误隔离(try-except) 
              + Schema 缓存

关键数据

指标优化前优化后提升
新增工具改动文件数5 个2 个60%
Schema 查询耗时12.3ms0.02ms99.8%
单工具失败影响整体崩溃仅该工具不可用-

6.2 下一步学习方向

前置知识

  • ✅ Python importlib 动态导入
  • ✅ JSON 配置文件解析
  • ✅ 抽象基类(ABC)设计模式

后续主题

  • 📖 下一篇:《第 25 篇:可选依赖的优雅处理——思维导图和语音识别的双引擎架构》

6.3 互动环节

思考题

  1. _build_init_kwargs 中的 elif 链条会随工具增多而膨胀,如何重构成更优雅的设计?(提示:装饰器 / 反射 / 工厂模式)

  2. 如果需要支持"热更新"——运行时新增工具而不重启服务,需要在 ToolRegistry 中增加哪些机制?

  3. 当工具数量超过 100 个时,启动时的 auto_discover 可能变慢,如何实现"真正的懒加载"——首次调用时才加载?

讨论话题

在你的项目中,是如何管理插件/工具的扩展的?是硬编码注册、配置驱动,还是更高级的依赖注入框架?欢迎在评论区分享你的经验!


下期预告:《第 25 篇:可选依赖的优雅处理》

  • 🧠 思维导图工具:graphviz 可选,XMind 可选,如何优雅降级?
  • 🎙️ 语音识别工具:Whisper 本地 vs 云端 API,双引擎切换
  • 📦 ImportError 的三种处理策略:跳过 / 降级 / 提示安装
  • 🔧 工具能力探测:运行时检测哪些 action 可用

敬请期待!


附录 A:完整代码清单

文件路径行数作用
src/tools/base.py242 行BaseTool 抽象基类、ToolResult、ActionDef
src/tools/registry.py469 行ToolRegistry 完整实现
config/tools.json1070 行38+ 工具配置定义

总代码量:约 1781 行
已注册工具:38 个
支持 Actions:170+ 个


附录 B:参考资料

  1. Python importlib 官方文档
  2. OpenAI Function Calling 文档
  3. 抽象基类(ABC)设计模式
  4. 上一篇:《第 23 篇:从工具调用推断用户档案——无感建档系统的分阶段采集策略》
  5. 下一篇:《第 25 篇:可选依赖的优雅处理——思维导图和语音识别的双引擎架构》

版权声明:本文为 CSDN 博主「翁勇刚」的原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接及本声明。

原文链接https://blog.csdn.net/yweng18/article/details/xxxxxx(待发布后更新)