106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""Dynamic Local Skill Loader."""
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
import yaml
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class LocalSkill:
|
|
"""Represents a local skill loaded from SKILL.md."""
|
|
|
|
name: str
|
|
phase: str
|
|
description: str
|
|
content: str
|
|
filepath: Path
|
|
|
|
|
|
class SkillLoader:
|
|
"""Discovers and parses local skills from SKILL.md files."""
|
|
|
|
def __init__(self, skills_dir: Path) -> None:
|
|
self.skills_dir = skills_dir
|
|
self._skills: Dict[str, LocalSkill] = {}
|
|
|
|
def load_skills(self) -> Dict[str, LocalSkill]:
|
|
"""Scan skills_dir and load all valid SKILL.md files."""
|
|
self._skills.clear()
|
|
if not self.skills_dir.exists():
|
|
logger.warning("Skills directory does not exist: %s", self.skills_dir)
|
|
return self._skills
|
|
|
|
for skill_file in self.skills_dir.glob("**/SKILL.md"):
|
|
try:
|
|
skill = self._parse_skill_file(skill_file)
|
|
if skill:
|
|
self._skills[skill.name] = skill
|
|
logger.info("Loaded skill '%s' (phase: %s)", skill.name, skill.phase)
|
|
except Exception as exc:
|
|
logger.error("Failed to parse skill file %s: %s", skill_file, exc)
|
|
|
|
return self._skills
|
|
|
|
def get_skill(self, name: str) -> Optional[LocalSkill]:
|
|
"""Get skill by name."""
|
|
if not self._skills:
|
|
self.load_skills()
|
|
return self._skills.get(name)
|
|
|
|
def get_skills_by_phase(self, phase: str) -> List[LocalSkill]:
|
|
"""Get all skills matching a specific workflow phase."""
|
|
if not self._skills:
|
|
self.load_skills()
|
|
return [skill for skill in self._skills.values() if skill.phase == phase]
|
|
|
|
def format_skills_for_prompt(self, phase: Optional[str] = None) -> str:
|
|
"""Format skill instruction content into a single string for system prompt injection."""
|
|
skills = self.get_skills_by_phase(phase) if phase else list(self._skills.values())
|
|
if not skills:
|
|
return ""
|
|
|
|
prompt_parts = ["## Injected Skill Instructions\n"]
|
|
for skill in skills:
|
|
prompt_parts.append(f"### Skill: {skill.name} (Phase: {skill.phase})\n{skill.content}\n")
|
|
return "\n".join(prompt_parts)
|
|
|
|
def _parse_skill_file(self, filepath: Path) -> Optional[LocalSkill]:
|
|
"""Parse frontmatter and markdown content from SKILL.md."""
|
|
raw_text = filepath.read_text(encoding="utf-8")
|
|
if not raw_text.startswith("---"):
|
|
return LocalSkill(
|
|
name=filepath.parent.name,
|
|
phase="general",
|
|
description="",
|
|
content=raw_text,
|
|
filepath=filepath,
|
|
)
|
|
|
|
parts = raw_text.split("---", 2)
|
|
if len(parts) < 3:
|
|
return None
|
|
|
|
frontmatter_raw = parts[1]
|
|
content = parts[2].strip()
|
|
|
|
try:
|
|
metadata = yaml.safe_load(frontmatter_raw) or {}
|
|
except Exception:
|
|
metadata = {}
|
|
|
|
name = metadata.get("name", filepath.parent.name)
|
|
phase = metadata.get("phase", "general")
|
|
description = metadata.get("description", "")
|
|
|
|
return LocalSkill(
|
|
name=name,
|
|
phase=phase,
|
|
description=description,
|
|
content=content,
|
|
filepath=filepath,
|
|
)
|