架构师AI杜

Appearance

第3章：技能系统 ​

3.1 技能系统概述 ​

技能系统是Nanobot2的核心功能之一，允许AI助手执行各种特定任务，如搜索、查询天气、计算数学表达式等。技能系统采用插件式架构，支持自定义技能的注册和执行。

3.1.1 技能系统架构 ​

技能系统由以下组件组成：

  1. 技能基类：定义技能的基本接口和行为
  2. 技能管理器：负责技能的注册、管理和执行
  3. 内置技能：系统自带的常用技能
  4. 自定义技能：用户或开发者创建的技能

3.2 技能基类设计 ​

技能基类是所有技能的基础，定义了技能必须实现的方法和属性。

3.2.1 核心实现 ​

python
from abc import ABC, abstractmethod

class Skill(ABC):
    """
    技能基类
    
    所有技能必须继承自此类并实现execute方法
    """
    
    def __init__(self):
        self.name = self.__class__.__name__.replace("Skill", "").lower()
        self.description = ""
        self.parameters = []
    
    @abstractmethod
    async def execute(self, **kwargs) -> str:
        """
        执行技能
        
        Args:
            **kwargs: 技能参数
        
        Returns:
            技能执行结果
        """
        pass
    
    def get_name(self) -> str:
        """
        获取技能名称
        
        Returns:
            技能名称
        """
        return self.name
    
    def get_description(self) -> str:
        """
        获取技能描述
        
        Returns:
            技能描述
        """
        return self.description
    
    def get_parameters(self) -> list:
        """
        获取技能参数
        
        Returns:
            技能参数列表
        """
        return self.parameters

3.2.2 技能属性 ​

属性描述默认值
name技能名称类名去除"Skill"后缀并转为小写
description技能描述空字符串
parameters技能参数列表空列表

3.3 技能管理器 ​

技能管理器负责技能的注册、管理和执行，是技能系统的核心组件。

3.3.1 核心实现 ​

python
class SkillManager:
    """
    技能管理器
    
    负责技能的注册、管理和执行
    """
    
    def __init__(self):
        self.skills = {}
    
    def register_skill(self, skill: Skill):
        """
        注册技能
        
        Args:
            skill: 技能实例
        """
        skill_name = skill.get_name()
        self.skills[skill_name] = skill
        print(f"注册技能: {skill_name}")
    
    def unregister_skill(self, skill_name: str):
        """
        注销技能
        
        Args:
            skill_name: 技能名称
        """
        if skill_name in self.skills:
            del self.skills[skill_name]
            print(f"注销技能: {skill_name}")
    
    def get_skill(self, skill_name: str) -> Skill:
        """
        获取技能
        
        Args:
            skill_name: 技能名称
        
        Returns:
            技能实例
        """
        return self.skills.get(skill_name)
    
    def list_skills(self) -> list:
        """
        列出所有技能
        
        Returns:
            技能名称列表
        """
        return list(self.skills.keys())
    
    def get_skill_info(self, skill_name: str) -> dict:
        """
        获取技能信息
        
        Args:
            skill_name: 技能名称
        
        Returns:
            技能信息字典
        """
        skill = self.get_skill(skill_name)
        if skill:
            return {
                "name": skill.get_name(),
                "description": skill.get_description(),
                "parameters": skill.get_parameters()
            }
        return {}
    
    async def execute_skill(self, skill_name: str, **kwargs) -> str:
        """
        执行技能
        
        Args:
            skill_name: 技能名称
            **kwargs: 技能参数
        
        Returns:
            技能执行结果
        """
        skill = self.get_skill(skill_name)
        if not skill:
            return f"未知技能: {skill_name}"
        
        try:
            result = await skill.execute(**kwargs)
            return result
        except Exception as e:
            return f"执行技能失败: {str(e)}"

3.3.2 全局技能管理器 ​

为了方便在系统的不同部分使用技能管理器，我们提供了一个全局技能管理器实例：

python
# 全局技能管理器实例
_skill_manager = None

def get_skill_manager() -> SkillManager:
    """
    获取全局技能管理器实例
    
    Returns:
        技能管理器实例
    """
    global _skill_manager
    if _skill_manager is None:
        _skill_manager = SkillManager()
    return _skill_manager

3.4 内置技能实现 ​

Nanobot2内置了几个常用技能，如搜索、天气、计算器和时间。

3.4.1 搜索技能 ​

python
class SearchSkill(Skill):
    """
    搜索技能
    
    用于搜索网络信息
    """
    
    def __init__(self):
        super().__init__()
        self.description = "搜索网络信息"
        self.parameters = ["query"]
    
    async def execute(self, **kwargs) -> str:
        """
        执行搜索
        
        Args:
            query: 搜索关键词
        
        Returns:
            搜索结果
        """
        query = kwargs.get("query", "")
        if not query:
            return "请提供搜索关键词"
        
        # 这里可以集成真实的搜索API，如Google Search API
        # 为了演示，我们返回模拟结果
        return f"搜索结果 for '{query}':\n这是搜索结果的示例内容。在实际实现中，这里会显示真实的搜索结果。"

3.4.2 天气技能 ​

python
class WeatherSkill(Skill):
    """
    天气技能
    
    用于查询天气信息
    """
    
    def __init__(self):
        super().__init__()
        self.description = "查询天气信息"
        self.parameters = ["city"]
    
    async def execute(self, **kwargs) -> str:
        """
        查询天气
        
        Args:
            city: 城市名称
        
        Returns:
            天气信息
        """
        city = kwargs.get("city", "")
        if not city:
            return "请提供城市名称"
        
        # 这里可以集成真实的天气API，如OpenWeatherMap API
        # 为了演示，我们返回模拟结果
        return f"{city}的天气:\n温度: 25°C\n天气: 晴\n风力: 3级\n湿度: 60%"

3.4.3 计算器技能 ​

python
class CalculatorSkill(Skill):
    """
    计算器技能
    
    用于计算数学表达式
    """
    
    def __init__(self):
        super().__init__()
        self.description = "计算数学表达式"
        self.parameters = ["expression"]
    
    async def execute(self, **kwargs) -> str:
        """
        计算数学表达式
        
        Args:
            expression: 数学表达式
        
        Returns:
            计算结果
        """
        expression = kwargs.get("expression", "")
        if not expression:
            return "请提供数学表达式"
        
        try:
            # 使用eval计算表达式
            # 注意：在生产环境中，应该使用更安全的方法计算表达式
            result = eval(expression)
            return f"计算结果：{expression} = {result}"
        except Exception as e:
            return f"计算失败: {str(e)}"

3.4.4 时间技能 ​

python
import time

class TimeSkill(Skill):
    """
    时间技能
    
    用于获取当前时间
    """
    
    def __init__(self):
        super().__init__()
        self.description = "获取当前时间"
        self.parameters = []
    
    async def execute(self, **kwargs) -> str:
        """
        获取当前时间
        
        Returns:
            当前时间
        """
        current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        return f"当前时间: {current_time}"

3.5 技能注册 ​

为了方便使用内置技能，我们提供了一个注册内置技能的函数：

python
def register_builtin_skills():
    """
    注册内置技能
    """
    from nanobot.skills import get_skill_manager
    from nanobot.skills.builtin import (
        SearchSkill,
        WeatherSkill,
        CalculatorSkill,
        TimeSkill
    )
    
    skill_manager = get_skill_manager()
    
    # 注册内置技能
    skill_manager.register_skill(SearchSkill())
    skill_manager.register_skill(WeatherSkill())
    skill_manager.register_skill(CalculatorSkill())
    skill_manager.register_skill(TimeSkill())
    
    print("内置技能注册完成")

3.6 自定义技能开发 ​

3.6.1 自定义技能示例 ​

要创建自定义技能，只需继承Skill基类并实现execute方法：

python
from nanobot.skills import Skill

class HelloSkill(Skill):
    """
    问候技能
    
    用于向用户问好
    """
    
    def __init__(self):
        super().__init__()
        self.description = "向用户问好"
        self.parameters = ["name"]
    
    async def execute(self, **kwargs) -> str:
        """
        执行问候
        
        Args:
            name: 用户名
        
        Returns:
            问候语
        """
        name = kwargs.get("name", "用户")
        return f"你好，{name}！欢迎使用Nanobot2！"

3.6.2 注册自定义技能 ​

创建自定义技能后，需要将其注册到技能管理器：

python
from nanobot.skills import get_skill_manager
from my_skills import HelloSkill

skill_manager = get_skill_manager()
skill_manager.register_skill(HelloSkill())

3.7 技能执行流程 ​

3.7.1 从用户命令执行技能 ​

  1. 命令解析：解析用户命令，提取技能名称和参数
  2. 技能查找：根据技能名称查找对应的技能
  3. 参数验证：验证提供的参数是否满足技能要求
  4. 技能执行：执行技能并获取结果
  5. 结果返回：将执行结果返回给用户

3.7.2 从Agent执行技能 ​

Agent可以根据用户的请求自动选择和执行相应的技能：

python
async def process_message(self, user_id: str, channel: str, content: str) -> str:
    # ... 前面的代码 ...
    
    # 检查是否需要执行技能
    skill_match = self._parse_skill_command(content)
    if skill_match:
        skill_name = skill_match["name"]
        parameters = skill_match["parameters"]
        
        # 执行技能
        result = await self.skill_manager.execute_skill(skill_name, **parameters)
        
        # 存储结果到会话
        session.add_message("assistant", result)
        
        return result
    
    # ... 后面的代码 ...

3.8 技能系统测试 ​

3.8.1 单元测试 ​

为技能系统编写单元测试，确保各功能正常工作：

python
import unittest
from unittest.mock import AsyncMock
from nanobot.skills import SkillManager
from nanobot.skills.builtin import CalculatorSkill

class TestSkillManager(unittest.TestCase):
    def setUp(self):
        self.skill_manager = SkillManager()
        self.calculator_skill = CalculatorSkill()
        self.skill_manager.register_skill(self.calculator_skill)
    
    async def test_execute_skill(self):
        # 测试执行技能
        result = await self.skill_manager.execute_skill("calculator", expression="1+1")
        self.assertEqual(result, "计算结果：1+1 = 2")
    
    async def test_unknown_skill(self):
        # 测试未知技能
        result = await self.skill_manager.execute_skill("unknown")
        self.assertEqual(result, "未知技能: unknown")

if __name__ == "__main__":
    unittest.main()

3.8.2 集成测试 ​

运行完整的集成测试，确保技能系统与其他模块正常协作：

bash
python -m pytest tests/test_skills.py -v

3.9 技能系统扩展 ​

3.9.1 技能发现机制 ​

为了方便自动发现和加载自定义技能，可以实现一个技能发现机制：

python
def discover_skills(directory: str):
    """
    发现并加载目录中的技能
    
    Args:
        directory: 技能目录
    """
    import os
    import importlib.util
    from nanobot.skills import get_skill_manager
    
    skill_manager = get_skill_manager()
    
    for filename in os.listdir(directory):
        if filename.endswith(".py") and not filename.startswith("_"):
            module_name = filename[:-3]
            module_path = os.path.join(directory, filename)
            
            # 加载模块
            spec = importlib.util.spec_from_file_location(module_name, module_path)
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
            
            # 查找并注册技能类
            for name, obj in module.__dict__.items():
                if (
                    isinstance(obj, type) and 
                    issubclass(obj, Skill) and 
                    obj != Skill
                ):
                    skill_manager.register_skill(obj())

3.9.2 技能配置 ​

为了使技能更加灵活，可以支持技能配置：

python
class ConfigurableSkill(Skill):
    """
    可配置技能基类
    """
    
    def __init__(self):
        super().__init__()
        self.config = {}
    
    def set_config(self, config: dict):
        """
        设置技能配置
        
        Args:
            config: 配置字典
        """
        self.config.update(config)
    
    def get_config(self, key: str, default: Any = None) -> Any:
        """
        获取技能配置
        
        Args:
            key: 配置键
            default: 默认值
        
        Returns:
            配置值
        """
        return self.config.get(key, default)

3.10 小结 ​

本章介绍了Nanobot2技能系统的设计和实现，包括：

  1. 技能系统架构：技能基类、技能管理器、内置技能和自定义技能
  2. 技能基类设计：定义技能的基本接口和行为
  3. 技能管理器：负责技能的注册、管理和执行
  4. 内置技能实现：搜索、天气、计算器和时间技能
  5. 自定义技能开发：如何创建和注册自定义技能
  6. 技能执行流程：从用户命令或Agent执行技能的流程
  7. 技能系统测试：单元测试和集成测试
  8. 技能系统扩展：技能发现机制和技能配置

技能系统使Nanobot2具备了执行特定任务的能力，大大增强了其实用性。在接下来的章节中，我们将学习多渠道集成，使Nanobot2能够通过不同的渠道与用户交互。