latest updates
This commit is contained in:
Tanmay Karande
270
skills/skills.py
Normal file
270
skills/skills.py
Normal file
@@ -0,0 +1,270 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user