mirror of
https://gitee.com/wanwujie/deer-flow
synced 2026-04-25 07:04:44 +08:00
feat: add skills system for specialized agent workflows (#6)
Implement a skills framework that enables specialized workflows for specific tasks (e.g., PDF processing, web page generation). Skills are discovered from the skills/ directory and automatically mounted in sandboxes with path mapping support. - Add SkillsConfig for configuring skills path and container mount point - Implement dynamic skill loading from SKILL.md files with YAML frontmatter - Add path mapping in LocalSandbox to translate container paths to local paths - Mount skills directory in AIO Docker sandbox containers - Update lead agent prompt to dynamically inject available skills - Add setup documentation and expand config.example.yaml Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
4
backend/src/skills/__init__.py
Normal file
4
backend/src/skills/__init__.py
Normal file
@@ -0,0 +1,4 @@
|
||||
from .loader import get_skills_root_path, load_skills
|
||||
from .types import Skill
|
||||
|
||||
__all__ = ["load_skills", "get_skills_root_path", "Skill"]
|
||||
77
backend/src/skills/loader.py
Normal file
77
backend/src/skills/loader.py
Normal file
@@ -0,0 +1,77 @@
|
||||
from pathlib import Path
|
||||
|
||||
from .parser import parse_skill_file
|
||||
from .types import Skill
|
||||
|
||||
|
||||
def get_skills_root_path() -> Path:
|
||||
"""
|
||||
Get the root path of the skills directory.
|
||||
|
||||
Returns:
|
||||
Path to the skills directory (deer-flow/skills)
|
||||
"""
|
||||
# backend directory is current file's parent's parent's parent
|
||||
backend_dir = Path(__file__).resolve().parent.parent.parent
|
||||
# skills directory is sibling to backend directory
|
||||
skills_dir = backend_dir.parent / "skills"
|
||||
return skills_dir
|
||||
|
||||
|
||||
def load_skills(skills_path: Path | None = None, use_config: bool = True) -> list[Skill]:
|
||||
"""
|
||||
Load all skills from the skills directory.
|
||||
|
||||
Scans both public and custom skill directories, parsing SKILL.md files
|
||||
to extract metadata.
|
||||
|
||||
Args:
|
||||
skills_path: Optional custom path to skills directory.
|
||||
If not provided and use_config is True, uses path from config.
|
||||
Otherwise defaults to deer-flow/skills
|
||||
use_config: Whether to load skills path from config (default: True)
|
||||
|
||||
Returns:
|
||||
List of Skill objects, sorted by name
|
||||
"""
|
||||
if skills_path is None:
|
||||
if use_config:
|
||||
try:
|
||||
from src.config import get_app_config
|
||||
|
||||
config = get_app_config()
|
||||
skills_path = config.skills.get_skills_path()
|
||||
except Exception:
|
||||
# Fallback to default if config fails
|
||||
skills_path = get_skills_root_path()
|
||||
else:
|
||||
skills_path = get_skills_root_path()
|
||||
|
||||
if not skills_path.exists():
|
||||
return []
|
||||
|
||||
skills = []
|
||||
|
||||
# Scan public and custom directories
|
||||
for category in ["public", "custom"]:
|
||||
category_path = skills_path / category
|
||||
if not category_path.exists() or not category_path.is_dir():
|
||||
continue
|
||||
|
||||
# Each subdirectory is a potential skill
|
||||
for skill_dir in category_path.iterdir():
|
||||
if not skill_dir.is_dir():
|
||||
continue
|
||||
|
||||
skill_file = skill_dir / "SKILL.md"
|
||||
if not skill_file.exists():
|
||||
continue
|
||||
|
||||
skill = parse_skill_file(skill_file, category=category)
|
||||
if skill:
|
||||
skills.append(skill)
|
||||
|
||||
# Sort by name for consistent ordering
|
||||
skills.sort(key=lambda s: s.name)
|
||||
|
||||
return skills
|
||||
63
backend/src/skills/parser.py
Normal file
63
backend/src/skills/parser.py
Normal file
@@ -0,0 +1,63 @@
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from .types import Skill
|
||||
|
||||
|
||||
def parse_skill_file(skill_file: Path, category: str) -> Skill | None:
|
||||
"""
|
||||
Parse a SKILL.md file and extract metadata.
|
||||
|
||||
Args:
|
||||
skill_file: Path to the SKILL.md file
|
||||
category: Category of the skill ('public' or 'custom')
|
||||
|
||||
Returns:
|
||||
Skill object if parsing succeeds, None otherwise
|
||||
"""
|
||||
if not skill_file.exists() or skill_file.name != "SKILL.md":
|
||||
return None
|
||||
|
||||
try:
|
||||
content = skill_file.read_text(encoding="utf-8")
|
||||
|
||||
# Extract YAML front matter
|
||||
# Pattern: ---\nkey: value\n---
|
||||
front_matter_match = re.match(r"^---\s*\n(.*?)\n---\s*\n", content, re.DOTALL)
|
||||
|
||||
if not front_matter_match:
|
||||
return None
|
||||
|
||||
front_matter = front_matter_match.group(1)
|
||||
|
||||
# Parse YAML front matter (simple key-value parsing)
|
||||
metadata = {}
|
||||
for line in front_matter.split("\n"):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
if ":" in line:
|
||||
key, value = line.split(":", 1)
|
||||
metadata[key.strip()] = value.strip()
|
||||
|
||||
# Extract required fields
|
||||
name = metadata.get("name")
|
||||
description = metadata.get("description")
|
||||
|
||||
if not name or not description:
|
||||
return None
|
||||
|
||||
license_text = metadata.get("license")
|
||||
|
||||
return Skill(
|
||||
name=name,
|
||||
description=description,
|
||||
license=license_text,
|
||||
skill_dir=skill_file.parent,
|
||||
skill_file=skill_file,
|
||||
category=category,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error parsing skill file {skill_file}: {e}")
|
||||
return None
|
||||
34
backend/src/skills/types.py
Normal file
34
backend/src/skills/types.py
Normal file
@@ -0,0 +1,34 @@
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@dataclass
|
||||
class Skill:
|
||||
"""Represents a skill with its metadata and file path"""
|
||||
|
||||
name: str
|
||||
description: str
|
||||
license: str | None
|
||||
skill_dir: Path
|
||||
skill_file: Path
|
||||
category: str # 'public' or 'custom'
|
||||
|
||||
@property
|
||||
def skill_path(self) -> str:
|
||||
"""Returns the relative path from skills root to this skill's directory"""
|
||||
return self.skill_dir.name
|
||||
|
||||
def get_container_path(self, container_base_path: str = "/mnt/skills") -> str:
|
||||
"""
|
||||
Get the full path to this skill in the container.
|
||||
|
||||
Args:
|
||||
container_base_path: Base path where skills are mounted in the container
|
||||
|
||||
Returns:
|
||||
Full container path to the skill directory
|
||||
"""
|
||||
return f"{container_base_path}/{self.category}/{self.skill_dir.name}"
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"Skill(name={self.name!r}, description={self.description!r}, category={self.category!r})"
|
||||
Reference in New Issue
Block a user