feat: skills wip

dreadnode/agent/agent.py

@@ -44,6 +44,7 @@
 from dreadnode.agent.tools import AnyTool, Tool, Toolset, discover_tools_on_obj
 from dreadnode.agent.tools.base import ToolMode
 from dreadnode.agent.tools.planning import update_todo
+from dreadnode.agent.tools.skills import Skills
 from dreadnode.agent.tools.tasking import finish_task, give_up_on_task
 from dreadnode.meta import Component, Config, Model, component
 from dreadnode.meta.introspect import get_config_model, get_inputs_and_params_from_config_model
@@ -129,6 +130,15 @@ def validate_tools(cls, value: t.Any) -> t.Any:
 
         return tools
 
+    def model_post_init(self, _context: t.Any) -> None:
+        """Initialize the agent and inject default tools."""
+        if (
+            not any(isinstance(t, Skills) for t in self.tools)
+            and not any(t.name == "view_skill" for t in self.all_tools)
+            and Skills.load("skills")
+        ):
+            self.tools.append(Skills())
+
     def __repr__(self) -> str:
         description = shorten_string(self.description or "", 50)
 
@@ -287,7 +297,7 @@ async def _generate(
         params = rg.GenerateParams(
             tools=[tool.api_definition for tool in self.all_tools],
         )
-        messages = inject_system_content(messages, self.get_prompt())
+        messages = inject_system_content(messages, self._get_system_prompt())
 
         if self.tool_mode == "auto" and self.tools:
             self.tool_mode = (
@@ -844,7 +854,9 @@ async def _stream_traced(
                 if last_event is not None:
                     # TODO(nick): Don't love having to inject here, but it's the most accurate in
                     # in terms of ensuring we don't miss the system component of messages
-                    final_messages = inject_system_content(last_event.messages, self.get_prompt())
+                    final_messages = inject_system_content(
+                        last_event.messages, self._get_system_prompt()
+                    )
                     log_outputs(messages=final_messages, token_usage=last_event.total_usage)
 
                 if isinstance(last_event, AgentEnd):
@@ -872,6 +884,35 @@ def get_prompt(self) -> str:
             prompt += f"\n\n<instructions>\n{self.instructions}\n</instructions>"
         return prompt
 
+    def get_skills_prompt(self) -> str:
+        """
+        Generates the skills portion of the prompt.
+        """
+        all_skills = []
+        for toolset in self.tools:
+            if isinstance(toolset, Skills):
+                all_skills.extend(Skills.load(toolset.skills_dir))
+
+        if all_skills:
+            skills_xml = ["<available_skills>"]
+            for skill in all_skills:
+                skills_xml.append("<skill>")
+                skills_xml.append(f"<name>{skill.name}</name>")
+                skills_xml.append(f"<description>{skill.description}</description>")
+                skills_xml.append("</skill>")
+            skills_xml.append("</available_skills>")
+            return "\n".join(skills_xml)
+        return ""
+
+    def _get_system_prompt(self) -> str:
+        """
+        Combines the agent's prompt and skills into a single system prompt.
+        """
+        prompt = self.get_prompt()
+        if skills_prompt := self.get_skills_prompt():
+            prompt += f"\n\n{skills_prompt}"
+        return prompt
+
     def reset(self) -> Thread:
         """Reset the agent's internal thread and returns the previous thread."""
         previous = self.thread

dreadnode/agent/tools/__init__.py

@@ -38,7 +38,15 @@
     "tool_method",
 ]
 
-__lazy_submodules__: list[str] = ["fs", "planning", "reporting", "tasking", "execute", "memory"]
+__lazy_submodules__: list[str] = [
+    "fs",
+    "planning",
+    "reporting",
+    "tasking",
+    "execute",
+    "memory",
+    "skills",
+]
 __lazy_components__: dict[str, str] = {}
 
 

dreadnode/agent/tools/skills.py

@@ -0,0 +1,107 @@
+import re
+import typing as t
+from pathlib import Path
+
+import yaml
+from loguru import logger
+from pydantic import Field, model_validator
+
+from dreadnode.agent.tools.base import Toolset, tool_method
+from dreadnode.meta import Config, Model
+
+
+class Skill(Model):
+    """
+    A skill represents a set of instructions or examples that an agent can use.
+    """
+
+    name: str = ""
+    """The name of the skill, derived from the frontmatter or filename."""
+    description: str = ""
+    """A brief description of the skill, derived from the frontmatter or content."""
+    content: str
+    """The content of the skill."""
+    path: Path | None = Field(default=None, exclude=True)
+    """The path to the skill file."""
+
+    @model_validator(mode="before")
+    @classmethod
+    def parse_frontmatter(cls, data: t.Any) -> t.Any:
+        if not isinstance(data, dict) or "content" not in data:
+            return data
+
+        content = data["content"]
+        match = re.match(r"^---\s*\n(.*?)\n---(?:\s*\n|$)", content, re.DOTALL)
+        if match:
+            try:
+                frontmatter = yaml.safe_load(match.group(1))
+                if isinstance(frontmatter, dict):
+                    data["name"] = frontmatter.get("name") or data.get("name")
+                    data["description"] = frontmatter.get("description") or data.get("description")
+                    data["content"] = content[match.end() :].strip()
+            except yaml.YAMLError as e:
+                logger.warning(f"Failed to parse frontmatter in skill: {e}")
+
+        if not data.get("name") and data.get("path"):
+            data["name"] = Path(data["path"]).stem
+
+        if not data.get("description"):
+            for line in data["content"].strip().splitlines():
+                stripped = line.strip()
+                if not stripped or stripped.startswith("#"):
+                    continue
+                data["description"] = stripped
+                break
+
+        if not data.get("description"):
+            data["description"] = "No description available."
+
+        return data
+
+
+class Skills(Toolset):
+    """Tools for interacting with agent skills."""
+
+    skills_dir: str = Config(default="skills", expose_as=str)
+    """The directory where the skills are stored."""
+
+    @tool_method
+    def view_skill(self, name: str) -> str:
+        """
+        View the detailed content of a skill by name.
+
+        Args:
+            name: The name of the skill to view.
+        """
+        skills = self.load(self.skills_dir)
+        skill = next((s for s in skills if s.name == name), None)
+
+        if skill:
+            return f"# Skill: {skill.name}\n\n{skill.content}"
+
+        return f"Skill '{name}' not found."
+
+    @staticmethod
+    def load(directory: str | Path) -> list[Skill]:
+        """
+        Load skills from a directory.
+        """
+        skills_path = Path(directory)
+        if not skills_path.exists() or not skills_path.is_dir():
+            return []
+
+        skills = []
+        for file in skills_path.glob("*"):
+            if file.is_file() and not file.name.startswith("."):
+                try:
+                    skills.append(
+                        Skill(
+                            content=file.read_text(encoding="utf-8"),
+                            path=file,
+                        )
+                    )
+                except Exception as e:  # noqa: BLE001
+                    logger.opt(exception=True).warning(f"Failed to load skill from {file}: {e}")
+                    continue
+
+        return sorted(skills, key=lambda s: s.name)