"""Skills module for DB-GPT agents. This module provides skills loading mechanism for agents, following the progressive disclosure pattern similar to deepagents' SkillsMiddleware. """ import logging import re from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional, Set from dbgpt.core import PromptTemplate logger = logging.getLogger(__name__) MAX_SKILL_FILE_SIZE = 10 * 1024 * 1024 MAX_SKILL_NAME_LENGTH = 64 MAX_SKILL_DESCRIPTION_LENGTH = 1024 @dataclass class SkillMetadata: """Metadata for a skill.""" name: str description: str path: str version: str = "1.0.0" author: Optional[str] = None skill_type: str = "custom" tags: List[str] = field(default_factory=list) license: Optional[str] = None allowed_tools: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary.""" return { "name": self.name, "description": self.description, "path": self.path, "version": self.version, "author": self.author, "skill_type": self.skill_type, "tags": self.tags, "license": self.license, "allowed_tools": self.allowed_tools, } @dataclass class LoadedSkill: """A loaded skill with metadata and optional content.""" def __init__( self, metadata: SkillMetadata, content: Optional[str] = None, ): """Initialize loaded skill. Args: metadata: Skill metadata. content: Full skill content (markdown instructions). """ self._metadata = metadata self._content = content @property def metadata(self) -> SkillMetadata: """Return skill metadata.""" return self._metadata @property def prompt_content(self) -> str: """Get the skill content for prompt injection.""" if self._content is None and self._metadata.path: try: with open(self._metadata.path, "r", encoding="utf-8") as f: self._content = f.read() except Exception as e: logger.error( f"Failed to load skill content from {self._metadata.path}: {e}" ) self._content = ( f"# {self._metadata.name}\n\n{self._metadata.description}" ) return ( self._content or f"# {self._metadata.name}\n\n{self._metadata.description}" ) def get_prompt_template(self) -> PromptTemplate: """Get prompt template with skill instructions. Returns: PromptTemplate with the skill's full instructions. """ return PromptTemplate.from_template(self.prompt_content) class SkillsLoader: """Loader for skills from filesystem sources.""" def __init__(self, sources: List[str]): """Initialize skills loader. Args: sources: List of skill source paths. """ self.sources = sources self._skills: Dict[str, LoadedSkill] = {} self._loaded = False def load_skills(self) -> Dict[str, LoadedSkill]: """Load skills from all configured sources. Skills are loaded in source order with later sources overriding earlier ones if they contain skills with the same name. Returns: Dictionary of loaded skills keyed by name. """ if self._loaded: return self._skills all_skills: Dict[str, LoadedSkill] = {} for source_path in self.sources: source_skills = _list_skills_from_directory(source_path) for skill_metadata in source_skills: loaded_skill = LoadedSkill(metadata=skill_metadata) all_skills[skill_metadata.name] = loaded_skill self._skills = all_skills self._loaded = True return self._skills def get_skill(self, name: str) -> Optional[LoadedSkill]: """Get a skill by name. Args: name: Skill name. Returns: LoadedSkill or None. """ if not self._loaded: self.load_skills() return self._skills.get(name) def list_skills(self) -> List[SkillMetadata]: """List all loaded skills. Returns: List of skill metadata. """ if not self._loaded: self.load_skills() return [skill.metadata for skill in self._skills.values()] def get_skills_by_type(self, skill_type: str) -> List[LoadedSkill]: """Get skills by type. Args: skill_type: Skill type to filter by. Returns: List of matching skills. """ if not self._loaded: self.load_skills() return [ skill for skill in self._skills.values() if skill.metadata.skill_type == skill_type ] def match_skills(self, user_input: str) -> List[LoadedSkill]: """Find skills that match user input based on description. Args: user_input: User input string. Returns: List of matching skills. """ if not self._loaded: self.load_skills() user_input_lower = user_input.lower() matches = [] for skill in self._skills.values(): description_lower = skill.metadata.description.lower() keywords = _extract_keywords(description_lower) for keyword in keywords: if keyword in user_input_lower: matches.append(skill) break return matches def _validate_skill_name(name: str) -> tuple[bool, str]: """Validate skill name. Requirements: - Max 64 characters - Lowercase alphanumeric and hyphens only (a-z, 0-9, -) - Cannot start or end with hyphen - No consecutive hyphens Args: name: Skill name to validate. Returns: (is_valid, error_message) tuple. """ if not name: return False, "name is required" if len(name) > MAX_SKILL_NAME_LENGTH: return False, "name exceeds 64 characters" if not re.match(r"^[a-z0-9]+(-[a-z0-9]+)*$", name): return False, "name must be lowercase alphanumeric with single hyphens only" return True, "" def _parse_skill_metadata( content: str, skill_path: str, directory_name: str ) -> Optional[SkillMetadata]: """Parse YAML frontmatter from SKILL.md content. Args: content: Content of the SKILL.md file. skill_path: Path to the SKILL.md file. directory_name: Name of the parent directory. Returns: SkillMetadata if parsing succeeds, None otherwise. """ if len(content) > MAX_SKILL_FILE_SIZE: logger.warning( "Skipping %s: content too large (%d bytes)", skill_path, len(content) ) return None frontmatter_pattern = r"^---\s*\n(.*?)\n---\s*\n" match = re.match(frontmatter_pattern, content, re.DOTALL) if not match: logger.warning("Skipping %s: no valid YAML frontmatter found", skill_path) return None frontmatter_str = match.group(1) try: import yaml frontmatter_data = yaml.safe_load(frontmatter_str) except ImportError: logger.error("PyYAML not installed, cannot parse SKILL.md files") return None except Exception as e: logger.warning("Failed to parse YAML in %s: %s", skill_path, e) return None if not isinstance(frontmatter_data, dict): logger.warning("Skipping %s: frontmatter is not a mapping", skill_path) return None name = frontmatter_data.get("name") description = frontmatter_data.get("description") if not name or not description: logger.warning( "Skipping %s: missing required 'name' or 'description'", skill_path ) return None is_valid, error = _validate_skill_name(str(name)) if not is_valid: logger.warning( "Skill '%s' in %s does not follow naming convention: %s", name, skill_path, error, ) description_str = str(description).strip() if len(description_str) > MAX_SKILL_DESCRIPTION_LENGTH: logger.warning( "Description exceeds %d characters in %s, truncating", MAX_SKILL_DESCRIPTION_LENGTH, skill_path, ) description_str = description_str[:MAX_SKILL_DESCRIPTION_LENGTH] allowed_tools = [] allowed_tools_value = frontmatter_data.get("allowed-tools") if allowed_tools_value and isinstance(allowed_tools_value, str): allowed_tools = allowed_tools_value.split(" ") return SkillMetadata( name=str(name), description=description_str, path=skill_path, version=frontmatter_data.get("version", "1.0.0"), author=frontmatter_data.get("author", "").strip() or None, license=frontmatter_data.get("license", "").strip() or None, allowed_tools=allowed_tools, skill_type=frontmatter_data.get("skill_type", "custom"), tags=frontmatter_data.get("tags", []), ) def _list_skills_from_directory(source_path: str) -> List[SkillMetadata]: """List all skills from a directory. Args: source_path: Path to the skills directory. Returns: List of skill metadata from successfully parsed SKILL.md files. """ skills: List[SkillMetadata] = [] base_path = Path(source_path) if not base_path.exists() or not base_path.is_dir(): logger.warning("Directory not found: %s", source_path) return skills for skill_dir in base_path.iterdir(): if not skill_dir.is_dir(): continue skill_md_path = skill_dir / "SKILL.md" if not skill_md_path.exists(): continue try: with open(skill_md_path, "r", encoding="utf-8") as f: content = f.read() directory_name = skill_dir.name skill_metadata = _parse_skill_metadata( content=content, skill_path=str(skill_md_path), directory_name=directory_name, ) if skill_metadata: skills.append(skill_metadata) except Exception as e: logger.warning("Failed to load skill from %s: %s", skill_md_path, e) return skills def _extract_keywords(description: str) -> List[str]: """Extract potential trigger keywords from description. Args: description: Skill description. Returns: List of keywords. """ keywords = [] pattern = r"(?:when|use|for|to)\s+(?:the\s+)?(?:user\s+)?(?:asks|requests|wants|needs)?\s*(?:to\s+)?([a-z\s]+?)(?:\s*(?:\.|,|;|or|\(|\)|use when|$))" matches = re.findall(pattern, description, re.IGNORECASE) for match in matches: words = [w.strip() for w in match.split() if len(w.strip()) > 2] keywords.extend(words) return list(set(keywords)) SKILLS_SYSTEM_PROMPT = """ ## Skills System You have access to a skills library that provides specialized capabilities and domain knowledge. {skills_locations} **Available Skills:** {skills_list} **How to Use Skills (Progressive Disclosure):** Skills follow a **progressive disclosure** pattern - you see their name and description above, but only read full instructions when needed: 1. **Recognize when a skill applies**: Check if the user's task matches a skill's description 2. **Read the skill's full instructions**: Use the path shown in the skill list above 3. **Follow the skill's instructions**: SKILL.md contains step-by-step workflows, best practices, and examples 4. **Access supporting files**: Skills may include helper scripts, configs, or reference docs - use absolute paths **When to Use Skills:** - User's request matches a skill's domain (e.g., "research X" -> web-research skill) - You need specialized knowledge or structured workflows - A skill provides proven patterns for complex tasks **Example Workflow:** User: "Can you research the latest developments in quantum computing?" 1. Check available skills -> See "web-research" skill with its path 2. Read the skill using the path shown 3. Follow the skill's research workflow (search -> organize -> synthesize) 4. Use any helper scripts with absolute paths Remember: Skills make you more capable and consistent. When in doubt, check if a skill exists for the task! """ class SkillsMiddleware: """Middleware for loading and exposing agent skills to the system prompt. Loads skills from sources and injects them into the system prompt using progressive disclosure (metadata first, full content on demand). Example: ```python from skills import SkillsMiddleware middleware = SkillsMiddleware( sources=[ "/path/to/skills/user/", "/path/to/skills/project/", ], ) ``` """ def __init__(self, sources: List[str]): """Initialize the skills middleware. Args: sources: List of skill source paths. """ self.sources = sources self._loader = SkillsLoader(sources) def load_skills(self) -> Dict[str, LoadedSkill]: """Load skills from all configured sources. Returns: Dictionary of loaded skills keyed by name. """ return self._loader.load_skills() def get_skill(self, name: str) -> Optional[LoadedSkill]: """Get a skill by name. Args: name: Skill name. Returns: LoadedSkill or None. """ return self._loader.get_skill(name) def list_skills(self) -> List[SkillMetadata]: """List all loaded skills. Returns: List of skill metadata. """ return self._loader.list_skills() def format_skills_locations(self) -> str: """Format skills locations for display in system prompt. Returns: Formatted string of skills locations. """ locations = [] for i, source_path in enumerate(self.sources): name = Path(source_path.rstrip("/")).name.capitalize() suffix = " (higher priority)" if i == len(self.sources) - 1 else "" locations.append(f"**{name} Skills**: `{source_path}`{suffix}") return "\n".join(locations) def format_skills_list(self) -> str: """Format skills metadata for display in system prompt. Returns: Formatted string of skills list. """ skills = self._loader.list_skills() if not skills: paths = [f"`{source_path}`" for source_path in self.sources] return f"(No skills available yet. You can create skills in {' or '.join(paths)})" lines = [] for skill in skills: lines.append(f"- **{skill.name}**: {skill.description}") lines.append(f" -> Read `{skill.path}` for full instructions") return "\n".join(lines) def create_skills_prompt_section(self) -> str: """Create the skills section for the system prompt. Returns: Formatted skills system prompt section. """ skills_locations = self.format_skills_locations() skills_list = self.format_skills_list() return SKILLS_SYSTEM_PROMPT.format( skills_locations=skills_locations, skills_list=skills_list, ) def get_skills_by_type(self, skill_type: str) -> List[LoadedSkill]: """Get skills by type. Args: skill_type: Skill type to filter by. Returns: List of matching skills. """ return self._loader.get_skills_by_type(skill_type) def match_skills(self, user_input: str) -> List[LoadedSkill]: """Find skills that match user input based on description. Args: user_input: User input string. Returns: List of matching skills. """ return self._loader.match_skills(user_input) def create_skills_middleware(sources: List[str]) -> SkillsMiddleware: """Factory function to create a skills middleware instance. Args: sources: List of skill source paths. Returns: SkillsMiddleware instance. """ return SkillsMiddleware(sources=sources)