From 74f5ce2f97c0fdbf7a2f5b60f56322ce4b5ade08 Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Sun, 21 Jun 2026 22:08:55 +0800 Subject: [PATCH] =?UTF-8?q?feat(launcher):=20=E5=BC=B9=E6=80=A7=20agent=20?= =?UTF-8?q?=E6=B1=A0=20=E2=80=94=20=E6=8C=89=E9=9C=80=E6=89=A9=E7=BC=A9?= =?UTF-8?q?=E5=8F=96=E4=BB=A3=E5=9B=BA=E5=AE=9A=E9=A2=84=E8=B5=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit agent_autoscale_loop(ENABLE_AGENT_AUTOSCALING):初始只起 AGENT_LAUNCH_MIN_POOL 个, 按每个 run 的 PENDING 任务积压超过空闲 agent 时 top-up,上限 MAX_POOL/per-user cap。 plan_launch_specs 加 count_override/id_start 支持非碰撞增补。trivial run 用少量 agent, 重 run 自动扩展。fail-soft,backend=none 时 no-op。 manifest:MIN/POOL 16→2 + ENABLE_AGENT_AUTOSCALING=1(初始小、按需长)。 Co-Authored-By: Claude Opus 4.8 (1M context) --- k8s/orchestrator-heicode-test.yaml | 9 +++-- orchestrator/agent_launcher.py | 14 +++++-- orchestrator/main.py | 65 ++++++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 6 deletions(-) diff --git a/k8s/orchestrator-heicode-test.yaml b/k8s/orchestrator-heicode-test.yaml index 250ff83..ccd5e9e 100644 --- a/k8s/orchestrator-heicode-test.yaml +++ b/k8s/orchestrator-heicode-test.yaml @@ -91,11 +91,14 @@ spec: - name: JINA_API_KEY valueFrom: secretKeyRef: { name: swarm-jina-key, key: JINA_API_KEY } - # --- agent 池窗口 --- + # --- agent 池窗口（弹性）--- + # 初始只起 2 个；autoscale loop 按 PENDING 积压按需扩到 MAX/per-user 上限。 + - name: ENABLE_AGENT_AUTOSCALING + value: "1" - name: AGENT_LAUNCH_MIN_POOL - value: "16" + value: "2" - name: AGENT_LAUNCH_POOL_SIZE - value: "16" + value: "2" - name: AGENT_LAUNCH_MAX_POOL value: "64" - name: MAX_AGENTS_PER_USER diff --git a/orchestrator/agent_launcher.py b/orchestrator/agent_launcher.py index 8e31222..78db573 100644 --- a/orchestrator/agent_launcher.py +++ b/orchestrator/agent_launcher.py @@ -460,6 +460,8 @@ def plan_launch_specs( orchestrator_url: str, user_id: Optional[str] = None, git_env: Optional[Dict[str, str]] = None, + count_override: Optional[int] = None, + id_start: int = 0, ) -> List[AgentLaunchSpec]: """Pure: plan the agent pool for a run (count capped by the per-user limit) + each agent's env. @@ -467,18 +469,24 @@ def plan_launch_specs( the configured pool so the pool is capability-diverse (the seed task has no required caps, so any agent can claim it; subtasks proposed later route by capability). `git_env` (from git_launch_env) carries GIT_REPO_URL so each agent clones the bound repo into its workspace. + + Elastic top-up: pass ``count_override`` (the delta to add) + ``id_start`` (current agent count) + to plan ADDITIONAL agents with non-colliding ids (``agent-{id_start+i+1}``). The caller is + responsible for capping the delta against MAX_POOL and the per-user limit (the autoscale loop). """ - count = launch_count(pool_size=pool_size, limit=limit, connected_user_agents=connected_user_agents) + count = count_override if count_override is not None else launch_count( + pool_size=pool_size, limit=limit, connected_user_agents=connected_user_agents) caps = pool_capabilities() or ["general"] base = model_api_base() mid = model_id(body) task_timeout = resolve_task_timeout(body) # #70:透传执行超时，与 run budget.duration_seconds 取较小 specs: List[AgentLaunchSpec] = [] for i in range(count): - cap_csv = caps[i % len(caps)] + idx = id_start + i + cap_csv = caps[idx % len(caps)] env = { "ORCHESTRATOR_URL": orchestrator_url, - "AGENT_ID": f"{run.swarm_id}-agent-{i+1}", + "AGENT_ID": f"{run.swarm_id}-agent-{idx+1}", "AGENT_CAPABILITIES": cap_csv, "OPENAI_API_BASE": base, # Non-secret execution-timeout knob; inline like the other config env (never a Secret). diff --git a/orchestrator/main.py b/orchestrator/main.py index 1036f94..fce7d0a 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -449,6 +449,69 @@ async def finalize_dispatch(agent, task, *, decision=None, dispatch_event=None) return True +async def _maybe_topup_agents(run, backlog: int) -> None: + """Elastic top-up for one run: if its PENDING backlog exceeds idle connected agents and the + pool is below the cap, launch the deficit (capped by MAX_POOL and the run's per-user limit). + New agents get non-colliding ids (id_start = current pool size). Fail-soft.""" + swarm_id = run.swarm_id + prefix = f"{swarm_id}-agent-" + live = sum(1 for aid in manager.active_connections if aid.startswith(prefix)) + launched = run.metadata.get("launched_agents") or [] + pool = max(live, len(launched)) # avoid re-launching agents still connecting + idle = sum(1 for a in await agent_registry.get_idle_agents() + if a.agent_id.startswith(prefix) and a.agent_id in manager.active_connections + and agent_has_capacity(a.agent_id)) + body = run.request_body or {} + cap = min(agent_launcher.max_pool_size(), max_agents_per_user(body)) + if backlog <= idle or pool >= cap: + return + delta = min(backlog - idle, cap - pool) + if delta <= 0: + return + user_id = ((body.get("metadata") or {}).get("runtime_headers") or {}).get("x_user_id") + specs = agent_launcher.plan_launch_specs( + run, body, connected_user_agents=0, limit=max_agents_per_user(body), + pool_size=delta, model_key=agent_launcher.resolve_model_key(body), + orchestrator_url=agent_launcher.orchestrator_ws_url(), user_id=user_id, + git_env=agent_launcher.resolve_git_grant(body), count_override=delta, id_start=pool) + launched_ids = await agent_launcher.launch(specs, swarm_id=swarm_id) + if launched_ids: + run.metadata["launched_agents"] = launched + launched_ids + await swarm_runtime.save_run(run) + logger.info("autoscale: run %s +%d agents (backlog=%d idle=%d pool=%d cap=%d)", + swarm_id, len(launched_ids), backlog, idle, pool, cap) + + +async def agent_autoscale_loop(): + """Elastic agent pool (demand-driven). Initial launch stays small (AGENT_LAUNCH_MIN_POOL); this + loop grows each run's pool only as its PENDING-task backlog demands, up to MAX_POOL / per-user + cap — so trivial runs use few agents and heavy runs scale out. Gated by ENABLE_AGENT_AUTOSCALING; + no-op when the launch backend is 'none'. Fail-soft (never raises into the event loop).""" + if os.getenv("ENABLE_AGENT_AUTOSCALING", "false").lower() not in {"1", "true", "yes"}: + return + if agent_launcher.launch_backend() == "none": + return + interval = float(os.getenv("AUTOSCALE_INTERVAL_SECONDS", "5")) + while True: + try: + await asyncio.sleep(interval) + if await task_queue.get_pending_count() == 0: + continue + backlog: Dict[str, list] = {} # swarm_id -> [run, count] + for task in await task_queue.get_all_tasks(): + if task.status != TaskStatus.PENDING: + continue + run = await swarm_runtime.get_run_for_task(task.task_id) + if not run or run.status != "running": + continue + entry = backlog.setdefault(run.swarm_id, [run, 0]) + entry[1] += 1 + for run, n in backlog.values(): + await _maybe_topup_agents(run, n) + except Exception as e: + logger.error(f"Error in agent autoscale loop: {e}") + + async def compute_convergence_report(run, tasks, *, terminal: bool): """Build a run-state snapshot and evaluate the convergence report (#12, shadow by default). @@ -851,6 +914,7 @@ async def lifespan(app: FastAPI): # Start background tasks failure_task = asyncio.create_task(failure_detection_loop()) dispatch_task = asyncio.create_task(task_dispatch_loop()) + autoscale_task = asyncio.create_task(agent_autoscale_loop()) yield @@ -858,6 +922,7 @@ async def lifespan(app: FastAPI): logger.info("Shutting down orchestrator...") failure_task.cancel() dispatch_task.cancel() + autoscale_task.cancel() await redis_client.disconnect()