lambdagent/agentexample/agent67v2/core/bootstrap.py

"""
agent67v2.core.bootstrap — 统一启动引导

将 agent-config.yml 编译为可运行的 Lambda Term，
自动处理 subAgents 节的子代理注册和工具注入。

用法:
  # 方式 1: PaaS CLI
  python3 -m agentpaas agent create --name lambda2 --config agentexample/agent67v2/agent-config.yml
  python3 -m agentpaas chat lambda2

  # 方式 2: Python 直接调用
  from agent67v2.core.bootstrap import build_agent67v2
  agent = build_agent67v2()
  result = agent.apply("帮我看看当前目录")

流程:
  1. 读取 agent-config.yml
  2. 解析 subAgents 节，编译每个子代理
  3. 注册子代理为 Skill 到 SkillRegistry
  4. 创建 call_* 元工具 + ToolSearch
  5. 通过 overrides["tools"] 注入到 from_config()
  6. 返回完整的协调者 Term
"""
from __future__ import annotations

import json
import os
import sys
import yaml
from pathlib import Path
from typing import Any, Dict, Optional

PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent.parent
AGENT_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

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


# ════════════════════════════════════════════════════════════
# SubAgent 编译与注册
# ════════════════════════════════════════════════════════════

def _compile_sub_agent(name: str, config_path: Path) -> Optional[Term]:
    """编译单个子代理 YAML 为 Term"""
    if not config_path.exists():
        print(f"  ⚠️  子代理配置不存在: {config_path}")
        return None

    try:
        from lambdagent.fromconfig.compiler import from_config
        return from_config(str(config_path))
    except Exception as e:
        print(f"  ⚠️  子代理 {name} 编译失败: {e}")
        return None


def _register_sub_agents(config_path: Path, sub_agents_cfg: dict) -> Dict[str, Skill]:
    """
    编译并注册所有子代理为 Skill。

    Args:
        config_path: agent-config.yml 所在目录
        sub_agents_cfg: subAgents 配置节

    Returns:
        {agent_name: Skill} 映射
    """
    skills = {}
    pack = SkillPack(
        name="agent67v2",
        description="agent67v2 多智能体协作系统",
        version="2.0.0",
    )

    for agent_name, agent_def in sub_agents_cfg.items():
        # 解析子代理配置文件路径 (相对于 agent-config.yml 所在目录)
        rel_path = agent_def.get("config", f"agents/{agent_name}.yml")
        agent_yml = config_path / rel_path

        # 编译子代理
        term = _compile_sub_agent(agent_name, agent_yml)
        if term is None:
            continue

        # 创建 Skill
        skill = Skill(
            name=agent_name,
            term=term,
            description=agent_def.get("description", ""),
            signature=SkillSignature(input_type="Str", output_type="Str"),
            tags=agent_def.get("tags", []),
            version="2.0.0",
            author="agent67v2",
        )
        pack.add(skill)
        skills[agent_name] = skill

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

    return skills


# ════════════════════════════════════════════════════════════
# 元工具构建
# ════════════════════════════════════════════════════════════

def _make_agent_caller(skill: Skill):
    """为一个子代理 Skill 创建调用函数"""
    def call_agent(input_str: str) -> str:
        try:
            # 解析输入
            if isinstance(input_str, str):
                try:
                    data = json.loads(input_str)
                    task = data.get("task", input_str)
                except (json.JSONDecodeError, AttributeError):
                    task = input_str
            else:
                task = str(input_str)

            ctx = Context()
            result = skill.apply(task, ctx)
            return str(result)
        except Exception as e:
            return f"[ERROR] {skill._name} 执行失败: {e}"

    return call_agent


def _build_meta_tools(skills: Dict[str, Skill], sub_agents_cfg: dict) -> Dict[str, Any]:
    """
    构建协调者的元工具集合。

    根据 subAgents 配置中的 tool 字段，为每个子代理创建对应的 call_* 函数。
    """
    tools = {}

    for agent_name, agent_def in sub_agents_cfg.items():
        tool_name = agent_def.get("tool", f"call_{agent_name.replace('-agent', '')}")
        skill = skills.get(agent_name)
        if skill:
            tools[tool_name] = _make_agent_caller(skill)

    # ToolSearch 元工具
    from agent67v2.tools.tool_search import tool_search
    tools["ToolSearch"] = lambda x: tool_search.apply(x)

    return tools


# ════════════════════════════════════════════════════════════
# 主入口: build_agent67v2
# ════════════════════════════════════════════════════════════

def build_agent67v2(config_file: str = None, **overrides) -> Term:
    """
    从 agent-config.yml 构建完整的 agent67v2 协调者。

    流程:
      1. 读取 agent-config.yml
      2. 解析 subAgents，编译并注册子代理
      3. 构建 call_* 元工具
      4. 通过 overrides["tools"] 注入到 from_config()
      5. 返回可执行的协调者 Term

    Args:
        config_file: agent-config.yml 路径 (默认: 自动定位)
        **overrides: 传递给 from_config 的覆盖参数

    Returns:
        Term: 可执行的协调者 Lambda Term
    """
    # 定位配置文件
    if config_file:
        config_path = Path(config_file).resolve()
    else:
        config_path = AGENT_DIR / "agent-config.yml"

    if not config_path.exists():
        raise FileNotFoundError(f"配置文件不存在: {config_path}")

    config_dir = config_path.parent

    # 读取配置
    with open(config_path, "r", encoding="utf-8") as f:
        cfg = yaml.safe_load(f)

    # 编译并注册子代理
    sub_agents_cfg = cfg.get("subAgents", {})
    if sub_agents_cfg:
        print(f"  🔄 编译 {len(sub_agents_cfg)} 个子代理...")
        skills = _register_sub_agents(config_dir, sub_agents_cfg)
        print(f"  ✅ {len(skills)} 个子代理已注册为 Skill")

        # 构建元工具
        meta_tools = _build_meta_tools(skills, sub_agents_cfg)

        # 注入到 overrides
        existing_tools = overrides.get("tools", {})
        existing_tools.update(meta_tools)
        overrides["tools"] = existing_tools

    # 通过 from_config 编译协调者
    from lambdagent.fromconfig.compiler import from_config
    agent = from_config(str(config_path), **overrides)

    return agent


def build_and_chat(config_file: str = None, **overrides):
    """
    构建 agent67v2 并进入交互式对话。

    用于 PaaS chat 模式的直接调用。
    """
    agent = build_agent67v2(config_file, **overrides)

    print(f"\n{'═' * 50}")
    print(f"  🐑 lambda v2 — 多智能体协作版")
    print(f"  输入 'exit' 退出")
    print(f"{'═' * 50}\n")

    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n👋 lambda v2 下线了！")
            break

        if not user_input:
            continue
        if user_input.lower() in ("exit", "quit", "bye"):
            print("👋 lambda v2 下线了！")
            break

        print()
        ctx = Context()
        result = agent.apply(user_input, ctx)
        print(f"🐑 lambda v2: {result}")
        print()