diff --git a/src/daemon/skill_system.py b/src/daemon/skill_system.py new file mode 100644 index 0000000..7774763 --- /dev/null +++ b/src/daemon/skill_system.py @@ -0,0 +1,247 @@ +"""Skill System — 技能注册、加载、匹配、执行 + +三层自由度： +- 高自由（原则层）：只有名称+描述，Agent 自由发挥 +- 中自由（模板层）：有 prompt template，Agent 按模板填 +- 低自由（脚本层）：有脚本/工具，Agent 调用执行 +""" + +from __future__ import annotations + +import json +import logging +import re +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Tuple + +logger = logging.getLogger("moziplus-v2.skill") + + +class SkillFreedom(str, Enum): + HIGH = "high" # 原则层 + MEDIUM = "medium" # 模板层 + LOW = "low" # 脚本层 + + +@dataclass +class Skill: + """技能描述""" + id: str + name: str + description: str + freedom: str = SkillFreedom.HIGH.value + prompt_template: Optional[str] = None + script_path: Optional[str] = None + tags: List[str] = field(default_factory=list) + input_schema: Optional[Dict[str, Any]] = None + output_schema: Optional[Dict[str, Any]] = None + enabled: bool = True + created_at: Optional[str] = None + updated_at: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "id": self.id, + "name": self.name, + "description": self.description, + "freedom": self.freedom, + "prompt_template": self.prompt_template, + "script_path": self.script_path, + "tags": self.tags, + "enabled": self.enabled, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> Skill: + return cls(**{k: v for k, v in data.items() + if k in cls.__dataclass_fields__}) + + +class SkillRegistry: + """技能注册表""" + + def __init__(self, skills_dir: Optional[Path] = None): + self._skills: Dict[str, Skill] = {} + self.skills_dir = skills_dir + + if skills_dir and skills_dir.exists(): + self._load_from_dir(skills_dir) + + def _load_from_dir(self, dir_path: Path) -> None: + """从目录加载 skill 定义（每个 .json 一个）""" + for f in dir_path.glob("*.json"): + try: + data = json.loads(f.read_text()) + skill = Skill.from_dict(data) + self._skills[skill.id] = skill + except Exception: + logger.warning("Failed to load skill from %s", f) + + def register(self, skill: Skill) -> str: + self._skills[skill.id] = skill + return skill.id + + def unregister(self, skill_id: str) -> bool: + if skill_id in self._skills: + del self._skills[skill_id] + return True + return False + + def get(self, skill_id: str) -> Optional[Skill]: + return self._skills.get(skill_id) + + def list_skills( + self, + freedom: Optional[str] = None, + tags: Optional[List[str]] = None, + enabled_only: bool = True, + ) -> List[Skill]: + results = list(self._skills.values()) + + if enabled_only: + results = [s for s in results if s.enabled] + + if freedom: + results = [s for s in results if s.freedom == freedom] + + if tags: + results = [s for s in results + if any(t in s.tags for t in tags)] + + return results + + def match( + self, + query: str, + task_type: Optional[str] = None, + limit: int = 5, + ) -> List[Tuple[Skill, float]]: + """匹配技能 + + Returns: + [(Skill, relevance_score), ...] + """ + q_lower = query.lower() + scored = [] + + for skill in self._skills.values(): + if not skill.enabled: + continue + score = 0.0 + + # 名称匹配（权重最高） + if q_lower in skill.name.lower(): + score += 3.0 + + # 描述匹配 + if q_lower in skill.description.lower(): + score += 2.0 + + # 标签匹配 + if task_type and task_type.lower() in skill.tags: + score += 1.5 + + for tag in skill.tags: + if tag.lower() in q_lower: + score += 1.0 + + if score > 0: + scored.append((skill, score)) + + scored.sort(key=lambda x: x[1], reverse=True) + return scored[:limit] + + def count(self) -> int: + return len(self._skills) + + +class SkillExecutor: + """技能执行器""" + + def __init__( + self, + registry: Optional[SkillRegistry] = None, + allow_scripts: bool = False, + ): + self.registry = registry or SkillRegistry() + self.allow_scripts = allow_scripts + self._execution_log: List[Dict[str, Any]] = [] + + def build_prompt( + self, + skill_id: str, + variables: Optional[Dict[str, str]] = None, + ) -> Optional[str]: + """构建 skill prompt""" + skill = self.registry.get(skill_id) + if not skill: + return None + + if skill.freedom == SkillFreedom.HIGH.value: + return f"## Skill: {skill.name}\n{skill.description}\n\nPrinciple: Apply the above guidelines." + + if skill.freedom == SkillFreedom.MEDIUM.value: + template = skill.prompt_template or skill.description + if variables: + for key, value in variables.items(): + template = template.replace(f"{{{{{key}}}}}", value) + return template + + if skill.freedom == SkillFreedom.LOW.value: + parts = [f"## Skill: {skill.name}"] + parts.append(skill.description) + if skill.script_path: + parts.append(f"Execute: `{skill.script_path}`") + return "\n".join(parts) + + return skill.description + + def execute( + self, + skill_id: str, + context: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """执行技能 + + Returns: + {"status": "success"|"error", "prompt": str, "skill": str} + """ + skill = self.registry.get(skill_id) + if not skill: + return {"status": "error", "error": f"Skill not found: {skill_id}"} + + if not skill.enabled: + return {"status": "error", "error": f"Skill disabled: {skill_id}"} + + prompt = self.build_prompt(skill_id, context) + + # 脚本执行需要显式允许 + if skill.freedom == SkillFreedom.LOW.value and skill.script_path: + if not self.allow_scripts: + return { + "status": "error", + "error": "Script execution not allowed", + "prompt": prompt, + } + + result = { + "status": "success", + "prompt": prompt, + "skill": skill_id, + "freedom": skill.freedom, + } + + self._execution_log.append({ + "skill_id": skill_id, + "timestamp": datetime.utcnow().isoformat(), + "status": result["status"], + }) + + return result + + @property + def execution_log(self) -> List[Dict[str, Any]]: + return list(self._execution_log)