271 lines
8.3 KiB
Python
271 lines
8.3 KiB
Python
"""
|
|
Aetheel Skills System
|
|
=====================
|
|
Discovers, parses, and injects skill context into the system prompt.
|
|
|
|
Skills are markdown files in the workspace that teach the agent how to
|
|
handle specific types of requests. They are loaded once at startup and
|
|
injected into the system prompt when their trigger words match the
|
|
user's message.
|
|
|
|
Skill format (~/.aetheel/workspace/skills/<name>/SKILL.md):
|
|
---
|
|
name: weather
|
|
description: Check weather for any city
|
|
triggers: [weather, forecast, temperature, rain]
|
|
---
|
|
|
|
# Weather Skill
|
|
|
|
When the user asks about weather, use the following approach:
|
|
...
|
|
|
|
Usage:
|
|
from skills import SkillsManager
|
|
|
|
manager = SkillsManager("~/.aetheel/workspace")
|
|
manager.load_all()
|
|
context = manager.get_context("what's the weather like?")
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger("aetheel.skills")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Types
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class Skill:
|
|
"""A loaded skill with metadata and instructions."""
|
|
|
|
name: str
|
|
description: str
|
|
triggers: list[str]
|
|
body: str # The markdown body (instructions for the agent)
|
|
path: str # Absolute path to the SKILL.md file
|
|
|
|
def matches(self, text: str) -> bool:
|
|
"""Check if any trigger word appears in the given text."""
|
|
text_lower = text.lower()
|
|
return any(trigger.lower() in text_lower for trigger in self.triggers)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Skills Manager
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class SkillsManager:
|
|
"""
|
|
Discovers and manages skills from the workspace.
|
|
|
|
Skills live in {workspace}/skills/<name>/SKILL.md.
|
|
Each SKILL.md has YAML frontmatter (name, description, triggers)
|
|
and a markdown body with instructions for the agent.
|
|
"""
|
|
|
|
def __init__(self, workspace_dir: str):
|
|
self._workspace = os.path.expanduser(workspace_dir)
|
|
self._skills_dir = os.path.join(self._workspace, "skills")
|
|
self._skills: list[Skill] = []
|
|
|
|
@property
|
|
def skills(self) -> list[Skill]:
|
|
"""Return all loaded skills."""
|
|
return list(self._skills)
|
|
|
|
@property
|
|
def skills_dir(self) -> str:
|
|
"""Return the skills directory path."""
|
|
return self._skills_dir
|
|
|
|
def load_all(self) -> list[Skill]:
|
|
"""
|
|
Discover and load all skills from the workspace.
|
|
|
|
Scans {workspace}/skills/ for subdirectories containing SKILL.md.
|
|
Returns the list of loaded skills.
|
|
"""
|
|
self._skills = []
|
|
|
|
if not os.path.isdir(self._skills_dir):
|
|
logger.info(
|
|
f"Skills directory not found: {self._skills_dir} — "
|
|
"no skills loaded. Create it to add skills."
|
|
)
|
|
return []
|
|
|
|
for entry in sorted(os.listdir(self._skills_dir)):
|
|
skill_dir = os.path.join(self._skills_dir, entry)
|
|
if not os.path.isdir(skill_dir):
|
|
continue
|
|
|
|
skill_file = os.path.join(skill_dir, "SKILL.md")
|
|
if not os.path.isfile(skill_file):
|
|
logger.debug(f"Skipping {entry}/ — no SKILL.md found")
|
|
continue
|
|
|
|
try:
|
|
skill = self._parse_skill(skill_file)
|
|
self._skills.append(skill)
|
|
logger.info(
|
|
f"Loaded skill: {skill.name} "
|
|
f"(triggers={skill.triggers})"
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load skill from {skill_file}: {e}")
|
|
|
|
logger.info(f"Loaded {len(self._skills)} skill(s)")
|
|
return list(self._skills)
|
|
|
|
def reload(self) -> list[Skill]:
|
|
"""Reload all skills (same as load_all, but clears cache first)."""
|
|
return self.load_all()
|
|
|
|
def get_relevant(self, message: str) -> list[Skill]:
|
|
"""
|
|
Find skills relevant to the given message.
|
|
Matches trigger words against the message text.
|
|
"""
|
|
return [s for s in self._skills if s.matches(message)]
|
|
|
|
def get_context(self, message: str) -> str:
|
|
"""
|
|
Build a context string for skills relevant to the message.
|
|
|
|
Returns a formatted string ready for injection into the
|
|
system prompt, or empty string if no skills match.
|
|
"""
|
|
relevant = self.get_relevant(message)
|
|
if not relevant:
|
|
return ""
|
|
|
|
sections = ["# Active Skills\n"]
|
|
for skill in relevant:
|
|
sections.append(
|
|
f"## {skill.name}\n"
|
|
f"*{skill.description}*\n\n"
|
|
f"{skill.body}"
|
|
)
|
|
|
|
return "\n\n---\n\n".join(sections)
|
|
|
|
def get_all_context(self) -> str:
|
|
"""
|
|
Build a context string listing ALL loaded skills (for reference).
|
|
|
|
This is a lighter-weight summary — just names and descriptions,
|
|
not full bodies. Used to let the agent know what skills exist.
|
|
"""
|
|
if not self._skills:
|
|
return ""
|
|
|
|
lines = ["# Available Skills\n"]
|
|
for skill in self._skills:
|
|
triggers = ", ".join(skill.triggers[:5])
|
|
lines.append(f"- **{skill.name}**: {skill.description} (triggers: {triggers})")
|
|
|
|
return "\n".join(lines)
|
|
|
|
# -------------------------------------------------------------------
|
|
# Parsing
|
|
# -------------------------------------------------------------------
|
|
|
|
def _parse_skill(self, path: str) -> Skill:
|
|
"""
|
|
Parse a SKILL.md file with YAML frontmatter.
|
|
|
|
Format:
|
|
---
|
|
name: skill_name
|
|
description: What this skill does
|
|
triggers: [word1, word2, word3]
|
|
---
|
|
|
|
# Skill body (markdown instructions)
|
|
"""
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
|
|
# Parse YAML frontmatter (simple regex-based, no PyYAML needed)
|
|
frontmatter, body = self._split_frontmatter(content)
|
|
|
|
name = self._extract_field(frontmatter, "name") or Path(path).parent.name
|
|
description = self._extract_field(frontmatter, "description") or ""
|
|
triggers_raw = self._extract_field(frontmatter, "triggers") or ""
|
|
triggers = self._parse_list(triggers_raw)
|
|
|
|
if not triggers:
|
|
# Fall back to using the skill name as a trigger
|
|
triggers = [name]
|
|
|
|
return Skill(
|
|
name=name,
|
|
description=description,
|
|
triggers=triggers,
|
|
body=body.strip(),
|
|
path=path,
|
|
)
|
|
|
|
@staticmethod
|
|
def _split_frontmatter(content: str) -> tuple[str, str]:
|
|
"""Split YAML frontmatter from markdown body."""
|
|
# Match --- at start, then content, then ---
|
|
match = re.match(
|
|
r"^---\s*\n(.*?)\n---\s*\n(.*)",
|
|
content,
|
|
re.DOTALL,
|
|
)
|
|
if match:
|
|
return match.group(1), match.group(2)
|
|
return "", content
|
|
|
|
@staticmethod
|
|
def _extract_field(frontmatter: str, field_name: str) -> str | None:
|
|
"""Extract a simple field value from YAML frontmatter."""
|
|
pattern = rf"^{re.escape(field_name)}\s*:\s*(.+)$"
|
|
match = re.search(pattern, frontmatter, re.MULTILINE)
|
|
if match:
|
|
value = match.group(1).strip()
|
|
# Remove surrounding quotes if present
|
|
if (value.startswith('"') and value.endswith('"')) or (
|
|
value.startswith("'") and value.endswith("'")
|
|
):
|
|
value = value[1:-1]
|
|
return value
|
|
return None
|
|
|
|
@staticmethod
|
|
def _parse_list(raw: str) -> list[str]:
|
|
"""
|
|
Parse a YAML-like list string into a Python list.
|
|
|
|
Handles both:
|
|
- Inline: [word1, word2, word3]
|
|
- Comma-separated: word1, word2, word3
|
|
"""
|
|
if not raw:
|
|
return []
|
|
|
|
# Remove brackets if present
|
|
raw = raw.strip()
|
|
if raw.startswith("[") and raw.endswith("]"):
|
|
raw = raw[1:-1]
|
|
|
|
# Split by commas and clean up
|
|
items = []
|
|
for item in raw.split(","):
|
|
cleaned = item.strip().strip("'\"")
|
|
if cleaned:
|
|
items.append(cleaned)
|
|
|
|
return items
|