cfdaily
248 lines
7.1 KiB
Python
248 lines
7.1 KiB
Python
"""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)
|