""" 技能管理器 管理所有技能的注册、发现和调用 """ from typing import Dict, Optional, List, Type from app.skills.base import BaseSkill from app.utils.logger import logger class SkillManager: """技能管理器""" def __init__(self): """初始化技能管理器""" self._skills: Dict[str, BaseSkill] = {} logger.info("技能管理器初始化") def register(self, skill: BaseSkill) -> bool: """ 注册技能 Args: skill: 技能实例 Returns: 是否成功 """ if not skill.name: logger.error("技能名称不能为空") return False if skill.name in self._skills: logger.warning(f"技能已存在,将被覆盖: {skill.name}") self._skills[skill.name] = skill logger.info(f"技能注册成功: {skill.name}") return True def unregister(self, skill_name: str) -> bool: """ 注销技能 Args: skill_name: 技能名称 Returns: 是否成功 """ if skill_name in self._skills: del self._skills[skill_name] logger.info(f"技能注销成功: {skill_name}") return True logger.warning(f"技能不存在: {skill_name}") return False def get_skill(self, skill_name: str) -> Optional[BaseSkill]: """ 获取技能 Args: skill_name: 技能名称 Returns: 技能实例或None """ return self._skills.get(skill_name) def get_all_skills(self) -> List[BaseSkill]: """ 获取所有技能 Returns: 技能列表 """ return list(self._skills.values()) def get_enabled_skills(self) -> List[BaseSkill]: """ 获取所有启用的技能 Returns: 启用的技能列表 """ return [skill for skill in self._skills.values() if skill.enabled] async def execute_skill(self, skill_name: str, **kwargs) -> Dict: """ 执行技能 Args: skill_name: 技能名称 **kwargs: 技能参数 Returns: 执行结果 """ skill = self.get_skill(skill_name) if not skill: return { "success": False, "error": f"技能不存在: {skill_name}" } if not skill.enabled: return { "success": False, "error": f"技能已禁用: {skill_name}" } # 验证参数 valid, error = skill.validate_params(**kwargs) if not valid: return { "success": False, "error": error } # 执行技能 try: result = await skill.execute(**kwargs) return { "success": True, "data": result } except Exception as e: logger.error(f"技能执行失败 {skill_name}: {e}") return { "success": False, "error": str(e) } def enable_skill(self, skill_name: str) -> bool: """ 启用技能 Args: skill_name: 技能名称 Returns: 是否成功 """ skill = self.get_skill(skill_name) if skill: skill.enable() logger.info(f"技能已启用: {skill_name}") return True return False def disable_skill(self, skill_name: str) -> bool: """ 禁用技能 Args: skill_name: 技能名称 Returns: 是否成功 """ skill = self.get_skill(skill_name) if skill: skill.disable() logger.info(f"技能已禁用: {skill_name}") return True return False def get_skills_info(self) -> List[Dict]: """ 获取所有技能信息 Returns: 技能信息列表 """ return [skill.get_info() for skill in self._skills.values()] # 创建全局技能管理器实例 skill_manager = SkillManager()