lambdagent/agentexample/pptagent67/launch_paas.py

#!/usr/bin/env python3
"""
pptagent67/launch_paas.py — 通过 AgentPaaS API 或 Claude Code CLI 启动 PPT 生成
================================================================================

两种运行模式：
  1. AgentPaaS API 模式: 启动 PaaS 服务，注册所有 agent，通过 REST API 执行
  2. Claude Code CLI 桥接模式: 利用 Claude Max Plan 的 claude CLI 作为 LLM 后端

用法:
  # 模式 1: PaaS 全流程
  python launch_paas.py all --port 8000

  # 模式 2: 使用 Claude Code CLI (Max Plan)
  python launch_paas.py claude-code
  python launch_paas.py claude-code --topic "AI 在教育中的应用" --template business
  python launch_paas.py claude-code --workspace ./workspace/claude_run_xxx  # 断点续跑
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
import urllib.request
import urllib.error
from pathlib import Path
from typing import Dict, Optional

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

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

AGENT_CONFIGS = [
    ("ppt-67-orchestrator",         "PPT 主编排器",   AGENT_DIR / "orchestrator.yml"),
    ("ppt-67-knowledge-extractor",  "知识提取师",     AGENT_DIR / "agents" / "knowledge-extractor.yml"),
    ("ppt-67-outline-generator",    "大纲生成师",     AGENT_DIR / "agents" / "outline-generator.yml"),
    ("ppt-67-content-refiner",      "内容精修师",     AGENT_DIR / "agents" / "content-refiner.yml"),
    ("ppt-67-renderer",             "PPT 渲染师",     AGENT_DIR / "agents" / "ppt-renderer.yml"),
]

# 迭代控制参数
MAX_ROUNDS = 3
QUALITY_THRESHOLD = 0.70
MAX_TURNS_PER_PHASE = 25
PHASE_TIMEOUT = 1800
API_RETRY_MAX = 3
API_RETRY_BACKOFF = 30
PHASE_COOLDOWN = 10


# ═══════════════════════════════════════════════════════════
# 1. PaaS API Helpers
# ═══════════════════════════════════════════════════════════

def api_call(method: str, path: str, data: dict = None) -> dict:
    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=600) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        body = e.read().decode()
        print(f"  API Error {e.code}: {body[:500]}")
        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):
    for i 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


# ═══════════════════════════════════════════════════════════
# 2. 命令: serve / register / run / all
# ═══════════════════════════════════════════════════════════

def cmd_serve(args):
    print("Starting AgentPaaS server...")
    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}, URL: http://127.0.0.1:{args.port}")
    if not wait_for_server(30):
        print("  ERROR: Server failed to start")
        proc.kill()
        return None
    print("  ✓ Server ready")
    return proc


def cmd_register(args):
    import yaml
    print(f"\nRegistering {len(AGENT_CONFIGS)} agents...")
    agent_ids = {}
    for agent_id, name, yml_path in AGENT_CONFIGS:
        with open(yml_path, "r", encoding="utf-8") as f:
            config = yaml.safe_load(f)
        result = api_call("POST", "/agents", {
            "name": name,
            "description": config.get("description", ""),
            "config": config,
            "tags": ["pptagent67", config.get("type", "react")],
        })
        if "agent_id" in result:
            agent_ids[agent_id] = result["agent_id"]
            print(f"  ✓ {name:20s} → {result['agent_id']}")
        else:
            print(f"  ✗ {name:20s} → {result.get('error', 'Unknown')}")

    mapping_path = AGENT_DIR / ".paas_agent_ids.json"
    mapping_path.write_text(json.dumps(agent_ids, indent=2), encoding="utf-8")
    return agent_ids


def cmd_run(args):
    mapping_path = AGENT_DIR / ".paas_agent_ids.json"
    if not mapping_path.exists():
        print("ERROR: Run 'register' first.")
        return
    agent_ids = json.loads(mapping_path.read_text())
    orchestrator_id = agent_ids.get("ppt-67-orchestrator")
    if not orchestrator_id:
        print("ERROR: Orchestrator not registered.")
        return

    topic = args.topic if hasattr(args, "topic") and args.topic else "[默认主题]"
    result = api_call("POST", f"/agents/{orchestrator_id}/run", {
        "input": topic,
        "parameters": {"template": getattr(args, "template", "academic")},
        "context": {"sub_agent_ids": agent_ids},
    })
    if "error" not in result:
        print(f"  ✓ Run completed: {result.get('run_id')}")
    else:
        print(f"  ✗ Failed: {result['error'][:500]}")


def cmd_all(args):
    global API_KEY
    proc = cmd_serve(args)
    if not proc:
        return
    try:
        key_result = subprocess.run(
            [sys.executable, "-m", "agentpaas", "create-tenant", "--name", "ppt-lab"],
            capture_output=True, text=True, cwd=str(PROJECT_ROOT),
        )
        for line in key_result.stdout.splitlines():
            if line.startswith("API Key:"):
                API_KEY = line.split(":", 1)[1].strip()
                os.environ["AGENTPAAS_API_KEY"] = API_KEY
        cmd_register(args)
        cmd_run(args)
    finally:
        proc.terminate()
        proc.wait(timeout=5)


# ═══════════════════════════════════════════════════════════
# 3. 命令: claude-code — 使用 Claude Code CLI
# ═══════════════════════════════════════════════════════════

def run_phase(phase_dir: Path, agent_name: str, desc: str,
              prompt_content: str, round_num: int) -> str:
    """执行单个阶段，带自动重试"""
    phase_dir.mkdir(parents=True, exist_ok=True)
    (phase_dir / "artifacts").mkdir(exist_ok=True)
    (phase_dir / "prompt.md").write_text(prompt_content, encoding="utf-8")

    print(f"\n{'━'*60}")
    print(f"  Phase: {phase_dir.name} — {desc}")
    print(f"  Agent: {agent_name} | Round: {round_num}")
    print(f"{'━'*60}")

    cmd = [
        "claude", "-p",
        "--model", "claude-opus-4-6",
        "--max-turns", str(MAX_TURNS_PER_PHASE),
        "--allowedTools", "Edit,Write,Read,Bash,Glob,Grep,WebSearch,WebFetch",
    ]

    for attempt in range(1, API_RETRY_MAX + 1):
        t0 = time.time()
        try:
            result = subprocess.run(
                cmd, input=prompt_content,
                capture_output=True, text=True,
                timeout=PHASE_TIMEOUT, cwd=str(phase_dir),
            )
            elapsed = time.time() - t0
            output = result.stdout or ""
            stderr = result.stderr or ""

            is_api_error = any(
                p in output or p in stderr
                for p in ["Request timed out", "overloaded_error", "529", "rate_limit"]
            )
            if is_api_error and attempt < API_RETRY_MAX:
                wait = API_RETRY_BACKOFF * (2 ** (attempt - 1))
                print(f"  ⚠ API error after {elapsed:.0f}s. Retrying in {wait}s...")
                time.sleep(wait)
                continue

            if result.returncode != 0 and stderr:
                output = f"[STDERR] {stderr[:500]}\n{output}"

            (phase_dir / "claude_output.md").write_text(output, encoding="utf-8")
            print(f"  ✓ Done in {elapsed:.0f}s | {len(output)} chars")
            return output

        except subprocess.TimeoutExpired:
            msg = f"[TIMEOUT] {phase_dir.name} timed out after {PHASE_TIMEOUT}s"
            print(f"  ✗ {msg}")
            (phase_dir / "claude_output.md").write_text(msg, encoding="utf-8")
            return msg
        except FileNotFoundError:
            print("  ✗ 'claude' CLI not found.")
            return "[ERROR] claude CLI not found"
        except Exception as e:
            if attempt < API_RETRY_MAX:
                time.sleep(API_RETRY_BACKOFF * attempt)
            else:
                msg = f"[ERROR] {e}"
                (phase_dir / "claude_output.md").write_text(msg, encoding="utf-8")
                return msg

    return "[FAILED] All retries exhausted"


def build_prompt(workspace: Path, phase_dir: Path, desc: str,
                 system_prompt: str, topic: str, round_num: int,
                 template: str, prev_outputs: list,
                 revision_context: str = "") -> str:
    input_parts = [
        f"workspace: {workspace}",
        f"phase_dir: {phase_dir}",
        f"round: {round_num}",
        f"template: {template}",
        f"\n## PPT 主题:\n{topic[:8000]}",
    ]
    if revision_context:
        input_parts.append(f"\n## 修改建议 (Round {round_num - 1}):\n{revision_context[:4000]}")
    for prev in prev_outputs[-2:]:
        input_parts.append(f"\n## Previous phase output:\n{prev[:3000]}")

    return f"""# {desc}

## System Instructions
{system_prompt[:8000]}

## Input Context
{chr(10).join(input_parts)}

## Important
- 将所有产出文件写入 {phase_dir}/
- 先写 work_plan.md，再执行，最后写 report.json 和 report.md
- report.json 必须包含 _meta 字段（含 phase, round, status）
- 这是 Round {round_num}{'，请根据修改建议重点改进' if round_num > 1 else ''}
"""


def cmd_claude_code(args):
    """通过 Claude Code CLI 执行 PPT 生成流程"""
    import yaml
    from datetime import datetime

    global MAX_ROUNDS, MAX_TURNS_PER_PHASE, PHASE_TIMEOUT, PHASE_COOLDOWN
    if hasattr(args, "max_rounds") and args.max_rounds:
        MAX_ROUNDS = args.max_rounds
    if hasattr(args, "max_turns") and args.max_turns:
        MAX_TURNS_PER_PHASE = args.max_turns

    topic = args.topic if hasattr(args, "topic") and args.topic else ""
    template = args.template if hasattr(args, "template") and args.template else "academic"

    if not topic:
        topic_path = AGENT_DIR / "TOPIC.md"
        if topic_path.exists():
            topic = topic_path.read_text(encoding="utf-8")
        else:
            print("ERROR: No topic. Use --topic or create TOPIC.md")
            return

    print(f"""
╔══════════════════════════════════════════════════════════════╗
║  PPTAgent-67 via Claude Code CLI                             ║
║  Template: {template:<10s} | Threshold: {QUALITY_THRESHOLD:.0%} | Max rounds: {MAX_ROUNDS}  ║
╚══════════════════════════════════════════════════════════════╝
""")

    # 工作区
    if hasattr(args, "workspace") and args.workspace:
        workspace = Path(args.workspace)
    else:
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        workspace = AGENT_DIR / "workspace" / f"claude_run_{ts}"
        workspace.mkdir(parents=True, exist_ok=True)
    print(f"  Workspace: {workspace}")
    print(f"  Topic: {topic[:80]}...")

    MAIN_PHASES = [
        ("01_knowledge_extraction", "knowledge-extractor", "从知识库提取信息"),
        ("02_outline_generation",   "outline-generator",   "生成 PPT 大纲"),
        ("03_content_refinement",   "content-refiner",     "精修内容 & 质量评估"),
        ("04_ppt_rendering",        "ppt-renderer",        "渲染 .pptx 文件"),
    ]

    for round_num in range(1, MAX_ROUNDS + 1):
        print(f"\n{'═'*60}")
        print(f"  ROUND {round_num}/{MAX_ROUNDS}")
        print(f"{'═'*60}")

        prev_outputs = []
        revision_context = ""

        for phase_idx, (phase_name, agent_name, desc) in enumerate(MAIN_PHASES):
            phase_dir = workspace / f"round_{round_num}" / phase_name

            # 跳过已完成
            if (phase_dir / "report.json").exists():
                print(f"\n  ⏭ Skipping {phase_name} (exists)")
                existing = (phase_dir / "claude_output.md").read_text(encoding="utf-8") \
                    if (phase_dir / "claude_output.md").exists() else ""
                prev_outputs.append(existing[:5000])
                continue

            agent_yml = AGENT_DIR / "agents" / f"{agent_name}.yml"
            with open(agent_yml, "r", encoding="utf-8") as f:
                config = yaml.safe_load(f)

            prompt = build_prompt(
                workspace, phase_dir, desc,
                config.get("systemPrompt", ""), topic,
                round_num, template, prev_outputs,
                revision_context=revision_context,
            )
            output = run_phase(phase_dir, agent_name, desc, prompt, round_num)
            prev_outputs.append(output[:5000])

            # 精修阶段后检查质量
            if phase_name == "03_content_refinement":
                report_path = phase_dir / "report.json"
                quality = 0.0
                if report_path.exists():
                    try:
                        data = json.loads(report_path.read_text(encoding="utf-8"))
                        quality = data.get("quality_score", 0)
                    except Exception:
                        pass

                print(f"\n  Quality: {quality:.2f} (threshold: {QUALITY_THRESHOLD})")
                if quality < QUALITY_THRESHOLD and round_num < MAX_ROUNDS:
                    print(f"  ⚠ Below threshold. Will iterate.")
                    revision_context = output[:4000]
                    break  # 跳过渲染，进入下一轮

            if phase_idx < len(MAIN_PHASES) - 1:
                time.sleep(PHASE_COOLDOWN)

        # 检查是否渲染完成
        pptx_dir = workspace / f"round_{round_num}" / "04_ppt_rendering"
        if (pptx_dir / "report.json").exists():
            print(f"\n  ✅ PPT rendered! Round {round_num} complete.")
            break

    print(f"\n{'═'*60}")
    print(f"  Pipeline complete. Workspace: {workspace}")
    print(f"{'═'*60}\n")


# ═══════════════════════════════════════════════════════════
# CLI
# ═══════════════════════════════════════════════════════════

def main():
    parser = argparse.ArgumentParser(description="PPTAgent-67 PaaS Launcher")
    sub = parser.add_subparsers(dest="command")

    serve_p = sub.add_parser("serve")
    serve_p.add_argument("--port", type=int, default=8000)

    sub.add_parser("register")

    run_p = sub.add_parser("run")
    run_p.add_argument("--topic", type=str, default="")
    run_p.add_argument("--template", type=str, default="academic")

    all_p = sub.add_parser("all")
    all_p.add_argument("--port", type=int, default=8000)
    all_p.add_argument("--topic", type=str, default="")
    all_p.add_argument("--template", type=str, default="academic")

    cc_p = sub.add_parser("claude-code")
    cc_p.add_argument("--topic", type=str, default="")
    cc_p.add_argument("--template", type=str, default="academic",
                       choices=["academic", "business", "minimal", "tech"])
    cc_p.add_argument("--workspace", type=str, default=None)
    cc_p.add_argument("--max-rounds", type=int, default=None)
    cc_p.add_argument("--max-turns", type=int, default=None)

    args = parser.parse_args()
    commands = {
        "serve": cmd_serve,
        "register": cmd_register,
        "run": cmd_run,
        "all": cmd_all,
        "claude-code": cmd_claude_code,
    }

    if args.command in commands:
        commands[args.command](args)
    else:
        parser.print_help()
        print("\nExamples:")
        print('  python launch_paas.py claude-code --topic "AI 在医疗中的应用" --template business')
        print("  python launch_paas.py all --port 8000")


if __name__ == "__main__":
    main()