#!/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()