harness_zbt/backend/app/harness/registries.py

"""Prompt、Skill 与 Persona 注册中心。"""

from dataclasses import dataclass
from pathlib import Path

from app.core.errors import AppError

ASSET_ROOT = Path(__file__).resolve().parent / "assets"


@dataclass(frozen=True)
class SkillDefinition:
    id: str
    name: str
    personas: tuple[str, ...]
    tools: tuple[str, ...]
    instructions: str
    source: Path


@dataclass(frozen=True)
class PersonaDefinition:
    id: str
    name: str
    prompt_id: str
    skill_ids: tuple[str, ...]
    tool_names: tuple[str, ...]


class PromptRegistry:
    def __init__(self, root: Path | None = None) -> None:
        self._root = root or ASSET_ROOT / "prompts"
        self._items = {
            path.stem: path.read_text(encoding="utf-8").strip() for path in self._root.glob("*.md")
        }

    def get(self, prompt_id: str) -> str:
        prompt = self._items.get(prompt_id)
        if prompt is None:
            raise AppError(
                "HARNESS_PROMPT_NOT_FOUND",
                f"未找到Prompt：{prompt_id}",
                500,
            )
        return prompt

    def ids(self) -> tuple[str, ...]:
        return tuple(sorted(self._items))


class SkillRegistry:
    """发现标准 ``SKILL.md``，并解析本项目使用的最小前置元数据。"""

    def __init__(self, root: Path | None = None) -> None:
        self._root = root or ASSET_ROOT / "skills"
        self._items: dict[str, SkillDefinition] = {}
        for path in self._root.glob("*/SKILL.md"):
            skill = self._parse(path)
            if skill.id in self._items:
                raise AppError(
                    "HARNESS_SKILL_DUPLICATED",
                    f"Skill标识重复：{skill.id}",
                    500,
                )
            self._items[skill.id] = skill

    def get(self, skill_id: str) -> SkillDefinition:
        skill = self._items.get(skill_id)
        if skill is None:
            raise AppError(
                "HARNESS_SKILL_NOT_FOUND",
                f"未找到Skill：{skill_id}",
                500,
            )
        return skill

    def ids(self) -> tuple[str, ...]:
        return tuple(sorted(self._items))

    @staticmethod
    def _parse(path: Path) -> SkillDefinition:
        raw = path.read_text(encoding="utf-8").strip()
        if not raw.startswith("---\n"):
            raise AppError(
                "HARNESS_SKILL_INVALID",
                f"Skill缺少元数据：{path.parent.name}",
                500,
            )
        _, metadata_text, instructions = raw.split("---", maxsplit=2)
        metadata: dict[str, str] = {}
        for line in metadata_text.strip().splitlines():
            key, separator, value = line.partition(":")
            if separator:
                metadata[key.strip()] = value.strip()
        required = {"id", "name", "personas", "tools"}
        if not required.issubset(metadata):
            raise AppError(
                "HARNESS_SKILL_INVALID",
                f"Skill元数据不完整：{path.parent.name}",
                500,
            )
        return SkillDefinition(
            id=metadata["id"],
            name=metadata["name"],
            personas=tuple(_csv(metadata["personas"])),
            tools=tuple(_csv(metadata["tools"])),
            instructions=instructions.strip(),
            source=path,
        )


class PersonaRegistry:
    def __init__(self) -> None:
        self._items = {
            "customer": PersonaDefinition(
                id="customer",
                name="客户保障顾问",
                prompt_id="customer",
                skill_ids=("insurance-consultation", "enrollment-guidance"),
                tool_names=(
                    "list_available_products",
                    "calculate_insurance_quote",
                    "prepare_enrollment",
                    "list_my_orders",
                    "list_my_policies",
                ),
            ),
            "operation": PersonaDefinition(
                id="operation",
                name="运营数据助手",
                prompt_id="operation",
                skill_ids=("operation-insights",),
                tool_names=(
                    "get_operation_overview",
                    "list_recent_orders",
                    "list_recent_policies",
                    "get_attribution_performance",
                    "list_available_products",
                ),
            ),
        }

    def get(self, persona_id: str) -> PersonaDefinition:
        persona = self._items.get(persona_id)
        if persona is None:
            raise AppError(
                "HARNESS_PERSONA_NOT_FOUND",
                f"未找到Agent人格：{persona_id}",
                500,
            )
        return persona


def _csv(value: str) -> list[str]:
    return [item.strip() for item in value.split(",") if item.strip()]