nanobot 渐进式技能加载机制深度分析

def get_always_skills(self) -> list[str]:
    """Get skills marked as always=true that meet requirements."""
    result = []
    for s in self.list_skills(filter_unavailable=True):
        meta = self.get_skill_metadata(s["name"]) or {}
        skill_meta = self._parse_nanobot_metadata(meta.get("metadata", ""))
        if skill_meta.get("always") or meta.get("always"):
            result.append(s["name"])
    return result

def load_skills_for_context(self, skill_names: list[str]) -> str:
    """Load specific skills for inclusion in agent context."""
    parts = []
    for name in skill_names:
        content = self.load_skill(name)
        if content:
            content = self._strip_frontmatter(content)
            parts.append(f"### Skill: {name}\n\n{content}")
    return "\n\n---\n\n".join(parts) if parts else ""

# Skills - progressive loading
# 1. Always-loaded skills: include full content
always_skills = self.skills.get_always_skills()
if always_skills:
    always_content = self.skills.load_skills_for_context(always_skills)
    if always_content:
        parts.append(f"# Active Skills\n\n{always_content}")

def build_skills_summary(self) -> str:
    """Build a summary of all skills (name, description, path, availability)."""
    all_skills = self.list_skills(filter_unavailable=False)
    if not all_skills:
        return ""

    def escape_xml(s: str) -> str:
        return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")

    lines = ["<skills>"]
    for s in all_skills:
        name = escape_xml(s["name"])
        path = s["path"]
        desc = escape_xml(self._get_skill_description(s["name"]))
        skill_meta = self._get_skill_meta(s["name"])
        available = self._check_requirements(skill_meta)

        lines.append(f"  <skill available=\"{str(available).lower()}\">")
        lines.append(f"    <name>{name}</name>")
        lines.append(f"    <description>{desc}</description>")
        lines.append(f"    <location>{path}</location>")

        if not available:
            missing = self._get_missing_requirements(skill_meta)
            if missing:
                lines.append(f"    <requires>{escape_xml(missing)}</requires>")

        lines.append(f"  </skill>")
    lines.append("</skills>")

    return "\n".join(lines)

# 2. Available skills: only show summary (agent uses read_file to load)
skills_summary = self.skills.build_skills_summary()
if skills_summary:
    parts.append(f"""# Skills

The following skills extend your capabilities. To use a skill, read its SKILL.md file using the read_file tool.
Skills with available="false" need dependencies installed first - you can try installing them with apt/brew.

{skills_summary}""")

def _get_skill_meta(self, name: str) -> dict:
    """Get nanobot metadata for a skill (cached in frontmatter)."""
    meta = self.get_skill_metadata(name) or {}
    return self._parse_nanobot_metadata(meta.get("metadata", ""))

def get_skill_metadata(self, name: str) -> dict | None:
    """Get metadata from a skill's frontmatter."""
    content = self.load_skill(name)
    if not content:
        return None

    if content.startswith("---"):
        match = re.match(r"^---\n(.*?)\n---", content, re.DOTALL)
        if match:
            # Simple YAML parsing
            metadata = {}
            for line in match.group(1).split("\n"):
                if ":" in line:
                    key, value = line.split(":", 1)
                    metadata[key.strip()] = value.strip().strip('"\'')
            return metadata

    return None

{
    "name": "github",
    "description": "Interact with GitHub using the `gh` CLI...",
    "homepage": "https://github.com/cli/cli",
    "metadata": '{"nanobot":{"emoji":"🐙","requires":{"bins":["gh"]}}}'
}

def _parse_nanobot_metadata(self, raw: str) -> dict:
    """Parse nanobot metadata JSON from frontmatter."""
    try:
        data = json.loads(raw)
        return data.get("nanobot", {}) if isinstance(data, dict) else {}
    except (json.JSONDecodeError, TypeError):
        return {}

def _check_requirements(self, skill_meta: dict) -> bool:
    """Check if skill requirements are met (bins, env vars)."""
    requires = skill_meta.get("requires", {})
    for b in requires.get("bins", []):
        if not shutil.which(b):
            return False
    for env in requires.get("env", []):
        if not os.environ.get(env):
            return False
    return True

def list_skills(self, filter_unavailable: bool = True) -> list[dict[str, str]]:
    skills = []

    # Workspace skills (highest priority)
    if self.workspace_skills.exists():
        for skill_dir in self.workspace_skills.iterdir():
            if skill_dir.is_dir():
                skill_file = skill_dir / "SKILL.md"
                if skill_file.exists():
                    skills.append({"name": skill_dir.name, "path": str(skill_file), "source": "workspace"})

    # Built-in skills
    if self.builtin_skills and self.builtin_skills.exists():
        for skill_dir in self.builtin_skills.iterdir():
            if skill_dir.is_dir():
                skill_file = skill_dir / "SKILL.md"
                if skill_file.exists() and not any(s["name"] == skill_dir.name for s in skills):
                    skills.append({"name": skill_dir.name, "path": str(skill_file), "source": "builtin"})
    ...

def load_skill(self, name: str) -> str | None:
    # Check workspace first
    workspace_skill = self.workspace_skills / name / "SKILL.md"
    if workspace_skill.exists():
        return workspace_skill.read_text(encoding="utf-8")

    # Check built-in
    if self.builtin_skills:
        builtin_skill = self.builtin_skills / name / "SKILL.md"
        if builtin_skill.exists():
            return builtin_skill.read_text(encoding="utf-8")

    return None

class SkillsLoader:
    def __init__(self, workspace: Path):
        self.workspace = workspace
        self._cache = {}  # {skill_name: (content, mtime)}

    def load_skill(self, name: str) -> str | None:
        path = self._find_skill_path(name)
        if not path:
            return None

        mtime = path.stat().st_mtime
        cached = self._cache.get(name)

        if cached and cached[1] == mtime:
            return cached[0]

        content = path.read_text(encoding="utf-8")
        self._cache[name] = (content, mtime)
        return content

# workspace/skills/index.json
{
  "version": 1,
  "last_updated": "2026-02-03T12:00:00Z",
  "skills": {
    "github": {
      "name": "github",
      "description": "Interact with GitHub...",
      "metadata": {...},
      "available": true,
      "path": "workspace/skills/github/SKILL.md",
      "mtime": 1738584000
    }
  }
}

from concurrent.futures import ThreadPoolExecutor

def load_skills_for_context(self, skill_names: list[str]) -> str:
    parts = []

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = {
            executor.submit(self._load_single_skill, name): name
            for name in skill_names
        }
        for future in as_completed(futures):
            name = futures[future]
            try:
                content = future.result()
                if content:
                    content = self._strip_frontmatter(content)
                    parts.append(f"### Skill: {name}\n\n{content}")
            except Exception as e:
                # Log error but continue
                pass

    return "\n\n---\n\n".join(parts) if parts else ""

class SkillsLoader:
    def __init__(self, workspace: Path):
        self.workspace = workspace
        self._metadata_cache: dict[str, tuple[dict, float]] = {}  # {name: (metadata, mtime)}

    def _get_skill_meta_cached(self, name: str) -> dict:
        path = self._find_skill_path(name)
        if not path:
            return {}

        mtime = path.stat().st_mtime
        cached = self._metadata_cache.get(name)

        if cached and cached[1] == mtime:
            return cached[0]

        meta = self._get_skill_meta(name)
        self._metadata_cache[name] = (meta, mtime)
        return meta

def resolve_dependencies(self, skill_name: str) -> list[dict]:
    """Resolve and suggest installation commands for missing dependencies."""
    meta = self._get_skill_meta(skill_name)
    missing = []

    # Check CLI tools
    for bin_name in meta.get("requires", {}).get("bins", []):
        if not shutil.which(bin_name):
            missing.append({"type": "bin", "name": bin_name})

    # Suggest installation
    suggestions = []
    for install in meta.get("install", []):
        if any(m["name"] in install.get("bins", []) for m in missing):
            suggestions.append(install)

    return suggestions

def update_skill(self, name: str, force: bool = False) -> bool:
    """Update a skill from the marketplace."""
    current_version = self._get_skill_version(name)
    latest_version = self._marketplace.get_latest_version(name)

    if not force and current_version == latest_version:
        return False

    skill_content = self._marketplace.download_skill(name, latest_version)
    self._save_skill(name, skill_content)
    return True

class SkillMarketplace:
    def search_skills(self, query: str, category: str | None = None) -> list[dict]:
        """Search for skills in the marketplace."""
        pass

    def get_skill_info(self, name: str) -> dict:
        """Get detailed information about a skill."""
        pass

    def download_skill(self, name: str, version: str | None = None) -> bytes:
        """Download a skill package (.skill file)."""
        pass

    def install_skill(self, name: str, version: str | None = None) -> str:
        """Install a skill to workspace."""
        pass

    def rate_skill(self, name: str, rating: int, comment: str) -> bool:
        """Rate and review a skill."""
        pass

Skills with available="false" need dependencies installed first - you can try installing them with apt/brew.

# Show missing requirements for unavailable skills
if not available:
    missing = self._get_missing_requirements(skill_meta)
    if missing:
        lines.append(f"    <requires>{escape_xml(missing)}</requires>")