diff --git a/dreadnode/agent/agent.py b/dreadnode/agent/agent.py
index 79fbb96..4d05fc0 100644
--- a/dreadnode/agent/agent.py
+++ b/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\n{self.instructions}\n"
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 = [""]
+ for skill in all_skills:
+ skills_xml.append("")
+ skills_xml.append(f"{skill.name}")
+ skills_xml.append(f"{skill.description}")
+ skills_xml.append("")
+ skills_xml.append("")
+ 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
diff --git a/dreadnode/agent/tools/__init__.py b/dreadnode/agent/tools/__init__.py
index 42e39a1..9f11cda 100644
--- a/dreadnode/agent/tools/__init__.py
+++ b/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] = {}
diff --git a/dreadnode/agent/tools/skills.py b/dreadnode/agent/tools/skills.py
new file mode 100644
index 0000000..bd1253b
--- /dev/null
+++ b/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)