lambdagent/agentexample/agentbuilder67/launch_paas.py

#!/usr/bin/env python3
"""
agentbuilder67/launch_paas.py — 多智能体 AgentBuilder 启动器
=============================================================

用户输入一句话描述 → 6 个智能体协作 → 输出可部署的 agent YAML 配置

两种运行模式:
  1. AgentPaaS API 模式: 启动 PaaS 服务，注册 builder agents，通过 REST API 执行
  2. Claude Code CLI 模式: 利用 Claude CLI 作为 LLM 后端，本地执行多阶段流水线

用法:
  # 模式 1: PaaS API
  python launch_paas.py serve                     # 启动服务
  python launch_paas.py register                  # 注册所有 builder agents
  python launch_paas.py build "做一个股票分析agent"  # 通过 API 构建

  # 模式 2: Claude Code CLI (推荐)
  python launch_paas.py claude-code "做一个能分析CSV数据的agent"
  python launch_paas.py claude-code --interactive  # 交互模式

  # 优化已部署的 agent
  python launch_paas.py optimize <agent_id>
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional

PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
BUILDER_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", "")

# ── Builder Agent 配置 ────────────────────────────────────
AGENT_CONFIGS = [
    ("builder-orchestrator", "AgentBuilder 总指挥", BUILDER_DIR / "orchestrator.yml"),
    ("builder-analyst",      "需求分析师",        BUILDER_DIR / "agents" / "analyst.yml"),
    ("builder-retriever",    "配置检索员",        BUILDER_DIR / "agents" / "retriever.yml"),
    ("builder-prompter",     "Prompt 工匠",       BUILDER_DIR / "agents" / "prompter.yml"),
    ("builder-assembler",    "配置组装师",        BUILDER_DIR / "agents" / "assembler.yml"),
    ("builder-critic",       "配置审查官",        BUILDER_DIR / "agents" / "critic.yml"),
    ("builder-optimizer",    "持续优化器",        BUILDER_DIR / "agents" / "optimizer.yml"),
]

# ── 运行参数 ──────────────────────────────────────────────
MAX_REVISION_ROUNDS = 3          # Critic 不通过时最大修正轮数
MAX_TURNS_PER_PHASE = 20         # 每个 agent 最大工具调用轮数
PHASE_TIMEOUT = 300              # 每个 phase 超时 (5 分钟，builder 比 research 快)
API_RETRY_MAX = 3                # API 失败最大重试次数
API_RETRY_BACKOFF = 15           # 重试间隔基数 (秒)
PHASE_COOLDOWN = 5               # 阶段间冷却 (秒)
CRITIC_PASS_SCORE = 7.0          # Critic 通过分数线


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

def api_call(method: str, path: str, data: dict = None) -> dict:
    import urllib.request
    import urllib.error

    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")
    if API_KEY:
        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) -> bool:
    import urllib.request
    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


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

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}")
    print(f"  URL: http://127.0.0.1:{args.port}")

    if not wait_for_server(30):
        print("  ERROR: Server failed to start within 30s")
        proc.kill()
        return None

    print("  ✓ Server ready")
    return proc


# ═══════════════════════════════════════════════════════════
# 3. 命令: register — 注册所有 builder agents
# ═══════════════════════════════════════════════════════════

def cmd_register(args):
    import yaml

    print(f"\nRegistering {len(AGENT_CONFIGS)} builder 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": ["agentbuilder", 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')}")

    mapping_path = BUILDER_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. 命令: build — 通过 PaaS API 构建 agent
# ═══════════════════════════════════════════════════════════

def cmd_build(args):
    mapping_path = BUILDER_DIR / ".paas_agent_ids.json"
    if not mapping_path.exists():
        print("ERROR: No agent ID mapping. Run 'register' first.")
        return

    agent_ids = json.loads(mapping_path.read_text())
    orchestrator_id = agent_ids.get("builder-orchestrator")
    if not orchestrator_id:
        print("ERROR: Orchestrator not registered.")
        return

    user_desc = args.description
    print(f"\n{'═'*60}")
    print(f"  AgentBuilder — building agent from description")
    print(f"  Input: {user_desc[:80]}...")
    print(f"{'═'*60}\n")

    result = api_call("POST", f"/agents/{orchestrator_id}/run", {
        "input": user_desc,
        "parameters": {},
        "context": {"sub_agent_ids": agent_ids},
    })

    if "error" not in result:
        print(f"\n  ✓ Build completed")
        print(f"    Run ID: {result.get('run_id')}")
        output = result.get("output", "")
        print(f"\n{output}")
    else:
        print(f"\n  ✗ Build failed: {result['error'][:500]}")


# ═══════════════════════════════════════════════════════════
# 5. Claude Code CLI — 核心多阶段流水线
# ═══════════════════════════════════════════════════════════

def run_claude_phase(phase_dir: Path, phase_name: str, desc: str,
                     prompt_content: str) -> str:
    """执行单个 Claude CLI 阶段，带重试。"""
    phase_dir.mkdir(parents=True, exist_ok=True)
    (phase_dir / "prompt.md").write_text(prompt_content, encoding="utf-8")

    print(f"\n{'━'*60}")
    print(f"  Phase: {phase_name} — {desc}")
    print(f"  Timeout: {PHASE_TIMEOUT}s | Max turns: {MAX_TURNS_PER_PHASE}")
    print(f"{'━'*60}")

    cmd = [
        "claude",
        "-p",
        "--model", "claude-sonnet-4-20250514",
        "--max-turns", str(MAX_TURNS_PER_PHASE),
        "--allowedTools", "Read,Glob,Grep",
    ]

    for attempt in range(1, API_RETRY_MAX + 1):
        t0 = time.time()
        label = f"[attempt {attempt}/{API_RETRY_MAX}]"

        try:
            print(f"  🚀 {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"]:
                if err_pattern in output or err_pattern in stderr:
                    is_api_error = True
                    break

            if is_api_error and attempt < API_RETRY_MAX:
                wait = API_RETRY_BACKOFF * (2 ** (attempt - 1))
                print(f"  ⚠ {label} API error after {elapsed:.0f}s. Retrying in {wait}s ...")
                time.sleep(wait)
                continue

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

        except subprocess.TimeoutExpired:
            elapsed = time.time() - t0
            msg = f"[TIMEOUT] Phase {phase_name} timed out after {elapsed:.0f}s"
            print(f"  ✗ {msg}")
            (phase_dir / "output.md").write_text(msg, encoding="utf-8")
            return msg

        except FileNotFoundError:
            print("  ✗ 'claude' CLI not found. Is Claude Code installed?")
            return "[ERROR] claude CLI not found"

    return f"[FAILED] All {API_RETRY_MAX} attempts failed"


def load_reference_docs() -> str:
    """加载 schema 参考和示例文档。"""
    parts = []
    schema_path = BUILDER_DIR / "prompts" / "schema_reference.md"
    examples_path = BUILDER_DIR / "prompts" / "examples.md"

    if schema_path.exists():
        parts.append(schema_path.read_text(encoding="utf-8"))
    if examples_path.exists():
        parts.append(examples_path.read_text(encoding="utf-8"))

    return "\n\n---\n\n".join(parts)


def load_agent_prompt(agent_name: str) -> str:
    """从 YAML 配置中读取 agent 的 systemPrompt。"""
    import yaml

    yml_path = BUILDER_DIR / "agents" / f"{agent_name}.yml"
    if not yml_path.exists():
        return ""
    with open(yml_path, "r", encoding="utf-8") as f:
        config = yaml.safe_load(f)
    return config.get("systemPrompt", "")


def build_phase_prompt(agent_name: str, user_desc: str,
                       context: Dict = None, revision: str = "") -> str:
    """构建阶段 prompt。"""
    system_prompt = load_agent_prompt(agent_name)
    reference = load_reference_docs()

    parts = [
        f"# Agent Configuration Builder — {agent_name}",
        f"\n## System Instructions\n{system_prompt}",
        f"\n## User Description\n{user_desc}",
    ]

    if context:
        parts.append("\n## Previous Phase Outputs")
        for key, value in context.items():
            parts.append(f"\n### {key}\n{str(value)[:4000]}")

    if revision:
        parts.append(f"\n## Revision Instructions (from Critic)\n{revision}")

    parts.append(f"\n## Reference: YAML Schema & Examples\n{reference[:6000]}")

    parts.append("""
## Output Instructions
- Output ONLY the result in the specified format (JSON or YAML)
- Do not wrap output in markdown code blocks unless it is YAML
- Be precise and concise
""")

    return "\n".join(parts)


def parse_critic_output(output: str) -> Dict:
    """从 Critic 输出中提取 JSON 评分。"""
    # 尝试直接解析
    try:
        return json.loads(output)
    except (json.JSONDecodeError, TypeError):
        pass

    # 尝试从 markdown 代码块中提取
    import re
    json_match = re.search(r'```(?:json)?\s*\n(.*?)\n```', output, re.DOTALL)
    if json_match:
        try:
            return json.loads(json_match.group(1))
        except (json.JSONDecodeError, TypeError):
            pass

    # 尝试找 { } 块
    brace_match = re.search(r'\{[^{}]*"overall_score"[^{}]*\}', output, re.DOTALL)
    if brace_match:
        try:
            return json.loads(brace_match.group(0))
        except (json.JSONDecodeError, TypeError):
            pass

    return {"overall_score": 0, "verdict": "unknown", "parse_error": True}


def extract_yaml_from_output(output: str) -> str:
    """从 Assembler 输出中提取 YAML 配置。"""
    import re

    # 尝试从 ```yaml 代码块中提取
    yaml_match = re.search(r'```ya?ml\s*\n(.*?)\n```', output, re.DOTALL)
    if yaml_match:
        return yaml_match.group(1).strip()

    # 尝试找以 agentId: 开头的块
    agent_match = re.search(r'(agentId:.*)', output, re.DOTALL)
    if agent_match:
        return agent_match.group(1).strip()

    return output.strip()


def cmd_claude_code(args):
    """
    通过 Claude Code CLI 执行多智能体构建流水线。

    Layer 1 (并行): Analyst + Retriever + Prompter
    Layer 2 (串行): Assembler → Critic
    Loop: 如果 Critic 不通过，修正后重新走 Layer 2
    """
    import yaml

    user_desc = args.description
    if not user_desc:
        print("ERROR: Please provide a description. Example:")
        print('  python launch_paas.py claude-code "做一个能分析股票K线的agent"')
        return

    # 检查 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

    # 创建工作区
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    workspace = BUILDER_DIR / "workspace" / f"build_{ts}"
    workspace.mkdir(parents=True, exist_ok=True)

    # 保存输入
    (workspace / "user_input.txt").write_text(user_desc, encoding="utf-8")

    print(f"""
╔══════════════════════════════════════════════════════════════╗
║  AgentBuilder — Multi-Agent Configuration Generator          ║
║  Model: Claude Sonnet 4                                      ║
║  Max revision rounds: {MAX_REVISION_ROUNDS}                                      ║
╚══════════════════════════════════════════════════════════════╝

  Input: {user_desc}
  Workspace: {workspace}
""")

    # ═══ Layer 1: 并行分析 ═══════════════════════════════
    print(f"\n{'═'*60}")
    print(f"  LAYER 1 — Parallel Analysis (Analyst + Retriever + Prompter)")
    print(f"{'═'*60}")

    # Layer 1 的三个阶段按顺序执行（CLI 模式下不方便真正并行，但各自独立）
    # Analyst
    analyst_output = run_claude_phase(
        workspace / "01_analyst", "analyst", "需求分析",
        build_phase_prompt("analyst", user_desc),
    )
    time.sleep(PHASE_COOLDOWN)

    # Retriever
    retriever_output = run_claude_phase(
        workspace / "02_retriever", "retriever", "配置检索",
        build_phase_prompt("retriever", user_desc, context={"analyst": analyst_output}),
    )
    time.sleep(PHASE_COOLDOWN)

    # Prompter
    prompter_output = run_claude_phase(
        workspace / "03_prompter", "prompter", "Prompt 撰写",
        build_phase_prompt("prompter", user_desc, context={"analyst": analyst_output}),
    )
    time.sleep(PHASE_COOLDOWN)

    # ═══ Layer 2: 组装 + 审查（可迭代）═══════════════════
    assembler_context = {
        "analyst_output": analyst_output,
        "retriever_output": retriever_output,
        "prompter_output": prompter_output,
    }

    final_config = None
    final_score = 0
    revision_instructions = ""

    for round_num in range(1, MAX_REVISION_ROUNDS + 1):
        print(f"\n{'═'*60}")
        print(f"  LAYER 2 — Assembly + Review (Round {round_num}/{MAX_REVISION_ROUNDS})")
        print(f"{'═'*60}")

        # Assembler
        assembler_output = run_claude_phase(
            workspace / f"04_assembler_r{round_num}", "assembler",
            f"配置组装 (Round {round_num})",
            build_phase_prompt("assembler", user_desc,
                               context=assembler_context,
                               revision=revision_instructions),
        )
        time.sleep(PHASE_COOLDOWN)

        # 提取 YAML
        generated_yaml = extract_yaml_from_output(assembler_output)
        (workspace / f"04_assembler_r{round_num}" / "generated.yml").write_text(
            generated_yaml, encoding="utf-8"
        )

        # Lint 检查
        from tools.config_lint import lint_config
        try:
            lint_result = lint_config(generated_yaml)
            (workspace / f"04_assembler_r{round_num}" / "lint_result.json").write_text(
                json.dumps(lint_result, indent=2, ensure_ascii=False), encoding="utf-8"
            )
            print(f"  Lint: valid={lint_result['valid']}, score={lint_result['score']}, "
                  f"errors={len(lint_result['errors'])}, warnings={len(lint_result['warnings'])}")
        except Exception as e:
            lint_result = {"valid": False, "errors": [str(e)], "warnings": [], "score": 0}
            print(f"  Lint error: {e}")

        # Critic
        critic_context = {
            "user_description": user_desc,
            "generated_config": generated_yaml,
            "lint_result": json.dumps(lint_result, ensure_ascii=False),
        }
        critic_output = run_claude_phase(
            workspace / f"05_critic_r{round_num}", "critic",
            f"配置审查 (Round {round_num})",
            build_phase_prompt("critic", user_desc, context=critic_context),
        )

        # 解析 Critic 评分
        critic_result = parse_critic_output(critic_output)
        (workspace / f"05_critic_r{round_num}" / "scores.json").write_text(
            json.dumps(critic_result, indent=2, ensure_ascii=False), encoding="utf-8"
        )

        overall_score = critic_result.get("overall_score", 0)
        verdict = critic_result.get("verdict", "unknown")
        print(f"\n  📊 Critic: score={overall_score}/10, verdict={verdict}")

        if overall_score >= CRITIC_PASS_SCORE or verdict == "approve":
            print(f"  ✅ Approved! Score {overall_score} >= {CRITIC_PASS_SCORE}")
            final_config = generated_yaml
            final_score = overall_score
            break
        elif round_num < MAX_REVISION_ROUNDS:
            # 提取修改指令
            revision_instructions = ""
            rev_inst = critic_result.get("revision_instructions", {})
            if isinstance(rev_inst, dict):
                for target, instruction in rev_inst.items():
                    if instruction:
                        revision_instructions += f"\n[{target}]: {instruction}"
            critical = critic_result.get("critical_issues", [])
            if critical:
                revision_instructions += "\n\nCritical issues:\n" + "\n".join(
                    f"- {issue}" for issue in critical
                )
            suggestions = critic_result.get("suggestions", [])
            if suggestions:
                revision_instructions += "\n\nSuggestions:\n" + "\n".join(
                    f"- {s}" for s in suggestions
                )

            print(f"  ⚠ Score {overall_score} < {CRITIC_PASS_SCORE}, revising...")
            print(f"  Revision notes: {revision_instructions[:200]}...")
            time.sleep(PHASE_COOLDOWN)
        else:
            print(f"  ❌ Max rounds reached. Using best result (score={overall_score})")
            final_config = generated_yaml
            final_score = overall_score

    # ═══ 输出最终结果 ═══════════════════════════════════
    if final_config:
        output_path = workspace / "final_agent.yml"
        output_path.write_text(final_config, encoding="utf-8")

        print(f"\n{'═'*60}")
        print(f"  ✅ AgentBuilder Complete!")
        print(f"{'═'*60}")
        print(f"  Score: {final_score}/10")
        print(f"  Output: {output_path}")
        print(f"  Workspace: {workspace}")
        print(f"\n  Generated config:")
        print(f"{'─'*60}")
        print(final_config[:2000])
        if len(final_config) > 2000:
            print(f"  ... ({len(final_config)} chars total)")
        print(f"{'─'*60}")

        # 尝试解析并展示摘要
        try:
            config = yaml.safe_load(final_config)
            if isinstance(config, dict):
                print(f"\n  Summary:")
                print(f"    Agent ID: {config.get('agentId', '?')}")
                print(f"    Name: {config.get('name', '?')}")
                print(f"    Type: {config.get('type', '?')}")
                print(f"    Model: {config.get('model', {}).get('provider', '?')}"
                      f"/{config.get('model', {}).get('name', '?')}")
                tools = config.get('mcp', {}).get('localTools', [])
                print(f"    Tools: {', '.join(tools)}")
        except Exception:
            pass
    else:
        print(f"\n  ❌ Build failed. Check workspace: {workspace}")


# ═══════════════════════════════════════════════════════════
# 6. 命令: interactive — 交互式构建
# ═══════════════════════════════════════════════════════════

def cmd_interactive(args):
    """交互式 AgentBuilder：对话式构建 + 实时预览。"""
    print(f"""
╔══════════════════════════════════════════════════════════════╗
║  AgentBuilder Interactive Mode                                ║
║  输入你想要的 agent 描述，我来生成配置                           ║
║  输入 quit 退出 | 输入 save 保存当前配置                         ║
╚══════════════════════════════════════════════════════════════╝
""")

    current_config = None

    while True:
        try:
            user_input = input("\n🔧 描述你的 agent > ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nBye!")
            break

        if not user_input:
            continue
        if user_input.lower() == "quit":
            break
        if user_input.lower() == "save" and current_config:
            save_path = BUILDER_DIR / "workspace" / f"agent_{datetime.now().strftime('%H%M%S')}.yml"
            save_path.parent.mkdir(parents=True, exist_ok=True)
            save_path.write_text(current_config, encoding="utf-8")
            print(f"  ✓ Saved to {save_path}")
            continue

        # 构建 agent（复用 claude-code 逻辑）
        args.description = user_input
        cmd_claude_code(args)


# ═══════════════════════════════════════════════════════════
# 7. 命令: optimize — 优化已部署的 agent
# ═══════════════════════════════════════════════════════════

def cmd_optimize(args):
    """通过 Claude CLI 运行 Optimizer agent。"""
    agent_id = args.agent_id

    print(f"\n{'═'*60}")
    print(f"  AgentBuilder Optimizer")
    print(f"  Target: {agent_id}")
    print(f"{'═'*60}\n")

    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    workspace = BUILDER_DIR / "workspace" / f"optimize_{ts}"
    workspace.mkdir(parents=True, exist_ok=True)

    optimizer_prompt = load_agent_prompt("optimizer")
    prompt = f"""# Agent Optimizer

## System Instructions
{optimizer_prompt}

## Target Agent
agent_id: {agent_id}

## Task
1. 调用 get_run_stats 获取运行数据
2. 调用 get_user_feedback 获取用户反馈
3. 调用 get_current_config 获取当前配置
4. 分析数据，找出优化点
5. 生成优化补丁

## Environment
- AgentPaaS API: {PAAS_URL}
- API Key: {'set' if API_KEY else 'not set'}

输出优化建议 JSON。
"""

    output = run_claude_phase(
        workspace / "optimizer", "optimizer",
        f"优化 {agent_id}",
        prompt,
    )

    print(f"\n  Optimizer output:")
    print(output[:2000])


# ═══════════════════════════════════════════════════════════
# 8. 命令: all — 一键启动全流程
# ═══════════════════════════════════════════════════════════

def cmd_all(args):
    """一键: 启动服务 + 注册 agents + 构建。"""
    global API_KEY

    proc = cmd_serve(args)
    if not proc:
        return

    try:
        # 创建租户
        result = subprocess.run(
            [sys.executable, "-m", "agentpaas", "create-tenant", "--name", "builder-lab"],
            capture_output=True, text=True, cwd=str(PROJECT_ROOT),
        )
        print(result.stdout)
        for line in 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_build(args)

    finally:
        print("\nShutting down server...")
        proc.terminate()
        proc.wait(timeout=5)


# ═══════════════════════════════════════════════════════════
# CLI Entry Point
# ═══════════════════════════════════════════════════════════

def _apply_overrides(args):
    global MAX_REVISION_ROUNDS, MAX_TURNS_PER_PHASE, PHASE_TIMEOUT
    if hasattr(args, "max_rounds") and args.max_rounds:
        MAX_REVISION_ROUNDS = args.max_rounds
    if hasattr(args, "max_turns") and args.max_turns:
        MAX_TURNS_PER_PHASE = args.max_turns
    if hasattr(args, "timeout") and args.timeout:
        PHASE_TIMEOUT = args.timeout


def main():
    parser = argparse.ArgumentParser(
        description="AgentBuilder — Multi-agent Configuration Generator",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python launch_paas.py claude-code "做一个能分析CSV数据的agent"
  python launch_paas.py claude-code "帮我做一个能上网搜索的调研助手"
  python launch_paas.py claude-code --interactive
  python launch_paas.py optimize agent-abc123
  python launch_paas.py all --port 8000 -d "股票K线分析agent"
""")
    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 builder agents to PaaS")

    # build (via PaaS API)
    build_p = sub.add_parser("build", help="Build agent via PaaS API")
    build_p.add_argument("description", type=str, help="Agent description")

    # claude-code (via CLI)
    cc_p = sub.add_parser("claude-code", help="Build agent via Claude Code CLI")
    cc_p.add_argument("description", type=str, nargs="?", default="",
                       help="Agent description")
    cc_p.add_argument("--interactive", "-i", action="store_true",
                       help="Interactive mode")
    cc_p.add_argument("--max-rounds", type=int, default=None,
                       help=f"Max revision rounds (default: {MAX_REVISION_ROUNDS})")
    cc_p.add_argument("--max-turns", type=int, default=None,
                       help=f"Max turns per phase (default: {MAX_TURNS_PER_PHASE})")
    cc_p.add_argument("--timeout", type=int, default=None,
                       help=f"Phase timeout in seconds (default: {PHASE_TIMEOUT})")

    # optimize
    opt_p = sub.add_parser("optimize", help="Optimize a deployed agent")
    opt_p.add_argument("agent_id", type=str, help="Target agent ID")

    # all
    all_p = sub.add_parser("all", help="Serve + register + build (one-shot)")
    all_p.add_argument("--port", type=int, default=8000)
    all_p.add_argument("-d", "--description", type=str, required=True,
                       help="Agent description")

    args = parser.parse_args()

    # 应用参数覆盖
    _apply_overrides(args)

    commands = {
        "serve": cmd_serve,
        "register": cmd_register,
        "build": cmd_build,
        "claude-code": lambda a: cmd_interactive(a) if getattr(a, "interactive", False) else cmd_claude_code(a),
        "optimize": cmd_optimize,
        "all": cmd_all,
    }

    if args.command in commands:
        commands[args.command](args)
    else:
        parser.print_help()


if __name__ == "__main__":
    main()