From 6e4abd226cc021967b9d29beca9b9d266a190aba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=93=E9=97=AE?= Date: Mon, 22 Dec 2025 16:02:48 +0800 Subject: [PATCH] add `include_skills` param --- .../aworld_cli/core/markdown_agent_loader.py | 17 +++++- aworld/utils/skill_loader.py | 57 ++++++++++++++++++- 2 files changed, 70 insertions(+), 4 deletions(-) diff --git a/aworld/experimental/aworld_cli/core/markdown_agent_loader.py b/aworld/experimental/aworld_cli/core/markdown_agent_loader.py index edea4ce0f..56a772710 100644 --- a/aworld/experimental/aworld_cli/core/markdown_agent_loader.py +++ b/aworld/experimental/aworld_cli/core/markdown_agent_loader.py @@ -245,6 +245,10 @@ def parse_markdown_agent(md_file_path: Path) -> Optional[LocalAgent]: - skills_path: Path to directory containing skill.md files (optional, e.g., "../skills"). Skills will be automatically collected from subdirectories containing skill.md files. Path can be relative (to markdown file directory) or absolute. + - include_skills: Specify which skills to include (optional). + - Comma-separated list: "screenshot,notify" (exact match for each name) + - Regex pattern: "screen.*" (pattern match) + - If not specified, uses INCLUDE_SKILLS environment variable or loads all skills The markdown body content will be used as part of the system prompt. @@ -376,8 +380,17 @@ def parse_markdown_agent(md_file_path: Path) -> Optional[LocalAgent]: if not skills_dir.exists(): logger.warning(f"⚠️ Skills directory not found: {skills_dir}, skipping skill collection") else: - # Collect skills from the directory - collected_skills = collect_skill_docs(skills_dir) + # Get include_skills from front_matter or environment variable + include_skills = front_matter.get("include_skills") + if include_skills is None: + # Support both new and old environment variable names for backward compatibility + include_skills = os.environ.get("INCLUDE_SKILLS") or os.environ.get("SKILL_FILTER") + + if include_skills: + logger.info(f"🔍 Including skills: {include_skills}") + + # Collect skills from the directory with optional filter + collected_skills = collect_skill_docs(skills_dir, include_skills=include_skills) if collected_skills: # Convert collected skills to the format expected by AgentConfig # collect_skill_docs returns: {skill_name: {name, description, tool_list, usage, type, active, skill_path}} diff --git a/aworld/utils/skill_loader.py b/aworld/utils/skill_loader.py index 7a36a8ee1..4d3a4176c 100644 --- a/aworld/utils/skill_loader.py +++ b/aworld/utils/skill_loader.py @@ -6,8 +6,9 @@ from __future__ import annotations import json +import re from pathlib import Path -from typing import Any, Dict, List, Tuple, Union +from typing import Any, Dict, List, Optional, Tuple, Union from aworld.logs.util import logger def extract_front_matter(content_lines: List[str]) -> Tuple[Dict[str, Any], int]: @@ -57,12 +58,20 @@ def extract_front_matter(content_lines: List[str]) -> Tuple[Dict[str, Any], int] return front_matter, end_index + 1 -def collect_skill_docs(root_path: Union[str, Path]) -> Dict[str, Dict[str, Any]]: +def collect_skill_docs( + root_path: Union[str, Path], + include_skills: Optional[Union[str, List[str]]] = None +) -> Dict[str, Dict[str, Any]]: """ Collect skill documentation metadata from all subdirectories containing skill.md files. Args: root_path (Union[str, Path]): Root directory to search for skill documentation files. + include_skills (Optional[Union[str, List[str]]]): Specify which skills to include. + - Comma-separated string: "screenshot,notify" (exact match for each name) + - Regex pattern string: "screen.*" (pattern match) + - List of strings: ["screenshot", "notify"] or ["screen.*"] (mix of exact and regex) + - If None: collect all skills Returns: Dict[str, Dict[str, Any]]: Mapping from skill names to metadata containing @@ -71,6 +80,10 @@ def collect_skill_docs(root_path: Union[str, Path]) -> Dict[str, Dict[str, Any]] Example: >>> collect_skill_docs(Path(".")) {'tts': {'name': 'tts', 'desc': '...', 'tool_list': {...}, 'usage': '...', 'skill_path': '...'}} + >>> collect_skill_docs(Path("."), "screenshot,notify") + {'screenshot': {...}, 'notify': {...}} + >>> collect_skill_docs(Path("."), "screen.*") + {'screenshot': {...}} """ results: Dict[str, Dict[str, Any]] = {} logger.debug("Starting to collect skill : %s", root_path) @@ -79,6 +92,41 @@ def collect_skill_docs(root_path: Union[str, Path]) -> Dict[str, Dict[str, Any]] else: root_dir = root_path + # Parse include_skills parameter + filter_patterns: List[str] = [] + if include_skills: + if isinstance(include_skills, str): + # Check if it's comma-separated list + if "," in include_skills: + filter_patterns = [pattern.strip() for pattern in include_skills.split(",")] + else: + # Single regex pattern + filter_patterns = [include_skills] + elif isinstance(include_skills, list): + filter_patterns = include_skills + else: + logger.warning(f"⚠️ Invalid include_skills type: {type(include_skills)}, ignoring filter") + filter_patterns = [] + + def should_include_skill(skill_name: str) -> bool: + """Check if skill should be included based on filter patterns""" + if not filter_patterns: + return True + + for pattern in filter_patterns: + # Try exact match first + if pattern == skill_name: + return True + # Try regex match + try: + if re.match(pattern, skill_name): + return True + except re.error as e: + logger.warning(f"⚠️ Invalid regex pattern '{pattern}': {e}, treating as exact match") + if pattern == skill_name: + return True + return False + for skill_file in root_dir.glob("**/skill.md"): logger.debug("Finished collecting skill: %s", skill_file) content = skill_file.read_text(encoding="utf-8").splitlines() @@ -95,6 +143,11 @@ def collect_skill_docs(root_path: Union[str, Path]) -> Dict[str, Dict[str, Any]] tool_list = {} skill_name = skill_file.parent.name + + # Apply filter + if not should_include_skill(skill_name): + logger.debug(f"⏭️ Skipping skill '{skill_name}' (not matching filter)") + continue results[skill_name] = { "name": skill_name,