lambdagent/agentexample/agent67v2/skills/registry.py

"""
agent67v2.skills.registry — 微代理技能注册

将 5 个微代理注册为可复用的 Skill，
任何编排器都能通过 SkillRegistry 发现和调用它们。

Lambda 语义:
  register_all() =
    let code  = Skill("code-agent", from_config("code-agent.yml"))  in
    let shell = Skill("shell-agent", from_config("shell-agent.yml")) in
    let web   = Skill("web-agent", from_config("web-agent.yml"))    in
    let mem   = Skill("memory-agent", from_config("memory-agent.yml")) in
    let sys   = Skill("system-agent", from_config("system-agent.yml")) in
    Γ_skills = {code, shell, web, mem, sys}

复用场景:
  - agent67v2 协调者通过 Handoff 调用
  - research67 编排器可以复用 code-agent 做代码实验
  - agentbuilder67 可以复用 web-agent 搜索参考配置
  - 新的编排器只需 SkillRegistry().get("code-agent") 即可使用
"""
from __future__ import annotations

import sys
from pathlib import Path
from typing import Dict, Optional

# 路径设置
AGENT_DIR = Path(__file__).resolve().parent.parent  # agent67v2/
AGENTS_DIR = AGENT_DIR / "agents"
PROJECT_ROOT = AGENT_DIR.parent.parent  # lambdagentpaas/

sys.path.insert(0, str(PROJECT_ROOT))

from lambdagent.skills import Skill, SkillSignature, SkillPack, SkillRegistry
from lambdagent.core import Term, Context


# ════════════════════════════════════════════════════════════
# SubAgentTerm: 子代理作为 Lambda Term
# ════════════════════════════════════════════════════════════

class SubAgentTerm(Term):
    """
    子代理的 Lambda 封装。

    将 YAML 配置定义的子代理包装为一个可被 Skill/Handoff 调用的 Term。
    实际执行时通过 from_config() 编译 YAML → 执行 ReAct 循环。

    Lambda 语义:
        SubAgentTerm(config_path) = λtask. from_config(config_path)(task)
    """

    def __init__(self, name: str, config_path: Path):
        super().__init__(name)
        self.config_path = config_path
        self._compiled = None  # 延迟编译

    def _ensure_compiled(self):
        """延迟编译: 只在第一次调用时加载 YAML"""
        if self._compiled is None:
            try:
                from lambdagent.fromconfig.compiler import from_config
                self._compiled = from_config(str(self.config_path))
            except Exception as e:
                # 编译失败时 fallback: 返回错误信息
                self._compiled = None
                self._compile_error = str(e)

    def apply(self, input_val: any, ctx: Context | None = None) -> any:
        ctx = ctx or Context()

        # 解析输入
        if isinstance(input_val, str):
            try:
                import json
                data = json.loads(input_val)
                task = data.get("task", input_val)
            except (json.JSONDecodeError, AttributeError):
                task = input_val
        elif isinstance(input_val, dict):
            task = input_val.get("task", str(input_val))
        else:
            task = str(input_val)

        import time
        t0 = time.time()

        self._ensure_compiled()

        if self._compiled is not None:
            result = self._compiled.apply(task, ctx)
        else:
            # 编译失败，直接返回错误
            result = f"[{self._name}] 编译失败: {getattr(self, '_compile_error', 'unknown')}"

        elapsed = (time.time() - t0) * 1000
        ctx.log(f"SubAgent:{self._name}", self._trace_id, str(task)[:100],
                str(result)[:100], elapsed)
        return result


# ════════════════════════════════════════════════════════════
# 微代理定义
# ════════════════════════════════════════════════════════════

AGENT_DEFINITIONS = {
    "code-agent": {
        "config": "code-agent.yml",
        "description": "代码文件操作与 Git 版本控制专家",
        "tags": ["code", "file", "git", "test", "编程", "代码"],
    },
    "shell-agent": {
        "config": "shell-agent.yml",
        "description": "终端命令执行专家，安全可控的 Shell 操作",
        "tags": ["shell", "bash", "terminal", "命令", "终端", "脚本"],
    },
    "web-agent": {
        "config": "web-agent.yml",
        "description": "网络搜索、知识库管理与文档生成专家",
        "tags": ["web", "search", "knowledge", "doc", "网络", "搜索", "知识库"],
    },
    "memory-agent": {
        "config": "memory-agent.yml",
        "description": "记忆存储、任务管理、定时调度与用户画像专家",
        "tags": ["memory", "task", "schedule", "notify", "记忆", "任务", "调度"],
    },
    "system-agent": {
        "config": "system-agent.yml",
        "description": "macOS 系统控制: 浏览器、应用、系统信息、截图",
        "tags": ["system", "browser", "app", "screenshot", "macOS", "系统"],
    },
}


# 自动注册 research skill pack (如果可用)
def _register_research_skills():
    """注册 research skill pack 到 SkillRegistry"""
    try:
        from lambdagent.skillpacks.research import register_all as register_research
        register_research()
    except ImportError:
        pass  # research skill pack 不可用时静默跳过


# ════════════════════════════════════════════════════════════
# 注册函数
# ════════════════════════════════════════════════════════════

def create_agent_skill(name: str, definition: dict) -> Skill:
    """创建一个微代理的 Skill"""
    config_path = AGENTS_DIR / definition["config"]

    term = SubAgentTerm(name, config_path)

    return Skill(
        name=name,
        term=term,
        description=definition["description"],
        signature=SkillSignature(input_type="Str", output_type="Str"),
        tags=definition["tags"],
        version="2.0.0",
        author="agent67v2",
    )


def register_all() -> SkillPack:
    """
    注册所有微代理为 Skill，返回 SkillPack。

    调用后，任何地方都可以通过以下方式使用:
        registry = SkillRegistry()
        code_agent = registry.get("code-agent")
        result = code_agent.apply("读取 main.py")
    """
    pack = SkillPack(
        name="agent67v2",
        description="agent67v2 多智能体协作系统的 5 个专业微代理",
        version="2.0.0",
        author="agent67v2",
    )

    for name, definition in AGENT_DEFINITIONS.items():
        skill = create_agent_skill(name, definition)
        pack.add(skill)

    # 注册到全局 Registry
    registry = SkillRegistry()
    registry.register_pack(pack)

    # 同时注册 research skill pack
    _register_research_skills()

    return pack


def get_agent_skill(name: str) -> Optional[Skill]:
    """获取单个微代理的 Skill"""
    registry = SkillRegistry()
    skill = registry.get(name)
    if skill is None:
        # 尝试注册后获取
        register_all()
        skill = registry.get(name)
    return skill


def build_handoff_registry() -> Dict[str, Term]:
    """
    构建 Handoff 路由表。

    返回 {agent_name: Term} 字典，可直接传给 Handoff 构造。
    """
    register_all()
    registry = SkillRegistry()
    return {
        name: registry.get(name)
        for name in AGENT_DEFINITIONS.keys()
        if registry.get(name) is not None
    }


# ════════════════════════════════════════════════════════════
# 自动注册 (import 时执行)
# ════════════════════════════════════════════════════════════

_pack = None

def ensure_registered() -> SkillPack:
    """确保微代理已注册 (幂等)"""
    global _pack
    if _pack is None:
        _pack = register_all()
    return _pack