OpenMesh/coworker/skills/base.py

"""Skill loading — Anthropic SKILL.md format with progressive disclosure.

A skill is a folder containing `SKILL.md` (YAML frontmatter: name, description,
optional allowed-tools) + a markdown body of instructions + optional resources/scripts.

Progressive disclosure: at session start only the catalog (name + description) is injected
into the agent's context; the full body is loaded on demand via the `load_skill` tool.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Optional, Union

import aisuite as ai


@dataclass
class Skill:
    name: str
    description: str
    instructions: str = ""  # full body — loaded on demand
    path: Optional[str] = None
    allowed_tools: list[str] = field(default_factory=list)
    triggers: list[str] = field(default_factory=list)  # 触发关键词，用于自动路由
    title: str = ""  # 中文标题，用于 UI 显示（如 "办公可视化看板"）


class SkillLoader:
    def __init__(self, dirs: list[str | Path]) -> None:
        self._dirs = [Path(d) for d in dirs]
        self._skills: dict[str, Skill] = {}
        self.rescan()

    def rescan(self) -> None:
        """Re-read the skill dirs. load_skill rescans on a miss so a skill created AFTER
        the session's engine was built is still loadable (the catalog line stays static
        until the next session, but an explicitly requested skill must not 404)."""
        self._skills = {}
        for directory in self._dirs:
            self._discover(directory)

    def _discover(self, directory: Path) -> None:
        if not directory.is_dir():
            return
        for sub in sorted(directory.iterdir()):
            md = sub / "SKILL.md"
            if md.is_file():
                skill = _parse_skill(md)
                self._skills[skill.name] = skill

    def names(self) -> list[str]:
        return list(self._skills)

    def get(self, name: str) -> Optional[Skill]:
        return self._skills.get(name)

    def catalog(self) -> list[dict]:
        # 按优先级排序：office-viz 排最前（数据分析可视化优先），其余按字母顺序
        def sort_key(s: Skill) -> tuple[int, str]:
            if s.name == "office-viz":
                return (0, "")
            return (1, s.name)

        return [
            {"name": s.name, "description": s.description, "title": s.title}
            for s in sorted(self._skills.values(), key=sort_key)
        ]


def _parse_skill(md: Path) -> Skill:
    text = md.read_text(encoding="utf-8")
    name, description, allowed, body = md.parent.name, "", [], text
    triggers: list[str] = []
    title: str = ""
    if text.startswith("---"):
        end = text.find("\n---", 3)
        if end != -1:
            frontmatter = text[3:end]
            body = text[end + 4 :].lstrip("\n")
            for line in frontmatter.splitlines():
                if ":" not in line:
                    continue
                key, value = line.split(":", 1)
                key, value = key.strip().lower(), value.strip()
                if key == "name" and value:
                    name = value
                elif key == "description":
                    description = value
                elif key in ("allowed-tools", "allowed_tools"):
                    allowed = [t.strip() for t in value.split(",") if t.strip()]
                elif key == "title":
                    title = value
                elif key in ("triggers", "trigger-keywords", "trigger_keywords"):
                    # 支持逗号分隔: triggers: 关键词1, 关键词2, 关键词3
                    raw = value
                    # 去掉首尾可能的引号
                    if raw.startswith('"') and raw.endswith('"'):
                        raw = raw[1:-1]
                    elif raw.startswith("'") and raw.endswith("'"):
                        raw = raw[1:-1]
                    triggers = [t.strip() for t in raw.split(",") if t.strip()]
    return Skill(
        name=name,
        description=description,
        instructions=body.strip(),
        path=str(md.parent),
        allowed_tools=allowed,
        triggers=triggers,
        title=title,
    )


def skill_catalog_text(
    loader: SkillLoader, allowed: Optional[set[str]] = None
) -> str:
    catalog = [
        c for c in loader.catalog() if allowed is None or c["name"] in allowed
    ]
    if not catalog:
        return ""

    lines = [f"- {c['name']}: {c['description']}" for c in catalog]

    # 在列表前加优先级提示
    priority_notice = (
        "⚠️ Skill 选择优先级说明：\n"
        "  • 数据分析 + 可视化 + 看板 → 使用 office-viz（不要用 xlsx）\n"
        "  • 纯文件读写（无分析） → 使用 xlsx\n"
        "  • PDF 操作 → 使用 pdf\n"
        "  • ...\n\n"
    )
    return (
        priority_notice +
        "Available skills — call load_skill(name) to load one's full instructions when "
        "it's relevant to the task:\n" + "\n".join(lines)
    )


AllowedSkills = Union[set, Callable[[], set], None]


def skill_tools(loader: SkillLoader, allowed: AllowedSkills = None) -> list:
    """`allowed` gates load_skill: a set is a build-time snapshot; a CALLABLE is consulted
    on every call — the manager passes one so Settings disables apply to live sessions
    immediately, and skills created after the engine was built are still loadable
    (loader rescans on a miss)."""

    def _allowed_now() -> Optional[set]:
        return allowed() if callable(allowed) else allowed

    def load_skill(name: str) -> dict:
        """Load a skill's full instructions + resources path by name. Call this when a
        skill from the catalog is relevant to the current task."""
        skill = loader.get(name)
        if skill is None:
            loader.rescan()  # created after this session started? pick it up now
            skill = loader.get(name)
        gate = _allowed_now()
        if skill is None or (gate is not None and name not in gate):
            available = sorted(
                n for n in loader.names() if gate is None or n in gate
            )
            return {"error": f"unknown skill: {name}", "available": available}
        return {
            "name": skill.name,
            "instructions": skill.instructions,
            "resources_path": skill.path,
        }

    return [
        ai.tool(
            load_skill,
            metadata=ai.ToolMetadata(
                category="skills", risk_level="low", capabilities=["load_skill"]
            ),
        )
    ]