""" 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()