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)