auto-sync: 2026-05-17 06:10:53
This commit is contained in:
cfdaily
@@ -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)
|
||||
Reference in New Issue
Block a user