lambdagent/agentexample/agent67/launch_paas.py

#!/usr/bin/env python3
"""
agent67/launch_paas.py — 通过 AgentPaaS 运行 lambda 个人助理 🐂
================================================================

两种运行模式:
  1. PaaS API 模式: 启动 PaaS 服务，注册 agent，通过 REST API 对话
  2. Claude Code 模式: 注册到 PaaS（版本/追踪），LLM 调用走 claude CLI

用法:
  # 一键启动（推荐）
  python launch_paas.py all

  # 分步执行
  python launch_paas.py serve               # 启动 PaaS 服务
  python launch_paas.py register            # 注册 lambda agent
  python launch_paas.py chat                # 通过 PaaS API 与 lambda 对话
  python launch_paas.py chat --claude-code  # PaaS 追踪 + Claude Code 执行
"""
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  # Paper08/
AGENT_DIR = Path(__file__).resolve().parent                    # agent67/
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
# ═══════════════════════════════════════════════════════════

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_config() -> dict:
    """加载 agent67 的 YAML 配置"""
    import yaml
    config_path = AGENT_DIR / "agent-config.yml"
    with open(config_path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


# ═══════════════════════════════════════════════════════════
# 命令: 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", "agent67-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 — 注册 lambda agent 到 PaaS
# ═══════════════════════════════════════════════════════════

def cmd_register(args):
    """注册 lambda agent 到 AgentPaaS"""
    config = load_agent_config()

    print(f"\n📝 注册 lambda 🐂 到 AgentPaaS...")
    result = api_call("POST", "/agents", {
        "name": config.get("name", "Lambda 个人助理"),
        "description": config.get("description", "能操控电脑的全能型 AI 助手"),
        "config": config,
        "tags": ["agent67", "personal-assistant", "computer-use"],
        "environment": "production",
    })

    if "agent_id" in result:
        agent_id = result["agent_id"]
        version = result["version"]
        print(f"  ✓ 注册成功!")
        print(f"    Agent ID: {agent_id}")
        print(f"    Version:  v{version}")
        print(f"    Endpoint: POST /api/v1/agents/{agent_id}/run")

        # 保存 agent ID
        mapping = {"agent_id": agent_id, "version": version}
        mapping_path = AGENT_DIR / ".paas_agent_id.json"
        mapping_path.write_text(json.dumps(mapping, indent=2))
        return agent_id
    else:
        print(f"  ✗ 注册失败: {result.get('error', 'Unknown')}")
        return None


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

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

    mapping = json.loads(mapping_path.read_text())
    agent_id = mapping["agent_id"]

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

    print(f"\n{'═' * 60}")
    print(f"  🐂 lambda — 通过 AgentPaaS 运行")
    print(f"  Agent ID: {agent_id}")
    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(agent_id, args)
    else:
        _chat_paas_api(agent_id, args)


def _chat_paas_api(agent_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 下线了！")
            break

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

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

        if "error" not in result:
            print(f"🐂 lambda: {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(agent_id: str, args):
    """
    Claude Code 混合模式:
    - LLM 调用 → claude CLI (Max Plan, 无需 API Key)
    - 工具执行 → 本地 Python
    - 执行追踪 → 写入 PaaS (runs 表)

    这样兼顾了:
    ✅ 不需要 API Key
    ✅ 每次对话记录都在 PaaS 中追踪
    ✅ 版本管理、执行历史 由 PaaS 管理
    """
    from agent67.core.assistant import PersonalAssistant

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

    assistant = PersonalAssistant(model="sonnet", use_api=False)

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

        if not user_input:
            continue
        if user_input.lower() in ("exit", "quit", "bye"):
            print("👋 lambda 下线了！")
            break
        if user_input.lower() == "runs":
            _show_runs(agent_id)
            continue
        if user_input.lower() == "trace":
            assistant.print_trace()
            continue

        print()
        t0 = time.time()

        # 通过 ClaudeLam + 本地工具执行
        response = assistant.chat(user_input)
        duration_ms = int((time.time() - t0) * 1000)
        steps = len(assistant.ctx.trace)

        print(f"🐂 lambda: {response}")

        # 同步执行记录到 PaaS
        _record_run_to_paas(agent_id, user_input, response, duration_ms, steps)
        print()


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

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


def _show_runs(agent_id: str):
    """显示最近的执行历史"""
    result = api_call("GET", f"/agents/{agent_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 + 创建租户 + 注册 agent + 开始对话"""
    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

        # 注册 agent
        agent_id = cmd_register(args)
        if not agent_id:
            return

        # 开始对话
        print(f"\n{'═' * 60}")
        print(f"  🐂 lambda 已就绪! 开始对话吧")
        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 — 查看 agent 状态
# ═══════════════════════════════════════════════════════════

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

    mapping = json.loads(mapping_path.read_text())
    agent_id = mapping["agent_id"]

    print(f"\n📋 lambda 🐂 状态查询")
    print(f"  Agent ID: {agent_id}")

    # 查询 agent 信息
    result = api_call("GET", f"/agents/{agent_id}")
    if "error" not in result:
        print(f"  名称: {result.get('name')}")
        print(f"  版本: v{result.get('current_version')}")
        print(f"  状态: {result.get('status')}")
        print(f"  创建: {result.get('created_at', '')[:19]}")

    # 查询版本历史
    versions = api_call("GET", f"/agents/{agent_id}/versions")
    vlist = versions.get("versions", [])
    if vlist:
        print(f"\n  📦 版本历史 ({len(vlist)} 个):")
        for v in vlist[:5]:
            current = " ← current" if v.get("is_current") else ""
            print(f"    v{v['version']}{current} | {v.get('created_at', '')[:19]}")

    # 查询最近执行
    _show_runs(agent_id)


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

def main():
    parser = argparse.ArgumentParser(
        description="🐂 lambda — 通过 AgentPaaS 运行个人助理",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  python launch_paas.py all                  # 一键启动 (推荐)
  python launch_paas.py serve                # 只启动 PaaS 服务
  python launch_paas.py register             # 注册 agent
  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="注册 lambda agent 到 PaaS")

    # chat
    p_chat = sub.add_parser("chat", help="通过 PaaS 与 lambda 对话")
    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="查看 agent 状态")

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