OpenClaw自动化技能发现机制：从原理到实现

2次阅读

共计 3553 个字符，预计需要花费 9 分钟才能阅读完成。

在传统 OpenClaw 开发中，技能（Skill）需要手动注册到框架中，这种方式存在几个显著问题：

维护成本高：每次新增或删除技能都需要修改注册代码
容易出错：人工管理容易遗漏或重复注册
缺乏灵活性：无法动态加载新技能而不重启服务
技术债务积累：随着技能数量增加，注册代码会变得臃肿难维护

使用 Python 标准库的 importlib 实现技能模块的动态发现和加载：

import importlib
import pkgutil
from pathlib import Path

def discover_skills(package_path: str):
    """自动发现指定包路径下的所有技能模块"""
    package = importlib.import_module(package_path)
    for _, name, _ in pkgutil.iter_modules(package.__path__):
        yield importlib.import_module(f"{package_path}.{name}")

通过自定义装饰器标记技能类并收集元数据：

from typing import Dict, Type
from functools import wraps

_registry: Dict[str, Type["BaseSkill"]] = {}

def skill(name: str, version: str = "1.0"):
    """技能注册装饰器"""
    def decorator(cls):
        @wraps(cls)
        def wrapper(*args, **kwargs):
            return cls(*args, **kwargs)

        # 元数据注入
        wrapper.skill_name = name
        wrapper.skill_version = version

        # 注册到全局字典
        _registry[name] = wrapper
        return wrapper
    return decorator

使用 abc 模块定义技能接口规范：

from abc import ABC, abstractmethod
from typing import Any, Dict

class BaseSkill(ABC):
    """技能抽象基类"""

    @abstractmethod
    def execute(self, context: Dict[str, Any]) -> Any:
        """执行技能核心逻辑"""
        pass

    @classmethod
    @abstractmethod
    def health_check(cls) -> bool:
        """健康检查方法"""
        pass

以下是整合后的核心实现：

# skill_registry.py
import importlib
import pkgutil
from abc import ABC, abstractmethod
from functools import wraps
from pathlib import Path
from typing import Any, Dict, Type, Optional
import threading

class SkillRegistry:
    """线程安全的技能注册中心"""

    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    cls._instance = super().__new__(cls)
                    cls._instance._registry = {}
                    cls._instance._loaded = False
        return cls._instance

    def register(self, name: str, skill_cls: Type["BaseSkill"]) -> None:
        """注册技能类"""
        with self._lock:
            self._registry[name] = skill_cls

    def get_skill(self, name: str) -> Optional[Type["BaseSkill"]]:
        """获取技能类（懒加载）"""
        with self._lock:
            return self._registry.get(name)

    def discover(self, package_path: str) -> None:
        """发现并加载所有技能"""
        if self._loaded:
            return

        with self._lock:
            package = importlib.import_module(package_path)
            for _, name, _ in pkgutil.iter_modules(package.__path__):
                module = importlib.import_module(f"{package_path}.{name}")
                # 模块加载时会自动触发装饰器注册
            self._loaded = True

# base.py
class BaseSkill(ABC):
    """技能抽象基类"""

    @abstractmethod
    def execute(self, context: Dict[str, Any]) -> Any:
        """执行技能核心逻辑"""
        pass

    @classmethod
    @abstractmethod
    def health_check(cls) -> bool:
        """健康检查方法"""
        pass

# decorators.py
def skill(name: str, version: str = "1.0"):
    """技能注册装饰器"""
    def decorator(cls):
        if not issubclass(cls, BaseSkill):
            raise TypeError("Skill must inherit from BaseSkill")

        @wraps(cls)
        def wrapper(*args, **kwargs):
            return cls(*args, **kwargs)

        # 添加元数据
        wrapper.skill_name = name
        wrapper.skill_version = version

        # 自动注册
        registry = SkillRegistry()
        registry.register(name, wrapper)

        return wrapper
    return decorator

技能模块按需加载，避免启动时加载所有技能：

def get_skill(name: str) -> Optional[BaseSkill]:
    """按需获取技能实例"""
    skill_cls = SkillRegistry().get_skill(name)
    if skill_cls:
        return skill_cls()
    return None

使用 LRU 缓存最近使用的技能实例：

from functools import lru_cache

@lru_cache(maxsize=32)
def get_cached_skill(name: str) -> Optional[BaseSkill]:
    """带缓存的技能获取"""
    return get_skill(name)

在技能元数据中添加版本约束：

@skill(name="weather", version="2.1", 
       min_framework_version="1.3")
class WeatherSkill(BaseSkill):
    # ...

使用 exec 和ast模块实现安全执行环境：

import ast

def safe_execute(code: str, context: dict):
    """在受限环境中执行代码"""
    try:
        # 解析 AST 确保没有危险操作
        parsed = ast.parse(code)
        # 安全检查逻辑...

        # 在限制的 globals 中执行
        safe_globals = {"__builtins__": None}
        exec(code, safe_globals, context)
    except Exception as e:
        raise RuntimeError(f"Execution failed: {str(e)}")