#!/usr/bin/env python3 """ research67/launch_paas.py — 通过 AgentPaaS API 启动科研智能体 ================================================================ 两种运行模式： 1. AgentPaaS API 模式: 启动 PaaS 服务，注册所有 agent，通过 REST API 执行 2. Claude Code CLI 桥接模式: 利用 Claude Max Plan 的 claude CLI 作为 LLM 后端 用法: # 模式 1: 启动 PaaS 服务 + 注册 + 执行 python launch_paas.py serve # 启动服务 (端口 8000) python launch_paas.py register # 注册所有 agent 到 PaaS python launch_paas.py run # 通过 API 执行科研流程 python launch_paas.py all # serve + register + run (一键启动) # 模式 2: 使用 Claude Code CLI (Max Plan) python launch_paas.py claude-code # 通过 claude CLI 执行 """ from __future__ import annotations import argparse import hashlib import json import os import secrets import subprocess import sys import time import urllib.request import urllib.error from pathlib import Path from typing import Dict, List, 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 = [ ("research-67-orchestrator", "科研主编排器", AGENT_DIR / "orchestrator.yml"), ("research-67-idea-analyst", "创意分析师", AGENT_DIR / "agents" / "idea-analyst.yml"), ("research-67-lit-searcher", "文献检索员", AGENT_DIR / "agents" / "lit-searcher.yml"), ("research-67-exp-planner", "实验规划师", AGENT_DIR / "agents" / "exp-planner.yml"), ("research-67-exp-executor", "实验执行员", AGENT_DIR / "agents" / "exp-executor.yml"), ("research-67-result-analyzer", "结果分析师", AGENT_DIR / "agents" / "result-analyzer.yml"), ("research-67-paper-writer", "论文撰写员", AGENT_DIR / "agents" / "paper-writer.yml"), ("research-67-data-verifier", "数据核验员", AGENT_DIR / "agents" / "data-verifier.yml"), ("research-67-paper-reviewer", "论文评审员", AGENT_DIR / "agents" / "paper-reviewer.yml"), ("research-67-review-feedback", "评审反馈合成器", AGENT_DIR / "agents" / "review-feedback.yml"), ] # 迭代控制参数 MAX_ROUNDS = 5 # 最大迭代轮数 ACCEPTANCE_THRESHOLD = 0.50 # 目标接收概率 (50%) MAX_TURNS_PER_PHASE = 30 # 每个 agent 最大工具调用轮数 PHASE_TIMEOUT = 3600 # 每个 phase 超时 (60分钟，Round1实测单阶段可达20min+) API_RETRY_MAX = 3 # API 失败最大重试次数 API_RETRY_BACKOFF = 30 # 重试间隔基数 (秒) PHASE_COOLDOWN = 10 # 阶段间冷却 (秒，防 429/529) # ═══════════════════════════════════════════════════════════ # 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): """等待 PaaS 服务就绪""" 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 — 启动 AgentPaaS 服务 # ═══════════════════════════════════════════════════════════ def cmd_serve(args): """启动 AgentPaaS 并自动创建租户""" 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}") print(f" URL: http://127.0.0.1:{args.port}") print(" Waiting for server to be ready...") if not wait_for_server(30): print(" ERROR: Server failed to start within 30s") proc.kill() return None print(" ✓ Server ready") return proc def cmd_create_tenant(args): """创建租户并返回 API key""" print("Creating tenant...") result = subprocess.run( [sys.executable, "-m", "agentpaas", "create-tenant", "--name", "research-lab"], capture_output=True, text=True, cwd=str(PROJECT_ROOT), ) print(result.stdout) # Extract API key from output for line in result.stdout.splitlines(): if line.startswith("API Key:"): key = line.split(":", 1)[1].strip() return key return None # ═══════════════════════════════════════════════════════════ # 3. 命令: register — 注册所有 agent # ═══════════════════════════════════════════════════════════ def cmd_register(args): """注册所有 research67 agent 到 AgentPaaS""" import yaml print(f"\nRegistering {len(AGENT_CONFIGS)} agents to AgentPaaS...") 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": ["research67", config.get("type", "react")], "environment": "production", }) if "agent_id" in result: agent_ids[agent_id] = result["agent_id"] print(f" ✓ {name:20s} → {result['agent_id']} (v{result['version']})") else: print(f" ✗ {name:20s} → {result.get('error', 'Unknown error')}") # 保存 agent ID 映射 mapping_path = AGENT_DIR / ".paas_agent_ids.json" mapping_path.write_text(json.dumps(agent_ids, indent=2), encoding="utf-8") print(f"\n Agent IDs saved to {mapping_path}") return agent_ids # ═══════════════════════════════════════════════════════════ # 4. 命令: run — 通过 API 执行科研流程 # ═══════════════════════════════════════════════════════════ def cmd_run(args): """通过 AgentPaaS API 执行 orchestrator""" mapping_path = AGENT_DIR / ".paas_agent_ids.json" if not mapping_path.exists(): print("ERROR: No agent ID mapping found. Run 'register' first.") return agent_ids = json.loads(mapping_path.read_text()) orchestrator_id = agent_ids.get("research-67-orchestrator") if not orchestrator_id: print("ERROR: Orchestrator not registered.") return # 读取 IDEA idea_path = AGENT_DIR / "paper" / "paper01" / "IDEA.md" idea = idea_path.read_text(encoding="utf-8") if idea_path.exists() else "[IDEA.md not found]" print(f"\n{'═'*60}") print(f" Executing research67 orchestrator via AgentPaaS API") print(f" Agent ID: {orchestrator_id}") print(f" IDEA: {idea[:80]}...") print(f"{'═'*60}\n") result = api_call("POST", f"/agents/{orchestrator_id}/run", { "input": idea, "parameters": {}, "context": {"sub_agent_ids": agent_ids}, }) if "error" not in result: print(f"\n ✓ Run completed") print(f" Run ID: {result.get('run_id')}") print(f" Status: {result.get('status')}") print(f" Tokens: {result.get('usage', {}).get('total_tokens', 0)}") print(f" Duration: {result.get('usage', {}).get('duration_ms', 0)}ms") print(f"\n Output (first 500 chars):") print(f" {result.get('output', '')[:500]}") else: print(f"\n ✗ Execution failed: {result['error'][:500]}") # ═══════════════════════════════════════════════════════════ # 5. 命令: all — 一键启动全流程 # ═══════════════════════════════════════════════════════════ def cmd_all(args): """一键: 启动服务 + 创建租户 + 注册 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 print(f" API Key: {key}") # 注册 agents cmd_register(args) # 执行 cmd_run(args) finally: print("\nShutting down server...") proc.terminate() proc.wait(timeout=5) # ═══════════════════════════════════════════════════════════ # 6. 命令: claude-code — 使用 Claude Code CLI 作为 LLM 后端 # ═══════════════════════════════════════════════════════════ def run_phase(phase_dir: Path, agent_name: str, desc: str, prompt_content: str, round_num: int) -> str: """ 执行单个阶段，带自动重试和详细日志。 重试策略: - subprocess.TimeoutExpired → 不重试（已用完时间预算） - API 529/overloaded → 指数退避重试 - "Request timed out" → 指数退避重试（通常是 API 网关超时） - 其他错误 → 重试 1 次 """ phase_dir.mkdir(parents=True, exist_ok=True) (phase_dir / "artifacts").mkdir(exist_ok=True) # 写入 prompt (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" Max turns: {MAX_TURNS_PER_PHASE} | Timeout: {PHASE_TIMEOUT}s | Retries: {API_RETRY_MAX}") 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", ] last_error = "" for attempt in range(1, API_RETRY_MAX + 1): t0 = time.time() attempt_label = f"[attempt {attempt}/{API_RETRY_MAX}]" try: print(f" 🚀 {attempt_label} Running claude CLI ...") 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 "" # --- 检测 API 层面错误（进程成功退出但内容是错误信息）--- is_api_error = False for err_pattern in ["Request timed out", "overloaded_error", "529", "rate_limit", "500 Internal", "502 Bad Gateway", "API Error"]: if err_pattern in output or err_pattern in stderr: is_api_error = True last_error = (output + stderr)[:300] break if is_api_error and attempt < API_RETRY_MAX: wait = API_RETRY_BACKOFF * (2 ** (attempt - 1)) # 30s, 60s, 120s print(f" ⚠ {attempt_label} API error after {elapsed:.0f}s: {last_error[:100]}") print(f" ⏳ Retrying in {wait}s ...") time.sleep(wait) continue # --- 成功（或最后一次尝试，即使有 API 错误也保存）--- if result.returncode != 0 and stderr: output = f"[STDERR] {stderr[:500]}\n{output}" (phase_dir / "claude_output.md").write_text(output, encoding="utf-8") # 产出检查 report_path = phase_dir / "report.json" wp_path = phase_dir / "work_plan.md" artifacts = list((phase_dir / "artifacts").rglob("*")) artifacts_count = sum(1 for a in artifacts if a.is_file()) status_parts = [] if wp_path.exists(): status_parts.append("✅ work_plan") if report_path.exists(): status_parts.append("✅ report.json") else: status_parts.append("⚠ no report.json") status_parts.append(f"📎 {artifacts_count} artifacts") print(f" ✓ {attempt_label} Done in {elapsed:.0f}s | {len(output)} chars | {' | '.join(status_parts)}") return output except subprocess.TimeoutExpired: elapsed = time.time() - t0 msg = f"[TIMEOUT] Phase {phase_dir.name} timed out after {elapsed:.0f}s (limit {PHASE_TIMEOUT}s)" print(f" ✗ {msg}") # 超时不重试（已用完整个时间窗口），但保存已有产出 partial = "" if (phase_dir / "work_plan.md").exists(): partial += " (work_plan exists, partial progress saved)" (phase_dir / "claude_output.md").write_text(msg + partial, encoding="utf-8") return msg except FileNotFoundError: print(" ✗ 'claude' CLI not found. Is Claude Code installed?") return "[ERROR] claude CLI not found" except Exception as e: last_error = str(e) if attempt < API_RETRY_MAX: wait = API_RETRY_BACKOFF * attempt print(f" ⚠ {attempt_label} Error: {last_error[:100]}. Retrying in {wait}s ...") time.sleep(wait) else: msg = f"[ERROR] {last_error}" print(f" ✗ {msg}") (phase_dir / "claude_output.md").write_text(msg, encoding="utf-8") return msg # 所有重试都失败 msg = f"[FAILED] All {API_RETRY_MAX} attempts failed. Last: {last_error[:200]}" (phase_dir / "claude_output.md").write_text(msg, encoding="utf-8") return msg def build_prompt(workspace: Path, phase_dir: Path, desc: str, system_prompt: str, idea: str, round_num: int, prev_outputs: list, revision_context: str = "") -> str: """构建阶段 prompt""" input_parts = [ f"workspace: {workspace}", f"phase_dir: {phase_dir}", f"round: {round_num}", f"\n## IDEA:\n{idea[:8000]}", ] if revision_context: input_parts.append(f"\n## 评审反馈与修改指令 (来自 Round {round_num - 1}):\n{revision_context[:6000]}") for prev in prev_outputs[-2:]: input_parts.append(f"\n## Previous phase output:\n{prev[:3000]}") full_input = "\n".join(input_parts) return f"""# {desc} ## System Instructions {system_prompt[:8000]} ## Input Context {full_input} ## 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 read_review_score(workspace: Path, round_num: int) -> tuple: """读取评审报告的 acceptance_probability 和 overall_score 优先从 artifacts/review_report.json 读取（详细评审）， 其次从 report.json 读取（元数据摘要）。 """ review_dir = workspace / f"round_{round_num}" / "08_paper_review" # 优先: artifacts/review_report.json (完整评审) for candidate in [ review_dir / "artifacts" / "review_report.json", review_dir / "report.json", ]: if candidate.exists(): try: data = json.loads(candidate.read_text(encoding="utf-8")) prob = data.get("estimated_acceptance_probability", data.get("acceptance_probability", 0)) score = data.get("overall_score", 0) if score > 0 or prob > 0: return prob, score except Exception: continue return 0.0, 0 def read_revision_context(workspace: Path, round_num: int) -> str: """读取上一轮的评审 + 反馈合成结果，作为下一轮的 revision_context""" parts = [] # 评审报告摘要 (优先从 artifacts/review_report.json) review_dir = workspace / f"round_{round_num}" / "08_paper_review" review_report = review_dir / "artifacts" / "review_report.json" if not review_report.exists(): review_report = review_dir / "report.json" if review_report.exists(): try: data = json.loads(review_report.read_text(encoding="utf-8")) parts.append(f"### 上轮评审 (Round {round_num})") parts.append(f"- 总评分: {data.get('overall_score', '?')}/10") parts.append(f"- 接收概率: {data.get('estimated_acceptance_probability', data.get('acceptance_probability', '?'))}") parts.append(f"- 建议: {data.get('recommendation', '?')}") if "weaknesses" in data: parts.append("\n#### 弱点:") for w in data["weaknesses"][:6]: w_text = w if isinstance(w, str) else w.get("title", str(w)[:100]) parts.append(f" - {w_text[:150]}") if "required_revisions" in data: parts.append("\n#### 必须修订:") for r in data["required_revisions"][:6]: r_text = r if isinstance(r, str) else r.get("title", str(r)[:100]) parts.append(f" - {r_text[:150]}") except Exception: pass # 反馈合成的修改后 IDEA revised_idea = workspace / f"round_{round_num}" / "09_review_feedback" / "artifacts" / "revised_idea.md" if revised_idea.exists(): parts.append(f"\n### 修改后的 IDEA (Round {round_num} 反馈合成):") parts.append(revised_idea.read_text(encoding="utf-8")[:4000]) # 反馈合成的行动计划 action_plan = workspace / f"round_{round_num}" / "09_review_feedback" / "artifacts" / "action_plan.json" if action_plan.exists(): parts.append(f"\n### 各阶段行动计划:") parts.append(action_plan.read_text(encoding="utf-8")[:3000]) return "\n".join(parts) def cmd_claude_code(args): """ 通过 Claude Code CLI (Max Plan) 执行科研流程。 支持迭代: 评审 → 反馈合成 → 修改 IDEA → 重跑流程，直到 acceptance ≥ 50%。 可通过 --workspace 继续已有工作区的下一轮。 """ 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 if hasattr(args, "phase_timeout") and args.phase_timeout: PHASE_TIMEOUT = args.phase_timeout if hasattr(args, "cooldown") and args.cooldown is not None: PHASE_COOLDOWN = args.cooldown print(f""" ╔══════════════════════════════════════════════════════════════╗ ║ Research-67 via Claude Code CLI (Max Plan) ║ ║ Model: Claude Opus 4.6 (1M context) ║ ║ Max turns/phase: {MAX_TURNS_PER_PHASE:<3d} | Threshold: {ACCEPTANCE_THRESHOLD:.0%} | Max rounds: {MAX_ROUNDS} ║ ╚══════════════════════════════════════════════════════════════╝ """) # 检查 claude CLI try: r = subprocess.run(["claude", "--version"], capture_output=True, text=True, timeout=10) print(f" Claude CLI: {r.stdout.strip()}") except FileNotFoundError: print(" ERROR: 'claude' not found. Install Claude Code CLI.") return # 工作区: 复用已有 or 新建 if hasattr(args, "workspace") and args.workspace: workspace = Path(args.workspace) if not workspace.exists(): print(f" ERROR: Workspace not found: {workspace}") return print(f" Resuming workspace: {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" New workspace: {workspace}") # 读取 IDEA idea_path = AGENT_DIR / "paper" / "paper01" / "IDEA.md" idea_original = idea_path.read_text(encoding="utf-8") if idea_path.exists() else "" # 确定起始轮次 start_round = 1 for i in range(1, MAX_ROUNDS + 1): if (workspace / f"round_{i}" / "08_paper_review" / "report.json").exists(): prob, score = read_review_score(workspace, i) print(f" Round {i} review found: score={score}/10, acceptance={prob:.0%}") if prob >= ACCEPTANCE_THRESHOLD: print(f" ✓ Already meets threshold! No further rounds needed.") return start_round = i + 1 # 从评审后的下一步开始 else: break # 检查是否需要先跑 feedback (上一轮有评审但还没跑 feedback) need_feedback_first = False if start_round > 1: prev_round = start_round - 1 feedback_dir = workspace / f"round_{prev_round}" / "09_review_feedback" if not (feedback_dir / "report.json").exists(): need_feedback_first = True print(f" Round {prev_round} 评审已完成但 feedback 未执行，先执行反馈合成...") # ═══ 主循环 ═══ MAIN_PHASES = [ ("01_idea_analysis", "idea-analyst", "分析 IDEA 可行性"), ("02_literature_review", "lit-searcher", "检索相关文献"), ("03_experiment_plan", "exp-planner", "制定实验方案"), ("04_experiment_execution","exp-executor", "执行实验"), ("05_result_analysis", "result-analyzer", "分析实验结果"), ("06_paper_writing", "paper-writer", "撰写论文"), ("07_data_verification", "data-verifier", "核对数据"), ("08_paper_review", "paper-reviewer", "评审论文"), ] for round_num in range(start_round, MAX_ROUNDS + 1): print(f"\n{'═'*60}") print(f" 🔄 ROUND {round_num}/{MAX_ROUNDS}") print(f"{'═'*60}") # --- Phase 09: 评审反馈合成 (Round 2+ 之前，或 need_feedback_first) --- if round_num > 1 or need_feedback_first: feedback_round = round_num - 1 if round_num > 1 else start_round - 1 feedback_dir = workspace / f"round_{feedback_round}" / "09_review_feedback" if not (feedback_dir / "report.json").exists(): agent_yml = AGENT_DIR / "agents" / "review-feedback.yml" with open(agent_yml, "r", encoding="utf-8") as f: config = yaml.safe_load(f) review_context = read_revision_context(workspace, feedback_round) prompt = build_prompt( workspace, feedback_dir, "评审反馈合成 → 修改 IDEA", config.get("systemPrompt", ""), idea_original, feedback_round, [], revision_context=review_context ) run_phase(feedback_dir, "review-feedback", "评审反馈合成", prompt, feedback_round) # 读取修改后的 IDEA（如果有） idea = idea_original revision_context = "" if round_num > 1: revision_context = read_revision_context(workspace, round_num - 1) revised_idea_path = workspace / f"round_{round_num - 1}" / "09_review_feedback" / "artifacts" / "revised_idea.md" if revised_idea_path.exists(): idea = revised_idea_path.read_text(encoding="utf-8") print(f" 📝 Using revised IDEA from Round {round_num - 1}") # --- Main phases 01-08 --- prev_outputs = [] failed_phases = [] 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} (report.json exists)") existing_output = (phase_dir / "claude_output.md").read_text(encoding="utf-8") \ if (phase_dir / "claude_output.md").exists() else "" prev_outputs.append(existing_output[:5000]) continue # 上游连续失败 ≥2 个阶段 → 后续阶段跳过（防级联空转） if len(failed_phases) >= 2: consecutive_fail = all( f_idx >= phase_idx - 2 for f_idx in failed_phases[-2:] ) if consecutive_fail: print(f"\n ⏭ Skipping {phase_name} (upstream cascade failure)") prev_outputs.append("[SKIPPED] upstream failure") 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", ""), idea, round_num, prev_outputs, revision_context=revision_context if round_num > 1 else "" ) output = run_phase(phase_dir, agent_name, desc, prompt, round_num) prev_outputs.append(output[:5000]) # 检测失败 if any(tag in output for tag in ["[TIMEOUT]", "[ERROR]", "[FAILED]", "Request timed out", "overloaded_error"]): failed_phases.append(phase_idx) # 阶段间冷却（防 API 限速） if phase_idx < len(MAIN_PHASES) - 1: print(f" ⏳ Cooldown {PHASE_COOLDOWN}s before next phase...") time.sleep(PHASE_COOLDOWN) # --- 检查评审结果 --- prob, score = read_review_score(workspace, round_num) print(f"\n{'─'*60}") print(f" 📊 Round {round_num} 评审结果: score={score}/10, acceptance={prob:.0%}") if prob >= ACCEPTANCE_THRESHOLD: print(f" ✅ 达到目标阈值 ({ACCEPTANCE_THRESHOLD:.0%})! 停止迭代。") break elif round_num < MAX_ROUNDS: print(f" ⚠ 未达阈值 ({prob:.0%} < {ACCEPTANCE_THRESHOLD:.0%})，进入 Round {round_num + 1}...") else: print(f" ❌ 已达最大轮数 ({MAX_ROUNDS})，停止。最终: score={score}, acceptance={prob:.0%}") # 最终统计 print(f"\n{'═'*60}") print(f" Pipeline complete. Workspace: {workspace}") print(f" Total rounds: {round_num}") print(f" Final score: {score}/10, acceptance: {prob:.0%}") # 各轮得分追踪 print(f"\n 📈 Score progression:") for r in range(1, round_num + 1): rp, rs = read_review_score(workspace, r) marker = " ✅" if rp >= ACCEPTANCE_THRESHOLD else "" print(f" Round {r}: score={rs}/10, acceptance={rp:.0%}{marker}") # 文件统计 total_files = sum(1 for _ in workspace.rglob("*") if _.is_file()) total_size = sum(f.stat().st_size for f in workspace.rglob("*") if f.is_file()) print(f"\n 📦 Total files: {total_files}") print(f" 💾 Total size: {total_size / 1024 / 1024:.1f} MB") print(f"{'═'*60}\n") # ═══════════════════════════════════════════════════════════ # CLI Entry Point # ═══════════════════════════════════════════════════════════ def main(): parser = argparse.ArgumentParser(description="Research-67 AgentPaaS Launcher") sub = parser.add_subparsers(dest="command") # serve serve_p = sub.add_parser("serve", help="Start AgentPaaS server") serve_p.add_argument("--port", type=int, default=8000) # register sub.add_parser("register", help="Register all agents to PaaS") # run sub.add_parser("run", help="Execute research pipeline via API") # all all_p = sub.add_parser("all", help="Serve + register + run (one-shot)") all_p.add_argument("--port", type=int, default=8000) # claude-code cc_p = sub.add_parser("claude-code", help="Execute via Claude Code CLI (Max Plan)") cc_p.add_argument("--workspace", type=str, default=None, help="Resume from existing workspace (skip completed rounds)") cc_p.add_argument("--max-rounds", type=int, default=None, help=f"Override max rounds (default: {MAX_ROUNDS})") cc_p.add_argument("--max-turns", type=int, default=None, help=f"Override max turns per phase (default: {MAX_TURNS_PER_PHASE})") cc_p.add_argument("--phase-timeout", type=int, default=None, help=f"Override phase timeout in seconds (default: {PHASE_TIMEOUT})") cc_p.add_argument("--cooldown", type=int, default=None, help=f"Override inter-phase cooldown in seconds (default: {PHASE_COOLDOWN})") 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 all --port 8000 # 一键启动全流程 (需要 ANTHROPIC_API_KEY)") print(" python launch_paas.py claude-code # 使用 Claude Code CLI (Max Plan, 无需 API key)") if __name__ == "__main__": main()