From a03e659bc9817f3cc225ddb0cc0cd3a9311734ae Mon Sep 17 00:00:00 2001 From: Andrew Park Date: Wed, 13 May 2026 15:21:45 -0700 Subject: [PATCH] hybrid: ship CLI runner + registry + new_experiment.sh + README MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Final piece — make the ported paradigms runnable end-to-end through the same flow as the hybrid harness: - `runner.py`: `python -m openjarvis.agents.hybrid.runner --cell `. Loads cells from packaged registry TOMLs, constructs the registered agent via AgentRegistry, iterates tasks from OpenJarvis's existing GAIADataset / SWEBenchDataset, scores via gaia exact-match or the new Modal-backed SWE-bench harness scorer, writes `results.jsonl`+`summary.json` in the same shape as the hybrid harness so existing rescore / dashboard scripts work unmodified. Supports resume (drops errored rows), per-cell flock, ThreadPoolExecutor concurrency, and the same per-row heartbeat format. - `registry/{advisors,archon,conductor,minions,skillorchestra,toolorchestra}.toml`: 35 cells ported verbatim from the hybrid harness — same models, same N, same `method_cfg` — so OpenJarvis runs should reproduce the numbers in `hybrid-local-cloud-compute/docs/results.md`. ToolOrchestra cells rewritten from the original NotImplementedError stubs to point at the prompted port (cloud-as-orchestrator, no Nemotron-Orchestrator-8B). - `scripts/new_experiment.sh`: mirror of hybrid's scaffolder. Maps shortnames (`qwen3.5-27b`, `claude-opus-4-7`, …) to full HF ids + endpoints, picks method-appropriate `method_cfg` defaults, appends a `[cells.]` block to the right registry TOML. - `README.md`: quickstart, paradigm taxonomy, reproducibility table pointing at the hybrid results for headline numbers. End-to-end smoke: `python -m openjarvis.agents.hybrid.runner --help` works, all 6 agents register, all 35 cells load. --- src/openjarvis/agents/hybrid/README.md | 103 ++++ .../agents/hybrid/registry/advisors.toml | 41 ++ .../agents/hybrid/registry/archon.toml | 94 ++++ .../agents/hybrid/registry/conductor.toml | 54 ++ .../agents/hybrid/registry/minions.toml | 96 ++++ .../hybrid/registry/skillorchestra.toml | 47 ++ .../agents/hybrid/registry/toolorchestra.toml | 30 ++ src/openjarvis/agents/hybrid/runner.py | 495 ++++++++++++++++++ .../agents/hybrid/scripts/new_experiment.sh | 108 ++++ 9 files changed, 1068 insertions(+) create mode 100644 src/openjarvis/agents/hybrid/README.md create mode 100644 src/openjarvis/agents/hybrid/registry/advisors.toml create mode 100644 src/openjarvis/agents/hybrid/registry/archon.toml create mode 100644 src/openjarvis/agents/hybrid/registry/conductor.toml create mode 100644 src/openjarvis/agents/hybrid/registry/minions.toml create mode 100644 src/openjarvis/agents/hybrid/registry/skillorchestra.toml create mode 100644 src/openjarvis/agents/hybrid/registry/toolorchestra.toml create mode 100644 src/openjarvis/agents/hybrid/runner.py create mode 100755 src/openjarvis/agents/hybrid/scripts/new_experiment.sh diff --git a/src/openjarvis/agents/hybrid/README.md b/src/openjarvis/agents/hybrid/README.md new file mode 100644 index 00000000..263ad7bf --- /dev/null +++ b/src/openjarvis/agents/hybrid/README.md @@ -0,0 +1,103 @@ +# Hybrid local+cloud paradigm agents + +Six paradigms ported from +[`/matx/u/aspark/hybrid-local-cloud-compute`](../../../../../) — each is +registered as a standard OpenJarvis agent so the rest of the platform +(SDK, CLI, distillation, evals) can use them like any other agent. + +| Agent | Plan shape | Trains what? | Workers | +|-------------------|-----------------|----------------------|---------------------------| +| `minions` | reactive loop | nothing | 1 local + 1 cloud | +| `conductor` | static DAG | (paper: 7B planner) | up to 5 frontier+open | +| `archon` | static recipe | nothing (search) | K local + cloud rank/fuse | +| `advisors` | reactive loop | (paper: local model) | 1 local + 1 cloud | +| `skillorchestra` | per-query pick | (paper: profiler) | 1 local + 1 cloud | +| `toolorchestra` | reactive loop | (paper: 8B RL) | local + tools + LLM pool | + +Items in parentheses are what the *paper* trains. These OpenJarvis ports +are **inference-only** — none modify weights. The trained variants (advisor +RL, Orchestrator-8B, SkillOrchestra learn-phase) stay TODOs; the prompted +lower-bounds get you 80-90% of the headline accuracy at zero training cost. + +## What's where + +``` +src/openjarvis/agents/hybrid/ +├── _base.py LocalCloudAgent ABC + SDK helpers +├── _prices.py cloud-model pricing + temp-strip quirks +├── _prompts.py GAIA / SWE-bench answer-format instructions +├── advisors.py executor ↔ advisor ↔ executor (3-step) +├── conductor.py static DAG planner +├── minions.py HazyResearch Minions wrapper +├── archon.py Archon (generator → ranker → fuser) +├── skillorchestra.py skill-aware router +├── toolorchestra.py prompted multi-turn tool dispatcher +├── runner.py CLI: python -m ...hybrid.runner --cell NAME +├── registry/ .toml — one cell per (bench, local, cloud, N) +└── scripts/ + └── new_experiment.sh scaffold a new cell, run instructions +``` + +The Modal-backed SWE-bench-Verified scorer is in +`src/openjarvis/evals/scorers/swebench_harness.py` (next to the existing +structural scorer). + +## Quickstart + +```bash +cd /matx/u/aspark/OpenJarvis +source .env # API keys + +# 1. Start vLLM in another shell (see CLAUDE.md for the full recipe) +# CUDA_VISIBLE_DEVICES=0 .venv/bin/python -m vllm.entrypoints.openai.api_server \ +# --model Qwen/Qwen3.5-27B-FP8 --port 8001 ... + +# 2. (Optional) for Minions: install the upstream library +.venv/bin/uv pip install -e /matx/u/aspark/hybrid-local-cloud-compute/external/minions + +# 3. Run a smoke cell +.venv/bin/python -m openjarvis.agents.hybrid.runner \ + --cell minions-gaia-qwen27b-opus-3 +``` + +Outputs land in +`$OPENJARVIS_HYBRID_EXPERIMENTS_DIR//{results.jsonl,summary.json,config.json,logs/}` +(defaults to `~/.openjarvis-hybrid/experiments/`). The schema matches the +hybrid harness so the existing rescore / dashboard scripts work +unmodified. + +## Adding a cell + +```bash +src/openjarvis/agents/hybrid/scripts/new_experiment.sh \ + --method conductor --bench gaia \ + --local qwen3.5-27b --cloud claude-opus-4-7 --n 30 +``` + +That appends a `[cells.]` block to +`registry/conductor.toml` and prints the runner invocation. + +## Reproducing the hybrid harness numbers + +The cell configs in `registry/` are copies of the hybrid harness's +`experiments/registry/` — same models, same N, same `method_cfg`. The +agent code is a faithful port of each adapter (same prompts, same JSON +schemas, same patches), so running these cells in OpenJarvis should +reproduce the numbers in +`/matx/u/aspark/hybrid-local-cloud-compute/docs/results.md`: + +| paradigm | bench | N | acc | $/task | +|-----------------|----------|-----|--------|--------| +| minions | swebench | 500 | 0.274 | $0.09 | +| minions | gaia | 165 | 0.576 | $0.67 | +| conductor | swebench | 30 | 0.367 | $0.22 | +| skillorchestra | gaia | 30 | 0.500 | $0.02 | +| advisors | gaia | 30 | 0.533 | $0.02 | + +Baselines for comparison: +- `baseline-cloud-gaia-opus` = 0.570 / $1.09 per task +- `baseline-cloud-swebench-opus` = 0.236 / $0.95 per task + +The hybrid harness stays the authoritative reference for n=500 numbers +while we validate the port. Once OpenJarvis cells reproduce the headline +accuracies within noise, the harness can be deprecated in favor of these. diff --git a/src/openjarvis/agents/hybrid/registry/advisors.toml b/src/openjarvis/agents/hybrid/registry/advisors.toml new file mode 100644 index 00000000..4f366360 --- /dev/null +++ b/src/openjarvis/agents/hybrid/registry/advisors.toml @@ -0,0 +1,41 @@ +# Advisors cells (arXiv:2510.02453). +# +# Inference-only repro of the advisor paradigm: cloud drafts → local advisor +# critiques → cloud rewrites. The paper's RL-trained advisor isn't reproduced +# (no training in this harness) — these are the untrained-advisor lower bound. +# Adapter auto-resolves the local model name against /v1/models. + +[cells.advisors-gaia-qwen9b-opus-3] +method = "advisors" +bench = "gaia" +n = 3 +local = { model = "Qwen/Qwen3.5-9B", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = {} + +[cells.advisors-swebenchverified-qwen9b-opus-3] +method = "advisors" +bench = "swebench-verified" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = {} +concurrency = 3 + +[cells.advisors-gaia-qwen9b-opus-30] +method = "advisors" +bench = "gaia" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = {} +concurrency = 4 + +[cells.advisors-swebenchverified-qwen9b-opus-30] +method = "advisors" +bench = "swebench-verified" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = {} +concurrency = 3 diff --git a/src/openjarvis/agents/hybrid/registry/archon.toml b/src/openjarvis/agents/hybrid/registry/archon.toml new file mode 100644 index 00000000..7a3c3096 --- /dev/null +++ b/src/openjarvis/agents/hybrid/registry/archon.toml @@ -0,0 +1,94 @@ +# Archon cells. Inference-time architecture search (arXiv:2409.15254). +# +# Hybrid wiring: K local proposers (vLLM on 8001/8004) → 1 cloud ranker → +# 1 cloud fuser. method_cfg.architecture = "ensemble_rank_fuse" is the +# only non-debug preset; "single_local" exists for adapter smoke tests +# (no cloud calls). +# +# Naming: archon------ + +# ---- Stage 1 smokes (n=3) ---- + +[cells.archon-gaia-qwen27b-opus-ensemble-K3-3] +method = "archon" +bench = "gaia" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 3, max_tokens = 1024, temperature = 0.7 } + +# Cheap sanity smoke: local-only generator, zero cloud spend. Use to verify +# vLLM + Archon plumbing without paying for Opus. +[cells.archon-gaia-qwen27b-singlelocal-3] +method = "archon" +bench = "gaia" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { architecture = "single_local", max_tokens = 1024 } + +[cells.archon-swebenchverified-qwen27b-opus-ensemble-K3-3] +method = "archon" +bench = "swebench-verified" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 3, max_tokens = 1024, temperature = 0.7 } +concurrency = 3 + + +# ---- Stage 1.5 mid (n=30) ---- + +[cells.archon-gaia-qwen27b-opus-ensemble-K5-30] +method = "archon" +bench = "gaia" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 } +concurrency = 4 + +[cells.archon-gaia-qwen27b-singlelocal-30] +method = "archon" +bench = "gaia" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { architecture = "single_local", max_tokens = 2048 } +concurrency = 4 + +[cells.archon-gaia-gemma31b-opus-ensemble-K5-30] +method = "archon" +bench = "gaia" +n = 30 +local = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 } + +[cells.archon-swebenchverified-qwen27b-opus-ensemble-K5-30] +method = "archon" +bench = "swebench-verified" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 } +concurrency = 3 + + +# ---- Stage 2 full ---- + +[cells.archon-gaia-qwen27b-opus-ensemble-K5-165] +method = "archon" +bench = "gaia" +n = 165 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 } + +[cells.archon-gaia-gemma31b-opus-ensemble-K5-165] +method = "archon" +bench = "gaia" +n = 165 +local = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 } diff --git a/src/openjarvis/agents/hybrid/registry/conductor.toml b/src/openjarvis/agents/hybrid/registry/conductor.toml new file mode 100644 index 00000000..6e7eda06 --- /dev/null +++ b/src/openjarvis/agents/hybrid/registry/conductor.toml @@ -0,0 +1,54 @@ +# Conductor cells. Inference-only repro of arXiv 2512.04388 (Sakana, 2025). +# +# Stage-1 substitutes the (untrained) Qwen2.5-7B conductor with a zero-shot +# frontier planner (Opus). Workers default to local Qwen3.5-27B + Opus + +# gpt-5-mini; the adapter auto-drops the local worker if vLLM is down. +# Override `method_cfg.workers` to customize the pool. +# +# Naming: conductor--- + +[cells.conductor-gaia-opusplan-5] +method = "conductor" +bench = "gaia" +n = 5 +# vLLM is the default local worker; auto-skipped if /v1/models is unreachable. +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { worker_max_tokens = 4096, worker_temperature = 0.2, conductor_max_tokens = 2048 } + +[cells.conductor-gaia-gpt5miniplan-5] +method = "conductor" +bench = "gaia" +n = 5 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "gpt-5-mini", endpoint = "openai" } +method_cfg = { worker_max_tokens = 4096, worker_temperature = 0.2, conductor_max_tokens = 2048 } + +[cells.conductor-swebenchverified-opusplan-5] +method = "conductor" +bench = "swebench-verified" +n = 5 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { worker_max_tokens = 8192, worker_temperature = 0.2, conductor_max_tokens = 4096 } + + +# ---- n=30 (scaled-up cells for paradigms that worked at smoke tier) ---- + +[cells.conductor-gaia-opusplan-30] +method = "conductor" +bench = "gaia" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { worker_max_tokens = 4096, worker_temperature = 0.2, conductor_max_tokens = 2048 } +concurrency = 8 + +[cells.conductor-swebenchverified-opusplan-30] +method = "conductor" +bench = "swebench-verified" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { worker_max_tokens = 8192, worker_temperature = 0.2, conductor_max_tokens = 4096 } +concurrency = 8 diff --git a/src/openjarvis/agents/hybrid/registry/minions.toml b/src/openjarvis/agents/hybrid/registry/minions.toml new file mode 100644 index 00000000..bf3eeff4 --- /dev/null +++ b/src/openjarvis/agents/hybrid/registry/minions.toml @@ -0,0 +1,96 @@ +# Minions cells. One [cells.] block per (bench × local × cloud × N). +# +# Naming: minions---- +# HF model IDs verified Feb 2026 / Gemma 4 release: +# Qwen/Qwen3.5-27B (dense, 262K native context) +# google/gemma-4-31B-it (dense instruct, 256K context — 26B is the MoE variant) + +# ---- Stage 1 smokes (n=3) ---- + +[cells.minions-gaia-qwen27b-opus-3] +method = "minions" +bench = "gaia" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 3, worker_max_tokens = 4096 } + +[cells.minions-swebenchverified-qwen27b-opus-3] +method = "minions" +bench = "swebench-verified" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 3, worker_max_tokens = 8192 } + + +# ---- Stage 1.5 mid (n=30) — catches cells that smoke-pass but full-fail ---- + +[cells.minions-gaia-qwen27b-opus-30] +method = "minions" +bench = "gaia" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 4096 } + +[cells.minions-gaia-gemma31b-opus-30] +method = "minions" +bench = "gaia" +n = 30 +local = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 4096 } + +[cells.minions-swebenchverified-qwen27b-opus-30] +method = "minions" +bench = "swebench-verified" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 8192 } + +[cells.minions-swebenchverified-gemma31b-opus-30] +method = "minions" +bench = "swebench-verified" +n = 30 +local = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 8192 } + + +# ---- Stage 2 full ---- + +[cells.minions-gaia-qwen27b-opus-165] +method = "minions" +bench = "gaia" +n = 165 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 4096 } +concurrency = 4 + +[cells.minions-gaia-gemma31b-opus-165] +method = "minions" +bench = "gaia" +n = 165 +local = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 4096 } + +[cells.minions-swebenchverified-qwen27b-opus-500] +method = "minions" +bench = "swebench-verified" +n = 500 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 8192 } +concurrency = 4 + +[cells.minions-swebenchverified-gemma31b-opus-500] +method = "minions" +bench = "swebench-verified" +n = 500 +local = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 8192 } diff --git a/src/openjarvis/agents/hybrid/registry/skillorchestra.toml b/src/openjarvis/agents/hybrid/registry/skillorchestra.toml new file mode 100644 index 00000000..11db4a37 --- /dev/null +++ b/src/openjarvis/agents/hybrid/registry/skillorchestra.toml @@ -0,0 +1,47 @@ +# SkillOrchestra cells (arXiv:2602.19672). +# +# Deployment-time skill-aware routing only. The paper's offline pipeline +# (explore → learn → select over an SGLang model pool + FAISS wiki index) +# isn't reproduced here — we don't have the SGLang serving stack, the +# retriever index, or a training split to learn a routing policy on. What +# we DO reproduce is the inference-time orchestrator: an Opus router +# analyzes the question into skill weights, picks one of two agents +# (local Qwen-27B vs cloud Opus) under an explicit cost trade-off, then +# the chosen agent answers. Mirrors the eval_orchestrator paradigm in +# external/SkillOrchestra/skillorchestra/prompts/eval_orchestrator.py. +# See adapters/skillorchestra_adapter.py for full notes. + +[cells.skillorchestra-gaia-qwen27b-opus-3] +method = "skillorchestra" +bench = "gaia" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = {} + +[cells.skillorchestra-swebenchverified-qwen27b-opus-3] +method = "skillorchestra" +bench = "swebench-verified" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = {} +concurrency = 3 + +[cells.skillorchestra-gaia-qwen27b-opus-30] +method = "skillorchestra" +bench = "gaia" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = {} +concurrency = 4 + +[cells.skillorchestra-swebenchverified-qwen27b-opus-30] +method = "skillorchestra" +bench = "swebench-verified" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = {} +concurrency = 3 diff --git a/src/openjarvis/agents/hybrid/registry/toolorchestra.toml b/src/openjarvis/agents/hybrid/registry/toolorchestra.toml new file mode 100644 index 00000000..c6893725 --- /dev/null +++ b/src/openjarvis/agents/hybrid/registry/toolorchestra.toml @@ -0,0 +1,30 @@ +# ToolOrchestra cells. Prompted port — uses a cloud model (Opus) as the +# orchestrator, NOT the RL-trained Nemotron-Orchestrator-8B from the paper +# (arXiv:2511.21689). Treat results as preliminary until a real +# Orchestrator-8B deployment is wired up. +# +# Naming: toolorchestra---- + +[cells.toolorchestra-gaia-qwen27b-opus-3] +method = "toolorchestra" +bench = "gaia" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { max_turns = 6, orchestrator_max_tokens = 1024, worker_max_tokens = 4096 } + +[cells.toolorchestra-swebenchverified-qwen27b-opus-3] +method = "toolorchestra" +bench = "swebench-verified" +n = 3 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { max_turns = 6, orchestrator_max_tokens = 1024, worker_max_tokens = 8192 } + +[cells.toolorchestra-gaia-qwen27b-opus-30] +method = "toolorchestra" +bench = "gaia" +n = 30 +local = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" } +cloud = { model = "claude-opus-4-7", endpoint = "anthropic" } +method_cfg = { max_turns = 8, orchestrator_max_tokens = 1024, worker_max_tokens = 4096 } diff --git a/src/openjarvis/agents/hybrid/runner.py b/src/openjarvis/agents/hybrid/runner.py new file mode 100644 index 00000000..03654c71 --- /dev/null +++ b/src/openjarvis/agents/hybrid/runner.py @@ -0,0 +1,495 @@ +"""CLI runner for hybrid paradigm experiments. + +:: + + python -m openjarvis.agents.hybrid.runner --cell minions-gaia-qwen27b-opus-3 + +Reads a cell definition from ``registry/.toml`` (bundled with this +package or pointed at by ``OPENJARVIS_HYBRID_REGISTRY_DIR``), constructs +the registered agent, loads bench tasks via OpenJarvis's existing dataset +providers, runs every task, scores it, and writes +``//results.jsonl`` + ``summary.json``. + +The output schema matches ``hybrid-local-cloud-compute/runner.py`` so the +existing rescore / dashboard scripts can read OpenJarvis cells without +modification. +""" + +from __future__ import annotations + +import argparse +import fcntl +import json +import os +import sys +import threading +import time +import traceback +from concurrent.futures import ThreadPoolExecutor, as_completed +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional, Tuple + +try: + import tomllib # py3.11+ +except ModuleNotFoundError: + import tomli as tomllib # type: ignore[import-not-found,no-redef] + +from openjarvis.agents._stubs import AgentContext, AgentResult +from openjarvis.agents.hybrid._prompts import format_prompt as _format_prompt + + +PACKAGE_DIR = Path(__file__).parent +DEFAULT_REGISTRY_DIR = PACKAGE_DIR / "registry" +DEFAULT_EXPERIMENTS_DIR = Path( + os.environ.get( + "OPENJARVIS_HYBRID_EXPERIMENTS_DIR", + str(Path.home() / ".openjarvis-hybrid" / "experiments"), + ) +) + + +# ---------- Registry ---------- + +def load_registry(registry_dir: Optional[Path] = None) -> Dict[str, Dict[str, Any]]: + """Merge every ``/*.toml``. Cell names must be unique.""" + base = registry_dir or DEFAULT_REGISTRY_DIR + env_override = os.environ.get("OPENJARVIS_HYBRID_REGISTRY_DIR") + if env_override: + base = Path(env_override) + if not base.is_dir(): + return {} + cells: Dict[str, Dict[str, Any]] = {} + for p in sorted(base.glob("*.toml")): + data = tomllib.loads(p.read_text()) + for name, cell in data.get("cells", {}).items(): + if name in cells: + raise ValueError( + f"duplicate cell {name!r} (already defined before {p.name})" + ) + cells[name] = cell + return cells + + +# ---------- Bench dispatch ---------- + +def _load_gaia_tasks(n: Optional[int]) -> List[Dict[str, Any]]: + """GAIA validation. Each task is a dict with `task_id` + `question`.""" + from openjarvis.evals.datasets.gaia import GAIADataset + + ds = GAIADataset() + ds.load(max_samples=n) + out: List[Dict[str, Any]] = [] + for rec in ds.iter_records(): + # rec.problem is the formatted question prompt; rec.metadata carries + # the GAIA-specific fields including any reference answer. + out.append({ + "task_id": rec.record_id, + "question": rec.metadata.get("question", rec.problem), + "reference": rec.reference, + "metadata": dict(rec.metadata), + }) + return out + + +def _load_swebench_tasks(n: Optional[int]) -> List[Dict[str, Any]]: + """SWE-bench-Verified test. Each task carries patch-evaluation fields.""" + from openjarvis.evals.datasets.swebench import SWEBenchDataset + + ds = SWEBenchDataset() + ds.load(max_samples=n) + out: List[Dict[str, Any]] = [] + for rec in ds.iter_records(): + md = rec.metadata or {} + out.append({ + "task_id": md.get("instance_id", rec.record_id), + "repo": md.get("repo", ""), + "base_commit": md.get("base_commit", ""), + "problem_statement": md.get("problem_statement", rec.problem), + "hints_text": md.get("hints_text", ""), + "test_patch": md.get("test_patch", ""), + "FAIL_TO_PASS": md.get("FAIL_TO_PASS", []), + "PASS_TO_PASS": md.get("PASS_TO_PASS", []), + "version": md.get("version"), + "reference": rec.reference, + "metadata": dict(md), + }) + return out + + +def load_tasks(bench: str, n: Optional[int]) -> List[Dict[str, Any]]: + if bench == "gaia": + return _load_gaia_tasks(n) + if bench in ("swebench-verified", "swebench_verified", "swebench"): + return _load_swebench_tasks(n) + raise ValueError(f"unknown bench: {bench!r}") + + +# ---------- Scoring ---------- + +def _score_gaia(task: Dict[str, Any], answer: str) -> Dict[str, Any]: + """Exact-match-with-format-normalization GAIA scorer. + + Lightweight version: extracts the final-answer line and string-compares + against the reference. Use the OpenJarvis gaia_exact scorer for the + judge-tiebreaker path. + """ + import re + + ref = (task.get("reference") or "").strip() + if not ref: + return {"success": False, "score": 0.0, "details": {"reason": "no_reference"}} + m = re.search( + r"FINAL\s*ANSWER\s*:\s*(.+?)\s*$", + answer, + re.IGNORECASE | re.MULTILINE, + ) + pred = (m.group(1).strip() if m else answer.strip()).rstrip(".").strip() + success = pred.lower() == ref.lower() + return { + "success": success, + "score": 1.0 if success else 0.0, + "details": {"prediction": pred, "reference": ref}, + } + + +def _score_swebench(task: Dict[str, Any], answer: str) -> Dict[str, Any]: + """Modal-backed SWE-bench Verified harness scorer.""" + from openjarvis.evals.scorers.swebench_harness import ( + SWEBenchHarnessScorer, + extract_patch, + ) + from openjarvis.evals.core.types import EvalRecord + + patch = extract_patch(answer) + if patch is None: + return {"success": False, "score": 0.0, "details": {"reason": "no_patch_extracted"}} + + record = EvalRecord( + record_id=task["task_id"], + problem=task.get("problem_statement", ""), + reference="", + category="agentic", + metadata={"instance_id": task["task_id"]}, + ) + scorer = SWEBenchHarnessScorer(timeout_s=int(os.environ.get("SWEBENCH_TIMEOUT_S", "1800"))) + is_correct, details = scorer.score(record, answer) + return { + "success": bool(is_correct), + "score": 1.0 if is_correct else 0.0, + "details": details, + } + + +def score(bench: str, task: Dict[str, Any], answer: str) -> Dict[str, Any]: + if bench == "gaia": + return _score_gaia(task, answer) + if bench in ("swebench-verified", "swebench_verified", "swebench"): + return _score_swebench(task, answer) + raise ValueError(f"unknown bench: {bench!r}") + + +# ---------- Cell run ---------- + +def _cell_dir(cell_name: str, root: Path) -> Path: + d = root / cell_name + d.mkdir(parents=True, exist_ok=True) + (d / "logs").mkdir(exist_ok=True) + return d + + +@contextmanager +def _cell_lock(out_dir: Path, cell_name: str): + """Exclusive flock on ``/.lock`` to prevent concurrent runner stomps.""" + lock_path = out_dir / ".lock" + f = lock_path.open("a+") + try: + fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + f.seek(0) + prev = (f.read() or "?").strip() or "?" + f.close() + raise SystemExit( + f"[lock] another runner is already running cell {cell_name!r} " + f"(holder pid: {prev}). refusing to start a second instance." + ) + f.seek(0); f.truncate(); f.write(str(os.getpid())); f.flush() + try: + yield + finally: + try: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + except Exception: + pass + f.close() + try: + lock_path.unlink() + except FileNotFoundError: + pass + + +def _build_agent(cell: Dict[str, Any]): + """Construct the registered agent for this cell.""" + import openjarvis.agents # noqa: F401 — populate registry + from openjarvis.core.registry import AgentRegistry + + method = cell["method"] + if not AgentRegistry.contains(method): + raise ValueError( + f"agent {method!r} not registered. Available: " + f"{', '.join(sorted(AgentRegistry.keys()))}" + ) + agent_cls = AgentRegistry.get(method) + local = cell.get("local") or {} + cloud = cell.get("cloud") or {} + method_cfg = dict(cell.get("method_cfg") or {}) + + return agent_cls( + engine=None, # raw SDK calls — engine unused + model=cloud.get("model", ""), + local_model=local.get("model"), + local_endpoint=local.get("endpoint"), + cloud_endpoint=(cloud.get("endpoint") or "anthropic").lower(), + cfg=method_cfg, + ) + + +def _run_one(agent, bench: str, task: Dict[str, Any], log_dir: str) -> Dict[str, Any]: + """Run the agent on one task. Returns a hybrid-shape row.""" + prompt = _format_prompt(task) + ctx = AgentContext(metadata={"task": task, "task_id": task["task_id"]}) + t0 = time.time() + try: + result: AgentResult = agent.run(prompt, ctx) + meta = dict(result.metadata or {}) + out = { + "task_id": task["task_id"], + "answer": result.content or "", + "tokens_local": int(meta.get("tokens_local", 0)), + "tokens_cloud": int(meta.get("tokens_cloud", 0)), + "cost_usd": float(meta.get("cost_usd", 0.0)), + "latency_s": float(meta.get("latency_s", time.time() - t0)), + "traces": meta.get("traces", {}), + } + if "soft_error" in meta: + out["soft_error"] = meta["soft_error"] + return {**out, "error": None} + except Exception as e: + return { + "task_id": task["task_id"], + "answer": "", + "tokens_local": 0, "tokens_cloud": 0, + "cost_usd": 0.0, "latency_s": time.time() - t0, + "traces": {}, + "error": f"{type(e).__name__}: {e}\n{traceback.format_exc()}", + } + + +def _heartbeat(done: int, total: int, row: Dict[str, Any], t_start: float) -> None: + elapsed = time.time() - t_start + eta = (total - done) * (elapsed / max(done, 1)) + ok = "OK" if not row.get("error") else "ERR" + s = row.get("score") or {} + sc = s.get("score") + sc_str = f"{sc:.2f}" if isinstance(sc, (int, float)) else "—" + print( + f"[{done}/{total}] {ok} task={row['task_id']} score={sc_str} " + f"local={row['tokens_local']} cloud={row['tokens_cloud']} " + f"${row['cost_usd']:.3f} {row['latency_s']:.1f}s eta={eta/60:.1f}m", + flush=True, + ) + + +def _write_summary( + out_dir: Path, + cell_name: str, + cell: Dict[str, Any], + tasks: List[Dict[str, Any]], + t_start: float, +) -> None: + results_path = out_dir / "results.jsonl" + rows = [ + json.loads(l) + for l in results_path.read_text().splitlines() + if l.strip() + ] + n_done = len(rows) + n_err = sum(1 for r in rows if r.get("error")) + successes = [r for r in rows if r.get("score") and r["score"].get("success")] + acc = (len(successes) / n_done) if n_done else 0.0 + total_cost = sum(r.get("cost_usd", 0.0) for r in rows) + total_local = sum(r.get("tokens_local", 0) for r in rows) + total_cloud = sum(r.get("tokens_cloud", 0) for r in rows) + elapsed = time.time() - t_start + + summary = { + "cell": cell_name, + "method": cell["method"], + "bench": cell["bench"], + "n_target": cell["n"], + "n_done": n_done, + "n_err": n_err, + "accuracy": acc, + "tokens_local_total": total_local, + "tokens_cloud_total": total_cloud, + "cost_usd_total": total_cost, + "wall_time_s": elapsed, + "task_count": len(tasks), + } + (out_dir / "summary.json").write_text(json.dumps(summary, indent=2)) + print( + f"[summary] {cell_name}: n={n_done}/{cell['n']} err={n_err} " + f"acc={acc:.3f} cost=${total_cost:.2f} time={elapsed/60:.1f}m", + flush=True, + ) + + +def run_cell( + cell_name: str, + cell: Dict[str, Any], + *, + do_score: bool = True, + resume: bool = True, + root: Optional[Path] = None, +) -> None: + out_root = root or DEFAULT_EXPERIMENTS_DIR + out_dir = _cell_dir(cell_name, out_root) + with _cell_lock(out_dir, cell_name): + _run_cell_locked( + cell_name, cell, out_dir, + do_score=do_score, resume=resume, + ) + + +def _run_cell_locked( + cell_name: str, + cell: Dict[str, Any], + out_dir: Path, + *, + do_score: bool, + resume: bool, +) -> None: + (out_dir / "config.json").write_text( + json.dumps({"name": cell_name, **cell}, indent=2) + ) + + results_path = out_dir / "results.jsonl" + done_ids: set = set() + if resume and results_path.exists(): + # Keep only successful rows; drop errored rows so they retry. + kept: List[str] = [] + for line in results_path.read_text().splitlines(): + try: + row = json.loads(line) + except Exception: + continue + if not row.get("error"): + kept.append(line) + done_ids.add(row["task_id"]) + results_path.write_text("\n".join(kept) + ("\n" if kept else "")) + print( + f"[resume] {len(done_ids)} tasks already done (errored rows dropped)", + flush=True, + ) + + tasks = load_tasks(cell["bench"], n=cell["n"]) + print(f"[load] {cell['bench']} → {len(tasks)} tasks", flush=True) + + pending = [t for t in tasks if t["task_id"] not in done_ids] + concurrency = max(1, int(cell.get("concurrency", 1))) + if concurrency > 1: + print(f"[concurrency] {concurrency} workers", flush=True) + + agent = _build_agent(cell) + + t_start = time.time() + write_lock = threading.Lock() + completed = [0] + log_dir = str(out_dir / "logs") + + def _process(task: Dict[str, Any]) -> None: + row = _run_one(agent, cell["bench"], task, log_dir) + scored: Optional[Dict[str, Any]] = None + if do_score and row.get("error") is None: + try: + scored = score(cell["bench"], task, row["answer"]) + except Exception as e: + scored = { + "success": False, "score": 0.0, + "details": {"score_error": str(e)}, + } + full_row = {**row, "score": scored} + with write_lock, results_path.open("a") as f: + f.write(json.dumps(full_row) + "\n") + f.flush() + completed[0] += 1 + _heartbeat(completed[0], len(tasks), full_row, t_start) + + if concurrency == 1: + for task in pending: + _process(task) + else: + with ThreadPoolExecutor(max_workers=concurrency) as ex: + futures = [ex.submit(_process, t) for t in pending] + for fut in as_completed(futures): + fut.result() + + _write_summary(out_dir, cell_name, cell, tasks, t_start) + + +# ---------- CLI ---------- + +def main(argv: Optional[List[str]] = None) -> int: + p = argparse.ArgumentParser( + prog="python -m openjarvis.agents.hybrid.runner", + description="Run a hybrid paradigm experiment cell.", + ) + p.add_argument("--cell", required=True, help="Cell name from the registry TOMLs.") + p.add_argument( + "--registry-dir", + default=None, + help="Override registry dir (defaults to package registry/).", + ) + p.add_argument( + "--root", + default=None, + help="Override experiments output root.", + ) + p.add_argument("--no-score", action="store_true", help="Skip scoring.") + p.add_argument("--no-resume", action="store_true", help="Don't resume from results.jsonl.") + args = p.parse_args(argv) + + reg_dir = Path(args.registry_dir) if args.registry_dir else None + cells = load_registry(reg_dir) + if not cells: + print(f"[error] no cells found in {reg_dir or DEFAULT_REGISTRY_DIR}", file=sys.stderr) + return 2 + if args.cell not in cells: + print( + f"[error] unknown cell {args.cell!r}. Known: {', '.join(sorted(cells))}", + file=sys.stderr, + ) + return 2 + root = Path(args.root) if args.root else None + run_cell( + args.cell, cells[args.cell], + do_score=not args.no_score, + resume=not args.no_resume, + root=root, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + + +__all__ = [ + "DEFAULT_EXPERIMENTS_DIR", + "DEFAULT_REGISTRY_DIR", + "load_registry", + "load_tasks", + "main", + "run_cell", + "score", +] diff --git a/src/openjarvis/agents/hybrid/scripts/new_experiment.sh b/src/openjarvis/agents/hybrid/scripts/new_experiment.sh new file mode 100755 index 00000000..4d89ccd6 --- /dev/null +++ b/src/openjarvis/agents/hybrid/scripts/new_experiment.sh @@ -0,0 +1,108 @@ +#!/usr/bin/env bash +# Scaffold a new hybrid paradigm experiment cell. +# +# Appends a [cells.] block to +# src/openjarvis/agents/hybrid/registry/.toml +# (registry is split by method — minions.toml, conductor.toml, etc.). +# +# Usage: +# src/openjarvis/agents/hybrid/scripts/new_experiment.sh \ +# --method minions --bench gaia \ +# --local qwen3.5-27b --cloud claude-opus-4-7 --n 50 \ +# [--mode minion|minions] [--max-rounds 3] +# +# Then run: +# python -m openjarvis.agents.hybrid.runner --cell + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +REGISTRY_DIR="$(cd "$SCRIPT_DIR/.." && pwd)/registry" + +method=""; bench=""; local_model=""; cloud_model=""; n="" +mode="minion"; max_rounds=3 + +while [[ $# -gt 0 ]]; do + case "$1" in + --method) method="$2"; shift 2;; + --bench) bench="$2"; shift 2;; + --local) local_model="$2"; shift 2;; + --cloud) cloud_model="$2"; shift 2;; + --n) n="$2"; shift 2;; + --mode) mode="$2"; shift 2;; + --max-rounds) max_rounds="$2"; shift 2;; + *) echo "unknown arg: $1" >&2; exit 2;; + esac +done + +for v in method bench local_model cloud_model n; do + if [[ -z "${!v}" ]]; then + echo "missing required --${v//_/-}" >&2 + exit 2 + fi +done + +# Map shortnames → full HF id + endpoint +case "$local_model" in + qwen3.5-27b|qwen-27b|qwen27b) local_full="Qwen/Qwen3.5-27B-FP8"; local_ep="http://localhost:8001/v1";; + qwen3.5-9b|qwen-9b|qwen9b) local_full="Qwen/Qwen3.5-9B"; local_ep="http://localhost:8001/v1";; + gemma-4-31b|gemma-31b|gemma31b) local_full="google/gemma-4-31B-it"; local_ep="http://localhost:8004/v1";; + gemma-4-26b|gemma-26b|gemma26b) local_full="google/gemma-4-26B-A4B-it"; local_ep="http://localhost:8004/v1";; + *) local_full="$local_model"; local_ep="${LOCAL_ENDPOINT:-http://localhost:8001/v1}";; +esac + +case "$cloud_model" in + gpt-5|gpt-5-mini|gpt-4o) cloud_ep="openai";; + claude-opus-4-7|claude-sonnet-4-6) cloud_ep="anthropic";; + *) cloud_ep="${CLOUD_ENDPOINT:-anthropic}";; +esac + +local_short="$(echo "$local_model" | tr '/' '_')" +cloud_short="$cloud_model" +name="${method}-${bench}-${local_short}-${cloud_short}-${n}" + +registry_file="$REGISTRY_DIR/${method}.toml" +if [[ ! -f "$registry_file" ]]; then + echo "no registry file for method '${method}' — expected $registry_file" >&2 + echo "known: $(ls "$REGISTRY_DIR" | tr '\n' ' ')" >&2 + exit 2 +fi + +# Method-specific method_cfg defaults — keep simple, override later in TOML. +case "$method" in + minions) + cfg_block="method_cfg = { mode = \"${mode}\", max_rounds = ${max_rounds}, worker_max_tokens = 4096 }" + ;; + conductor) + cfg_block="method_cfg = { conductor_max_tokens = 2048, worker_max_tokens = 4096 }" + ;; + advisors) + cfg_block="method_cfg = { executor_max_tokens = 4096, advisor_max_tokens = 1024 }" + ;; + archon) + cfg_block="method_cfg = { architecture = \"ensemble_rank_fuse\", n_samples = 5, max_tokens = 2048 }" + ;; + skillorchestra) + cfg_block="method_cfg = { router_max_tokens = 1024, local_max_tokens = 4096, cloud_max_tokens = 4096 }" + ;; + toolorchestra) + cfg_block="method_cfg = { max_turns = 6, orchestrator_max_tokens = 1024, worker_max_tokens = 4096 }" + ;; + *) + cfg_block="method_cfg = {}" + ;; +esac + +cat >> "$registry_file" <