lambdagent/agentexample/agent67v2/launch_paas.py

#!/usr/bin/env python3
"""
agent67v2/launch_paas.py — 通过 AgentPaaS 运行 lambda v2 🐑
=============================================================

多智能体版本的 PaaS 部署。注册协调者 + 5个微代理。
与 agent67/launch_paas.py 保持一致的启动流程:
  PaaS 启动 → 创建租户 → 注册代理 → Claude Code CLI 对话 + PaaS 记录

用法:
  # 一键启动（推荐，无需 API Key）
  python launch_paas.py all

  # 分步执行
  python launch_paas.py serve               # 启动 PaaS 服务
  python launch_paas.py register            # 注册全部代理 (协调者 + 5微代理)
  python launch_paas.py chat                # 通过 PaaS 与 lambda v2 对话
  python launch_paas.py chat --claude-code  # PaaS 追踪 + Claude Code 执行
  python launch_paas.py status              # 查看状态
"""
from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.request
import urllib.error
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent  # lambdagentpaas/
AGENT_DIR = Path(__file__).resolve().parent                    # agent67v2/
AGENTS_DIR = AGENT_DIR / "agents"
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(AGENT_DIR.parent))  # agentexample/

PAAS_URL = os.environ.get("AGENTPAAS_URL", "http://127.0.0.1:8000")
API_KEY = os.environ.get("AGENTPAAS_API_KEY", "")


# ═══════════════════════════════════════════════════════════
# PaaS API Helpers (与 agent67 一致)
# ═══════════════════════════════════════════════════════════

def api_call(method: str, path: str, data: dict = None) -> dict:
    """调用 AgentPaaS REST API"""
    url = f"{PAAS_URL}/api/v1{path}"
    body = json.dumps(data).encode("utf-8") if data else None
    req = urllib.request.Request(url, data=body, method=method)
    req.add_header("Content-Type", "application/json")
    req.add_header("Authorization", f"Bearer {API_KEY}")
    try:
        with urllib.request.urlopen(req, timeout=300) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        body = e.read().decode()
        print(f"  ✗ API Error {e.code}: {body[:300]}")
        return {"error": body}
    except urllib.error.URLError as e:
        print(f"  ✗ Connection error: {e}")
        return {"error": str(e)}


def wait_for_server(timeout: int = 30) -> bool:
    """等待 PaaS 服务就绪"""
    for _ in range(timeout):
        try:
            req = urllib.request.Request(f"{PAAS_URL}/health")
            with urllib.request.urlopen(req, timeout=2):
                return True
        except Exception:
            time.sleep(1)
    return False


def load_agent_configs() -> dict:
    """加载所有 YAML 配置"""
    import yaml
    configs = {}
    # 5 个微代理
    for yml_file in sorted(AGENTS_DIR.glob("*.yml")):
        with open(yml_file, "r", encoding="utf-8") as f:
            configs[yml_file.stem] = yaml.safe_load(f)
    # 协调者
    orchestrator_path = AGENT_DIR / "orchestrator.yml"
    with open(orchestrator_path, "r", encoding="utf-8") as f:
        configs["coordinator"] = yaml.safe_load(f)
    return configs


# ═══════════════════════════════════════════════════════════
# 命令: serve — 启动 AgentPaaS 服务
# ═══════════════════════════════════════════════════════════

def cmd_serve(args):
    """启动 AgentPaaS 服务"""
    print("🚀 启动 AgentPaaS 服务...")

    proc = subprocess.Popen(
        [sys.executable, "-m", "agentpaas", "serve", "--port", str(args.port), "--dev"],
        cwd=str(PROJECT_ROOT),
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    )

    print(f"  PID: {proc.pid}")
    print(f"  URL: http://127.0.0.1:{args.port}")
    print("  等待服务就绪...")

    if not wait_for_server(30):
        print("  ✗ 服务启动超时")
        proc.kill()
        return None

    print("  ✓ PaaS 服务就绪")
    return proc


def cmd_create_tenant(args):
    """创建租户并返回 API key"""
    print("\n📋 创建租户...")
    result = subprocess.run(
        [sys.executable, "-m", "agentpaas", "create-tenant", "--name", "agent67v2-lab"],
        capture_output=True, text=True, cwd=str(PROJECT_ROOT),
    )
    for line in result.stdout.splitlines():
        print(f"  {line}")
        if line.startswith("API Key:"):
            return line.split(":", 1)[1].strip()
    return None


# ═══════════════════════════════════════════════════════════
# 命令: register — 注册全部代理到 PaaS
# ═══════════════════════════════════════════════════════════

def cmd_register(args):
    """注册协调者 + 5个微代理到 PaaS"""
    configs = load_agent_configs()
    agent_ids = {}

    # 1. 注册 5 个微代理
    print("\n📝 注册 5 个微代理...")
    for name, config in configs.items():
        if name == "coordinator":
            continue
        agent_name = config.get("name", name)
        print(f"  注册 {agent_name}...", end=" ")

        result = api_call("POST", "/agents", {
            "name": agent_name,
            "description": config.get("description", ""),
            "config": config,
            "tags": config.get("skill", {}).get("tags", []) + ["agent67v2", "micro-agent"],
            "environment": "production",
        })

        if "agent_id" in result:
            agent_ids[agent_name] = result["agent_id"]
            print(f"✅ {result['agent_id']}")
        else:
            print(f"❌ {result.get('error', 'Unknown')[:80]}")

    # 2. 注册协调者
    print(f"\n  注册协调者...", end=" ")
    orch_config = configs["coordinator"]
    result = api_call("POST", "/agents", {
        "name": "lambda-v2",
        "description": orch_config.get("description", ""),
        "config": orch_config,
        "tags": ["agent67v2", "coordinator", "multi-agent"],
        "environment": "production",
        "metadata": {"sub_agents": agent_ids},
    })

    if "agent_id" in result:
        agent_ids["coordinator"] = result["agent_id"]
        print(f"✅ {result['agent_id']}")
    else:
        print(f"❌ {result.get('error', 'Unknown')[:80]}")

    # 保存映射
    mapping_path = AGENT_DIR / ".paas_agent_ids.json"
    mapping_path.write_text(json.dumps(agent_ids, indent=2))
    print(f"\n  ✅ 共注册 {len(agent_ids)} 个代理")
    return agent_ids


# ═══════════════════════════════════════════════════════════
# 命令: chat — 通过 PaaS 与 lambda v2 对话
# ═══════════════════════════════════════════════════════════

def cmd_chat(args):
    """通过 AgentPaaS 与 lambda v2 对话（交互式）"""
    # 加载 agent IDs
    mapping_path = AGENT_DIR / ".paas_agent_ids.json"
    if not mapping_path.exists():
        print("✗ 未找到 agent IDs，请先运行 register")
        return

    agent_ids = json.loads(mapping_path.read_text())
    coordinator_id = agent_ids.get("coordinator")
    if not coordinator_id:
        print("✗ 未找到协调者 ID，请重新 register")
        return

    use_claude_code = args.claude_code or not os.environ.get("ANTHROPIC_API_KEY")

    print(f"\n{'═' * 60}")
    print(f"  🐑 lambda v2 — 多智能体协作版 (通过 AgentPaaS 运行)")
    print(f"  Coordinator ID: {coordinator_id}")
    print(f"  微代理: {', '.join(k for k in agent_ids if k != 'coordinator')}")
    if use_claude_code:
        print(f"  后端: Claude Code Max Plan (无需 API Key)")
    else:
        print(f"  后端: AgentPaaS 标准执行 (from_config)")
    print(f"{'═' * 60}")

    if use_claude_code:
        _chat_claude_code(coordinator_id, args)
    else:
        _chat_paas_api(coordinator_id, args)


def _chat_paas_api(coordinator_id: str, args):
    """纯 PaaS API 模式：所有执行都通过 PaaS"""
    print("\n  💡 所有请求通过 PaaS API 路由，执行记录自动保存")
    print("  输入 'exit' 退出 | 'runs' 查看执行历史\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
        if user_input.lower() == "runs":
            _show_runs(coordinator_id)
            continue

        print()
        result = api_call("POST", f"/agents/{coordinator_id}/run", {
            "input": user_input,
            "parameters": {},
            "context": {},
        })

        if "error" not in result:
            print(f"🐑 lambda v2: {result.get('output', '')}")
            usage = result.get("usage", {})
            print(f"  📊 [run_id={result['run_id']} | "
                  f"steps={usage.get('steps', 0)} | "
                  f"tokens={usage.get('total_tokens', 0)} | "
                  f"{usage.get('duration_ms', 0)}ms]")
        else:
            print(f"❌ 执行失败: {result['error'][:300]}")
        print()


def _chat_claude_code(coordinator_id: str, args):
    """
    Claude Code 混合模式 (与 agent67 一致):
    - LLM 调用 → claude CLI (Max Plan, 无需 API Key)
    - 工具执行 → 本地 Python (微代理)
    - 执行追踪 → 写入 PaaS (runs 表)

    兼顾:
    ✅ 不需要 API Key
    ✅ 每次对话记录都在 PaaS 中追踪
    ✅ 版本管理、执行历史由 PaaS 管理
    ✅ 多智能体协作
    """
    from agent67v2.core.coordinator import CoordinatorAssistant

    print("\n  💡 LLM 调用走 Claude Code CLI，执行记录同步到 PaaS")
    print("  输入 'exit' 退出 | 'runs' 查看执行历史 | 'trace' 追踪 | 'stats' 统计\n")

    # 检测后端
    try:
        from agent67.core.config import detect_backend
        model, use_api, backend = detect_backend()
    except ImportError:
        # 如果 agent67 不可用，直接检测 claude CLI
        if shutil.which("claude"):
            model, use_api, backend = "sonnet", False, "claude_code"
            print("✅ 使用 Claude Code Max Plan (无需 API Key)")
        else:
            model, use_api, backend = "sonnet", False, ""
            print("⚠️  未检测到 claude CLI，将使用标准 Lam")

    assistant = CoordinatorAssistant(
        model=model or "sonnet", use_api=use_api, backend=backend
    )

    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
        if user_input.lower() == "runs":
            _show_runs(coordinator_id)
            continue
        if user_input.lower() == "trace":
            assistant.print_trace()
            continue
        if user_input.lower() == "stats":
            assistant.print_stats()
            continue
        if user_input.lower() == "skills":
            from lambdagent.skills import SkillRegistry
            registry = SkillRegistry()
            print(f"\n📦 已注册技能 ({len(registry)} 个):")
            for name in registry.list_all():
                skill = registry.get(name)
                print(f"  • {name}: {skill.description}")
            print()
            continue

        print()
        t0 = time.time()

        # 通过 CoordinatorAssistant + 微代理执行
        response = assistant.chat(user_input)
        duration_ms = int((time.time() - t0) * 1000)
        steps = len(assistant.ctx.trace)

        # 显示结果（流式模式下已在 chat() 中显示）
        is_streaming = hasattr(assistant.brain, 'stream') and assistant.brain.stream
        if not is_streaming:
            print(f"🐑 lambda v2: {response}")

        # 同步执行记录到 PaaS
        _record_run_to_paas(coordinator_id, user_input, response, duration_ms,
                           steps, assistant._total_delegations)
        print()


def _record_run_to_paas(coordinator_id: str, input_text: str, output: str,
                        duration_ms: int, steps: int, delegations: int = 0):
    """
    将执行记录写入 PaaS（仅记录，不触发执行）。

    调用 /agents/{id}/runs/record 端点，
    与 /agents/{id}/run 不同，这个端点不会调 from_config/Lam，
    只写入 runs 表。
    """
    result = api_call("POST", f"/agents/{coordinator_id}/runs/record", {
        "input": input_text[:500],
        "output": output[:2000],
        "status": "completed",
        "duration_ms": duration_ms,
        "steps": steps,
        "source": "claude-code-cli",
        "metadata": {"delegations": delegations, "version": "v2-multi-agent"},
    })
    if "run_id" in result:
        run_id = result["run_id"]
        print(f"  📊 [PaaS: run_id={run_id} | steps={steps} | "
              f"delegations={delegations} | {duration_ms}ms]")


def _show_runs(coordinator_id: str):
    """显示最近的执行历史"""
    result = api_call("GET", f"/agents/{coordinator_id}/runs?limit=10")
    runs = result.get("runs", [])
    if not runs:
        print("  (暂无执行记录)")
        return
    print(f"\n📜 最近 {len(runs)} 次执行:")
    for r in runs:
        status_icon = "✅" if r.get("status") == "completed" else "❌"
        print(f"  {status_icon} {r['id']} | {r.get('status')} | "
              f"{r.get('duration_ms', 0)}ms | {r.get('created_at', '')[:19]}")
        if r.get("input"):
            print(f"     输入: {r['input'][:80]}")
    print()


# ═══════════════════════════════════════════════════════════
# 命令: all — 一键启动
# ═══════════════════════════════════════════════════════════

def cmd_all(args):
    """一键: 启动 PaaS + 创建租户 + 注册代理 + 开始对话"""
    global API_KEY

    proc = cmd_serve(args)
    if not proc:
        return

    try:
        # 创建租户
        key = cmd_create_tenant(args)
        if key:
            API_KEY = key
            os.environ["AGENTPAAS_API_KEY"] = key

        # 注册代理 (协调者 + 5微代理)
        agent_ids = cmd_register(args)
        if not agent_ids or "coordinator" not in agent_ids:
            return

        # 开始对话
        print(f"\n{'═' * 60}")
        print(f"  🐑 lambda v2 已就绪! 多智能体协作模式")
        print(f"  协调者 + {len(agent_ids) - 1} 个微代理")
        print(f"{'═' * 60}")
        cmd_chat(args)

    except KeyboardInterrupt:
        print("\n\n🛑 中断")
    finally:
        print("\n关闭 PaaS 服务...")
        proc.terminate()
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()
        print("  ✓ 服务已关闭")


# ═══════════════════════════════════════════════════════════
# 命令: status — 查看代理状态
# ═══════════════════════════════════════════════════════════

def cmd_status(args):
    """查看所有代理在 PaaS 中的状态"""
    mapping_path = AGENT_DIR / ".paas_agent_ids.json"
    if not mapping_path.exists():
        print("✗ 未注册，请先运行: python launch_paas.py register")
        return

    agent_ids = json.loads(mapping_path.read_text())
    print(f"\n📋 lambda v2 🐑 状态查询 ({len(agent_ids)} 个代理)")

    for name, aid in agent_ids.items():
        role = "🎯 协调者" if name == "coordinator" else "🔧 微代理"
        print(f"\n  {role}: {name}")
        print(f"    Agent ID: {aid}")

        result = api_call("GET", f"/agents/{aid}")
        if "error" not in result:
            print(f"    版本: v{result.get('current_version', '?')}")
            print(f"    状态: {result.get('status', 'unknown')}")

    # 协调者执行历史
    coordinator_id = agent_ids.get("coordinator")
    if coordinator_id:
        _show_runs(coordinator_id)


# ═══════════════════════════════════════════════════════════
# CLI 入口
# ═══════════════════════════════════════════════════════════

def main():
    parser = argparse.ArgumentParser(
        description="🐑 lambda v2 — 多智能体协作版 (通过 AgentPaaS 运行)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  python launch_paas.py all                  # 一键启动 (推荐, 无需 API Key)
  python launch_paas.py serve                # 只启动 PaaS 服务
  python launch_paas.py register             # 注册协调者 + 5微代理
  python launch_paas.py chat --claude-code   # Claude Code 模式对话
  python launch_paas.py status               # 查看状态
        """
    )
    sub = parser.add_subparsers(dest="command")

    # serve
    p_serve = sub.add_parser("serve", help="启动 AgentPaaS 服务")
    p_serve.add_argument("--port", type=int, default=8000)

    # register
    sub.add_parser("register", help="注册全部代理到 PaaS")

    # chat
    p_chat = sub.add_parser("chat", help="通过 PaaS 与 lambda v2 对话")
    p_chat.add_argument("--claude-code", action="store_true",
                        help="使用 Claude Code CLI 作为 LLM 后端")

    # all
    p_all = sub.add_parser("all", help="一键启动全流程")
    p_all.add_argument("--port", type=int, default=8000)
    p_all.add_argument("--claude-code", action="store_true", default=True,
                       help="使用 Claude Code CLI (默认)")

    # status
    sub.add_parser("status", help="查看代理状态")

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        return

    # 确保 claude-code 属性存在
    if not hasattr(args, "claude_code"):
        args.claude_code = bool(shutil.which("claude"))

    commands = {
        "serve": lambda: cmd_serve(args),
        "register": lambda: cmd_register(args),
        "chat": lambda: cmd_chat(args),
        "all": lambda: cmd_all(args),
        "status": lambda: cmd_status(args),
    }

    commands[args.command]()


if __name__ == "__main__":
    main()