admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-27 21:05:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: wire native web_search into GAIA cells across paradigms

Browse Source 
Adds opt-in `method_cfg.web_search = { enabled, max_uses }` schema and
plumbs the native Anthropic server-side `web_search_20250305` tool
through every hybrid paradigm's GAIA codepath. Default OFF preserves
back-compat with all currently running n=100 cells; only the next
rendered sweep opts in.

Engine layer: `_base.py` gets `build_web_search_tool`, `web_search_cfg`,
and `_call_anthropic_agent` (multi-turn loop with `gaia_max_turns`
default 8). Anthropic cost (`$0.01 / search`) added to `tokens_cloud`.

Per-paradigm wiring (cloud-side Anthropic calls only):
  * baseline_cloud: GAIA one-shot upgraded to agent loop with tools
  * advisors: executor1 + executor2 passes declare tools
  * minions: existing prefetch retained as legacy default; new schema
    overrides
  * skillorchestra: cloud-route specialist gets tools
  * conductor: anthropic worker steps get tools when enabled
  * toolorchestra: already used native web_search worker - unchanged,
    now surfaces web_search_uses in meta
  * archon: thread-local web_search tool injection on Anthropic
    ranker/fuser generators

Trace + aggregation:
  * Per-row `web_search_uses: int` in results.jsonl
  * Summary `web_search_uses_total` in summary.json
  * make_report.py adds `web_searches_mean` column to the HTML table

Tests: tests/agents/hybrid/test_web_search_wiring.py - confirms the
tool is declared when enabled, NOT declared when omitted (default),
and gracefully skipped on non-Anthropic endpoints (no fake local
web_search).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Andrew Park
2026-05-17 17:55:18 -07:00
co-authored by Claude Opus 4.7
parent 3b4b3963ef
commit 24da67d24d
10 changed files with 1653 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/ablation/make_report.py
+554
View File
@@ -0,0 +1,554 @@
#!/usr/bin/env python3
"""Build /matx/u/aspark/.openjarvis/experiments/hybrid/docs/index.html
plus a set of PNG pareto plots in the same dir. The HTML embeds the PNGs
via <img src="..."> so the browser never has to render anything itself —
just shows static images.
Re-running is idempotent: regenerates index.html + all PNGs from whatever
cells have summary.json at run-time.
"""
from __future__ import annotations
import html as html_lib
import json
import sys
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from statistics import median
from typing import Callable, Optional
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
HYBRID_ROOT = Path("/matx/u/aspark/.openjarvis/experiments/hybrid")
DOCS_DIR = HYBRID_ROOT / "docs"
OUT_HTML = DOCS_DIR / "index.html"
RAW_MD = DOCS_DIR / "results-table.md"
PLOTS_DIR = DOCS_DIR / "plots-n100"
PLOTS_DIR.mkdir(exist_ok=True)
CLOUD_TOKENS = {
"opus47": "Opus 4.7",
"haiku45": "Haiku 4.5",
"gpt5mini": "GPT-5 mini",
"gpt5": "GPT-5.5",
"gemini25pro": "Gemini 3.1 Pro",
"gemini25flash": "Gemini 3.1 Flash",
}
CLOUD_TOKEN_ORDER = [
"gpt5mini", "gemini25flash", "gemini25pro", "gpt5", "opus47", "haiku45",
]
BENCH_LABELS = {"gaia": "GAIA", "swe": "SWE-bench"}
PARADIGM_COLOR = {
"skillorchestra": "#3b82f6",
"cloud-only": "#9ca3af",
"minions": "#10b981",
"advisors": "#f59e0b",
}
PARADIGM_ORDER = ["cloud-only", "skillorchestra", "minions", "advisors"]
def parse_cell_name(name: str) -> Optional[dict]:
if not name.endswith("-n100"):
return None
parts = name[:-len("-n100")].split("-")
if not parts:
return None
paradigm = parts[0]
rest = parts[1:]
if paradigm == "cloud":
if rest and rest[0] == "only":
rest = rest[1:]
paradigm = "cloud-only"
local = None
elif paradigm in ("skillorchestra", "minions", "advisors"):
if not rest:
return None
local = rest[0]
rest = rest[1:]
else:
return None
if len(rest) < 2:
return None
cloud_token = None
for tok in CLOUD_TOKEN_ORDER:
if tok in rest:
cloud_token = tok
break
if cloud_token is None:
return None
bench = rest[-1]
if bench not in BENCH_LABELS:
return None
return {"paradigm": paradigm, "local": local, "cloud": cloud_token, "bench": bench}
@dataclass
class Cell:
name: str
paradigm: str
local: Optional[str]
cloud: str
bench: str
accuracy: float
cost_usd: float
tokens_local_total: int
tokens_cloud_total: int
n_done: int
latency_med_s: Optional[float]
tool_calls_mean: Optional[float]
web_searches_mean: Optional[float]
def load_cells() -> list[Cell]:
out = []
for d in sorted(HYBRID_ROOT.iterdir()):
if not (d.is_dir() and d.name.endswith("-n100")):
continue
sj = d / "summary.json"
if not sj.exists():
continue
parsed = parse_cell_name(d.name)
if parsed is None:
continue
try:
s = json.loads(sj.read_text())
except Exception:
continue
# latency median
rj = d / "results.jsonl"
lat_med = None
if rj.exists():
lats = []
for line in rj.read_text().splitlines():
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except Exception:
continue
v = row.get("latency_s")
if isinstance(v, (int, float)):
lats.append(float(v))
if lats:
lat_med = median(lats)
# tool calls mean — count any flavor of tool use per task:
# * SWE / mini-swe-agent fires events with kind="<something>_bash" (one per bash exec)
# * GAIA / agentic flows record tool calls inline on anthropic/openai events
# as `tool_calls: [...]` (anthropic server_tool_use, openai function calls)
# and Anthropic-side web search uses `n_web_searches`
# * minions does a hidden pre-fetch step recorded under metadata.traces.prefetch.n_searches
# Sum all of them so the metric isn't silently zero on GAIA.
#
# `web_searches_mean` is the same idea but ONLY counts Anthropic
# server-side web_search invocations — useful to see how often
# GAIA cells actually leveraged the new opt-in web_search tool.
tc_mean = None
ws_mean = None
logs = d / "logs"
if logs.is_dir():
counts = []
ws_counts = []
for log_p in logs.iterdir():
if not log_p.name.endswith(".json"):
continue
try:
lg = json.loads(log_p.read_text())
except Exception:
continue
ev = lg.get("events") or []
bash_c = sum(1 for e in ev if isinstance(e, dict)
and isinstance(e.get("kind"), str)
and "_bash" in e["kind"])
tc_c = sum(len(e.get("tool_calls") or [])
for e in ev if isinstance(e, dict))
ws_c = sum(int(e.get("n_web_searches") or 0)
for e in ev if isinstance(e, dict))
# n_web_searches is Anthropic's server-side web_search count; it's
# already represented as one tool_call per search above, so don't
# double-count — only fall back to it if tool_calls list is empty.
if tc_c == 0:
tc_c = ws_c
# prefetch (minions) — recorded out-of-band in metadata.traces.prefetch
meta = lg.get("metadata") or {}
pf = ((meta.get("traces") or {}).get("prefetch") or {}) if isinstance(meta, dict) else {}
pf_c = int(pf.get("n_searches") or 0) if isinstance(pf, dict) else 0
counts.append(bash_c + tc_c + pf_c)
ws_counts.append(ws_c + pf_c)
if counts:
tc_mean = sum(counts) / len(counts)
if ws_counts:
ws_mean = sum(ws_counts) / len(ws_counts)
out.append(Cell(
name=d.name,
paradigm=parsed["paradigm"],
local=parsed["local"],
cloud=parsed["cloud"],
bench=parsed["bench"],
accuracy=float(s.get("accuracy") or 0.0),
cost_usd=float(s.get("cost_usd_total") or 0.0),
tokens_local_total=int(s.get("tokens_local_total") or 0),
tokens_cloud_total=int(s.get("tokens_cloud_total") or 0),
n_done=int(s.get("n_done") or 0),
latency_med_s=lat_med,
tool_calls_mean=tc_mean,
web_searches_mean=ws_mean,
))
return out
@dataclass
class Axis:
key: str
title: str
description: str
filter_fn: Callable[[Cell], bool]
def axis_definitions() -> list[Axis]:
return [
Axis("all-cells", "0. All cells overview",
"Every cell on one plot. Color = paradigm. Pareto frontier at a glance.",
lambda c: True),
Axis("cloud-anthropic", "1. Cloud-size within Anthropic",
"Local = Qwen-27B; vary Anthropic cloud (Opus 4.7 vs Haiku 4.5). All paradigms overlaid.",
lambda c: c.cloud in ("opus47", "haiku45")),
Axis("cloud-openai", "2. Cloud-size within OpenAI",
"Local = Qwen-27B; vary OpenAI cloud (GPT-5.5 vs GPT-5 mini). All paradigms overlaid.",
lambda c: c.cloud in ("gpt5", "gpt5mini")),
Axis("cloud-google", "3. Cloud-size within Google",
"Local = Qwen-27B; vary Google cloud (Gemini 3.1 Pro vs Flash). All paradigms overlaid.",
lambda c: c.cloud in ("gemini25pro", "gemini25flash")),
Axis("cloud-family-frontier", "4. Cloud-family — frontier tier",
"Compare frontier clouds across vendors (Opus 4.7, GPT-5.5, Gemini 3.1 Pro).",
lambda c: c.cloud in ("opus47", "gpt5", "gemini25pro")),
Axis("cloud-family-mini", "5. Cloud-family — mini/flash tier",
"Compare cost-floor clouds across vendors (Haiku 4.5, GPT-5 mini, Gemini 3.1 Flash).",
lambda c: c.cloud in ("haiku45", "gpt5mini", "gemini25flash")),
Axis("paradigm-skillorch", "6. Skillorchestra only",
"Just the skillorchestra cells. See how its router behaves across cloud choices.",
lambda c: c.paradigm == "skillorchestra"),
Axis("paradigm-cloud-only", "7. Cloud-only baseline",
"Baseline cloud-only runs across all 6 clouds — no local model in the loop.",
lambda c: c.paradigm == "cloud-only"),
]
METRIC_SPECS = [
("cost_usd", "Cost (USD)", True),
("latency_med_s", "Latency median (s)", True),
("tokens_cloud_total", "Tokens cloud (total)", True),
]
def render_axis_png(axis: Axis, cells: list[Cell]) -> Optional[Path]:
pts = [c for c in cells if axis.filter_fn(c)]
if not pts:
return None
n_metrics = len(METRIC_SPECS)
fig, axes = plt.subplots(
n_metrics, 2,
figsize=(15, 4.2 * n_metrics),
squeeze=False,
)
fig.suptitle(axis.title, fontsize=16, fontweight="bold", y=0.998)
used_paradigms = sorted({c.paradigm for c in pts}, key=lambda p: PARADIGM_ORDER.index(p) if p in PARADIGM_ORDER else 99)
for row_idx, (mkey, mlabel, logx) in enumerate(METRIC_SPECS):
for col_idx, bench in enumerate(("gaia", "swe")):
ax = axes[row_idx][col_idx]
bench_pts = [c for c in pts if c.bench == bench]
for p in used_paradigms:
p_pts = [c for c in bench_pts if c.paradigm == p]
xs, ys, labels = [], [], []
for c in p_pts:
v = getattr(c, mkey)
if v is None:
continue
if logx and v <= 0:
continue
xs.append(v); ys.append(c.accuracy); labels.append(CLOUD_TOKENS.get(c.cloud, c.cloud))
if xs:
ax.scatter(xs, ys, c=PARADIGM_COLOR.get(p, "#000"),
s=120, alpha=0.85, edgecolors="white", linewidths=1.5,
label=p if (row_idx == 0 and col_idx == 0) else None)
for x, y, lbl in zip(xs, ys, labels):
ax.annotate(lbl, (x, y), xytext=(6, 6), textcoords="offset points",
fontsize=9, color="#1f2937", alpha=0.9)
if logx:
ax.set_xscale("log")
# Generous y-axis: 0 → max(observed)+0.15, floored at 0.7 so small numbers don't look cramped
ax.set_ylim(0, max(0.75, max((c.accuracy for c in bench_pts), default=0.5) + 0.15))
ax.set_xlabel(mlabel, fontsize=10)
ax.set_ylabel("accuracy" if col_idx == 0 else "")
ax.set_title(f"{mlabel} — {BENCH_LABELS[bench]}", fontsize=11, fontweight="bold")
ax.grid(True, color="#e5e7eb", linewidth=0.5)
ax.set_axisbelow(True)
for spine in ax.spines.values():
spine.set_color("#d1d5db")
# legend on top
handles, labels = axes[0][0].get_legend_handles_labels()
if handles:
fig.legend(handles, labels, loc="upper right", bbox_to_anchor=(0.99, 0.99),
ncol=len(used_paradigms), frameon=True, facecolor="white",
edgecolor="#d1d5db", fontsize=10)
plt.subplots_adjust(left=0.07, right=0.97, top=0.96, bottom=0.04,
hspace=0.55, wspace=0.18)
out = PLOTS_DIR / f"{axis.key}.png"
fig.savefig(out, dpi=110, bbox_inches="tight")
plt.close(fig)
return out
# ---------- Markdown → HTML (small) ----------
def md_to_html(md: str) -> str:
lines = md.splitlines()
out: list[str] = []
in_table = False
table_rows: list[list[str]] = []
def flush_table():
nonlocal table_rows
if not table_rows:
return
head = table_rows[0]
body = table_rows[2:] if len(table_rows) > 2 else []
out.append("<table class='md'><thead><tr>" +
"".join(f"<th>{html_lib.escape(h.strip())}</th>" for h in head) +
"</tr></thead><tbody>")
for row in body:
out.append("<tr>" + "".join(
f"<td>{html_lib.escape(cell.strip())}</td>" for cell in row) + "</tr>")
out.append("</tbody></table>")
table_rows = []
def inline(s: str) -> str:
s = html_lib.escape(s)
# bold
import re as _re
s = _re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", s)
s = _re.sub(r"`(.+?)`", r"<code>\1</code>", s)
return s
for raw in lines:
line = raw.rstrip()
if line.startswith("|") and "|" in line[1:]:
if not in_table:
in_table = True
table_rows = []
cells = [c for c in line.strip().strip("|").split("|")]
table_rows.append(cells)
continue
if in_table:
flush_table()
in_table = False
if line.startswith("### "):
out.append(f"<h3>{inline(line[4:])}</h3>")
elif line.startswith("## "):
out.append(f"<h2>{inline(line[3:])}</h2>")
elif line.startswith("# "):
out.append(f"<h1>{inline(line[2:])}</h1>")
elif line.startswith("> "):
out.append(f"<blockquote>{inline(line[2:])}</blockquote>")
elif line.startswith("- "):
out.append(f"<li>{inline(line[2:])}</li>")
elif line.strip() == "---":
out.append("<hr>")
elif line.strip() == "":
out.append("")
else:
out.append(f"<p>{inline(line)}</p>")
if in_table:
flush_table()
return "\n".join(out)
# ---------- HTML output ----------
CSS = """
body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif;
background: #fafbfc; color: #1a1a1a; max-width: 1280px; margin: 0 auto;
padding: 24px 16px 60px; line-height: 1.55; }
h1 { border-bottom: 2px solid #1a1a1a; padding-bottom: 8px; }
h2 { margin-top: 40px; border-bottom: 1px solid #d1d5db; padding-bottom: 6px; }
.card { background: white; border: 1px solid #e5e7eb; border-radius: 8px;
box-shadow: 0 1px 3px rgba(0,0,0,0.05); padding: 16px 20px; margin: 18px 0; }
.hero { display: grid; grid-template-columns: repeat(auto-fit, minmax(220px, 1fr));
gap: 12px; margin: 16px 0 24px; }
.hero .item { background: white; border: 1px solid #e5e7eb; border-left: 3px solid #3b82f6;
border-radius: 6px; padding: 12px 14px; }
.hero .item.cost { border-left-color: #10b981; }
.hero .label { font-size: 0.78em; color: #6b7280; text-transform: uppercase; letter-spacing: 0.5px; }
.hero .value { font-size: 1.15em; font-weight: 600; margin-top: 4px; }
.hero .sub { font-size: 0.85em; color: #4b5563; margin-top: 2px; }
table.summary { width: 100%; border-collapse: collapse; font-size: 0.85em; margin-top: 12px; }
table.summary th, table.summary td { border: 1px solid #e5e7eb; padding: 6px 8px; text-align: left; }
table.summary th { background: #f3f4f6; cursor: pointer; user-select: none; }
table.summary tr:nth-child(even) { background: #fafafa; }
table.summary .best { background: #dcfce7 !important; }
section { margin-top: 28px; }
section img { max-width: 100%; height: auto; display: block; margin: 8px 0;
border: 1px solid #e5e7eb; border-radius: 6px; background: white; }
.legend { font-size: 0.85em; color: #6b7280; }
.legend .sw { display: inline-block; width: 11px; height: 11px; border-radius: 50%;
vertical-align: middle; margin-right: 4px; }
.desc { color: #4b5563; }
table.md { width: 100%; border-collapse: collapse; font-size: 0.85em; margin: 12px 0; }
table.md th, table.md td { border: 1px solid #e5e7eb; padding: 4px 8px; text-align: left; }
table.md th { background: #f3f4f6; }
table.md tr:nth-child(even) { background: #fafafa; }
code { background: #f3f4f6; padding: 1px 4px; border-radius: 3px; font-size: 0.92em; }
hr { border: none; border-top: 1px solid #d1d5db; margin: 20px 0; }
blockquote { border-left: 3px solid #d1d5db; padding-left: 12px; color: #4b5563; margin: 8px 0; }
"""
SORT_JS = """
document.querySelectorAll('table.summary th').forEach((th, idx) => {
th.addEventListener('click', () => {
const tbl = th.closest('table');
const tbody = tbl.querySelector('tbody');
const rows = Array.from(tbody.querySelectorAll('tr'));
const asc = th.dataset.dir !== 'asc';
rows.sort((a, b) => {
const av = a.cells[idx].dataset.sort ?? a.cells[idx].innerText;
const bv = b.cells[idx].dataset.sort ?? b.cells[idx].innerText;
const af = parseFloat(av), bf = parseFloat(bv);
const an = isNaN(af) ? av : af, bn = isNaN(bf) ? bv : bf;
return (an < bn ? -1 : an > bn ? 1 : 0) * (asc ? 1 : -1);
});
rows.forEach(r => tbody.appendChild(r));
tbl.querySelectorAll('th').forEach(t => delete t.dataset.dir);
th.dataset.dir = asc ? 'asc' : 'desc';
});
});
"""
def build_html(cells: list[Cell], plots: list[tuple[Axis, Path]]) -> str:
ts = datetime.now().strftime("%Y-%m-%d %H:%M")
# headline cards
by_bench = {"gaia": [c for c in cells if c.bench == "gaia"],
"swe": [c for c in cells if c.bench == "swe"]}
def best_acc(lst):
return max(lst, key=lambda c: c.accuracy) if lst else None
def best_acc_per_dollar(lst):
cand = [c for c in lst if c.cost_usd > 0]
return max(cand, key=lambda c: c.accuracy / c.cost_usd) if cand else None
hero = ['<div class="hero">']
for bench_key, label in (("gaia", "GAIA"), ("swe", "SWE-bench")):
b = best_acc(by_bench[bench_key])
if b:
hero.append(f"""<div class="item"><div class="label">Best acc — {label}</div>
<div class="value">{b.accuracy:.3f} · {b.paradigm} × {CLOUD_TOKENS.get(b.cloud, b.cloud)}</div>
<div class="sub">cost: ${b.cost_usd:.2f}</div></div>""")
for bench_key, label in (("gaia", "GAIA"), ("swe", "SWE-bench")):
b = best_acc_per_dollar(by_bench[bench_key])
if b:
ratio = b.accuracy / b.cost_usd
hero.append(f"""<div class="item cost"><div class="label">Best acc / $ — {label}</div>
<div class="value">{b.accuracy:.3f} for ${b.cost_usd:.2f}</div>
<div class="sub">{b.paradigm} × {CLOUD_TOKENS.get(b.cloud, b.cloud)} ({ratio:.3f} acc/$)</div></div>""")
hero.append("</div>")
# summary table
legend_html = '<div class="legend">'
for p in PARADIGM_ORDER:
col = PARADIGM_COLOR.get(p, "#000")
legend_html += f'<span class="sw" style="background:{col}"></span>{p}&nbsp;&nbsp;'
legend_html += '</div>'
rows_html = []
sorted_cells = sorted(cells, key=lambda c: (c.bench, -c.accuracy))
# mark best per bench
best_per_bench = {b: best_acc(by_bench[b]) for b in by_bench}
for c in sorted_cells:
is_best = best_per_bench.get(c.bench) is c
cls = "best" if is_best else ""
rows_html.append(f"""<tr class="{cls}">
<td>{c.paradigm}</td>
<td>{c.local or '—'}</td>
<td>{CLOUD_TOKENS.get(c.cloud, c.cloud)}</td>
<td>{BENCH_LABELS[c.bench]}</td>
<td data-sort="{c.accuracy}">{c.accuracy:.3f}</td>
<td data-sort="{c.cost_usd}">${c.cost_usd:.2f}</td>
<td data-sort="{c.latency_med_s or 0}">{('%.1f' % c.latency_med_s + 's') if c.latency_med_s else '—'}</td>
<td data-sort="{c.tool_calls_mean if c.tool_calls_mean is not None else 0}">{('%.1f' % c.tool_calls_mean) if c.tool_calls_mean is not None else '—'}</td>
<td data-sort="{c.web_searches_mean if c.web_searches_mean is not None else 0}">{('%.2f' % c.web_searches_mean) if c.web_searches_mean is not None else '—'}</td>
<td data-sort="{c.tokens_local_total}">{c.tokens_local_total:,}</td>
<td data-sort="{c.tokens_cloud_total}">{c.tokens_cloud_total:,}</td>
</tr>""")
summary_table = f"""<table class="summary"><thead><tr>
<th>paradigm</th><th>local</th><th>cloud</th><th>bench</th>
<th>accuracy ↕</th><th>cost ↕</th><th>latency_med ↕</th><th>tool_calls ↕</th>
<th>web_searches ↕</th>
<th>tokens_local ↕</th><th>tokens_cloud ↕</th>
</tr></thead><tbody>{''.join(rows_html)}</tbody></table>"""
# axis sections (embed PNGs)
axis_sections = []
for axis, png_path in plots:
rel = f"plots-n100/{png_path.name}"
axis_sections.append(f"""<section>
<h2>{html_lib.escape(axis.title)}</h2>
<p class="desc">{html_lib.escape(axis.description)}</p>
<img src="{rel}" alt="{axis.title}">
</section>""")
md_html = ""
if RAW_MD.exists():
md_html = md_to_html(RAW_MD.read_text())
return f"""<!doctype html>
<html><head><meta charset="utf-8">
<title>OpenJarvis Hybrid n=100 Ablation</title>
<style>{CSS}</style>
</head><body>
<h1>OpenJarvis Hybrid n=100 Ablation</h1>
<p class="desc">Comparing local-cloud paradigms across 6 cloud models on GAIA + SWE-bench-Verified · {len(cells)} cells · generated {ts}</p>
<div class="card">
<h2 style="border:0; margin-top:0">Headline findings</h2>
{''.join(hero)}
</div>
<div class="card">
<h2 style="border:0; margin-top:0">All cells (sortable)</h2>
{legend_html}
{summary_table}
</div>
{''.join(axis_sections)}
<div class="card">
<h2 style="border:0; margin-top:0">results-table.md (full)</h2>
{md_html}
</div>
<script>{SORT_JS}</script>
</body></html>
"""
def main():
cells = load_cells()
print(f"loaded {len(cells)} cells")
plots = []
for axis in axis_definitions():
png = render_axis_png(axis, cells)
if png is not None:
plots.append((axis, png))
print(f" rendered {png.name}")
html = build_html(cells, plots)
OUT_HTML.write_text(html)
print(f"wrote {OUT_HTML} ({len(html):,} bytes)")
if __name__ == "__main__":
main()
src/openjarvis/agents/hybrid/_base.py
+233
View File
@@ -64,6 +64,35 @@ ANTHROPIC_WEB_SEARCH_TOOL = {
}
def build_web_search_tool(max_uses: int = 8) -> Dict[str, Any]:
"""Build an Anthropic server-side web_search tool block with a custom cap.
Web_search is server-side: Anthropic runs the searches internally before
returning, so ``max_uses`` is the only knob the caller has to bound cost
per task. Defaults to 8 (matches ``ANTHROPIC_WEB_SEARCH_TOOL``).
"""
return {
"type": "web_search_20250305",
"name": "web_search",
"max_uses": int(max_uses),
}
def web_search_cfg(method_cfg: Optional[Dict[str, Any]]) -> Tuple[bool, int]:
"""Parse ``method_cfg.web_search = { enabled, max_uses }``.
Defaults: enabled=False, max_uses=8. ``enabled`` defaults to False so
existing cells stay one-shot (backwards compat). Cells opting in flip
``enabled=true`` in the registry. Returns ``(enabled, max_uses)``.
"""
if not method_cfg:
return False, 8
ws = method_cfg.get("web_search")
if not isinstance(ws, dict):
return False, 8
return bool(ws.get("enabled", False)), int(ws.get("max_uses", 8))
# ---------- Thread-local trace buffer ----------
#
# Every call through ``_call_anthropic`` / ``_call_openai`` / ``_call_vllm``
@@ -352,6 +381,75 @@ class LocalCloudAgent(BaseAgent):
})
return text, p, c
@staticmethod
def _call_gemini(
model: str,
*,
user: str,
system: Optional[str] = None,
max_tokens: int = 4096,
temperature: float = 0.0,
timeout: float = 600.0,
trace_role: str = "cloud",
) -> Tuple[str, int, int]:
"""Single Gemini Developer-API call. Returns (text, p_tok, c_tok).
Tool-call parity with Anthropic is intentionally NOT implemented —
skillorchestra / baseline-cloud only need text generation, and the
google-genai tool-config plumbing diverges enough from the
Anthropic/OpenAI shape that wiring it would double the surface
area of this file. If a future paradigm needs Gemini tool use,
extend this helper rather than hand-rolling it in the agent.
Captures the call into the active per-task trace via the
``"gemini"`` kind so the dashboard's trace renderer picks it up.
"""
from google import genai
from google.genai import types
client = genai.Client(http_options=types.HttpOptions(timeout=int(timeout * 1000)))
cfg = types.GenerateContentConfig(
temperature=temperature,
max_output_tokens=max_tokens,
)
if system:
cfg.system_instruction = system
t0 = time.time()
resp = client.models.generate_content(
model=model,
contents=user,
config=cfg,
)
latency = time.time() - t0
# `resp.text` is the convenience accessor that concatenates every
# text part in the first candidate. Empty if the model emitted
# only non-text parts (which we don't request).
text = (resp.text or "") if hasattr(resp, "text") else ""
um = getattr(resp, "usage_metadata", None)
p = int(getattr(um, "prompt_token_count", 0) or 0) if um else 0
c = int(getattr(um, "candidates_token_count", 0) or 0) if um else 0
finish_reason = None
try:
finish_reason = str(resp.candidates[0].finish_reason)
except Exception:
pass
_record_event({
"kind": "gemini",
"role": trace_role,
"model": model,
"system": system,
"user": user,
"response": text,
"tokens_in": p,
"tokens_out": c,
"temperature": temperature,
"max_tokens": max_tokens,
"finish_reason": finish_reason,
"latency_s": latency,
"ts": time.time(),
})
return text, p, c
@staticmethod
def _call_vllm(
model: str,
@@ -426,6 +524,122 @@ class LocalCloudAgent(BaseAgent):
})
return text, p, c
@staticmethod
def _call_anthropic_agent(
model: str,
*,
user: str,
system: Optional[str] = None,
max_tokens: int = 4096,
temperature: float = 0.0,
tools: Optional[list] = None,
max_turns: int = 8,
timeout: float = 600.0,
max_retries: int = 5,
trace_role: str = "cloud",
) -> Tuple[str, int, int, int, int]:
"""Multi-turn Anthropic loop with optional tools.
Returns ``(final_text, prompt_tokens_sum, completion_tokens_sum,
n_web_searches_sum, turns)``. The loop appends the assistant
response (preserving server-tool blocks so Anthropic's continuation
is valid) and re-prompts until the model stops with ``end_turn``
or we hit ``max_turns``. Server-side tools (web_search) execute
inside a single message and don't require a tool_result echo —
the model just continues thinking with the search results
already in its context. Client-side tools aren't executed here;
if the model emits a client-side ``tool_use`` block we stop
(paradigms that want client tools should call ``_call_anthropic``
directly and handle their own dispatch).
"""
import anthropic
client = anthropic.Anthropic(timeout=timeout, max_retries=max_retries)
# Build the conversation. We grow ``messages`` across turns so the
# model sees its own prior assistant content (text + server tool
# use + web_search_tool_result). For server tools Anthropic is
# happy as long as we feed the raw assistant content back; no
# synthetic user tool_result block is needed.
messages: List[Dict[str, Any]] = [{"role": "user", "content": user}]
p_total = 0
c_total = 0
n_searches_total = 0
last_text = ""
turns = 0
for turn in range(max(1, max_turns)):
turns = turn + 1
kwargs: Dict[str, Any] = {
"model": model,
"max_tokens": max_tokens,
"messages": messages,
}
if system:
kwargs["system"] = system
if supports_temperature(model):
kwargs["temperature"] = temperature
if tools:
kwargs["tools"] = tools
t0 = time.time()
msg = client.messages.create(**kwargs)
latency = time.time() - t0
text = "".join(b.text for b in msg.content if hasattr(b, "text"))
srv = getattr(msg.usage, "server_tool_use", None)
n_searches = getattr(srv, "web_search_requests", 0) if srv else 0
content_blocks = [_serialize_block(b) for b in msg.content]
tool_use_blocks = [
b for b in content_blocks
if b.get("type") in ("tool_use", "server_tool_use")
]
tool_result_blocks = [
b for b in content_blocks
if b.get("type") in ("web_search_tool_result", "tool_result")
]
stop_reason = getattr(msg, "stop_reason", None)
_record_event({
"kind": "anthropic",
"role": trace_role,
"model": model,
"system": system if turn == 0 else None,
"user": user if turn == 0 else None,
"turn": turn,
"response": text,
"content_blocks": content_blocks,
"tool_calls": tool_use_blocks,
"tool_results": tool_result_blocks,
"tokens_in": msg.usage.input_tokens,
"tokens_out": msg.usage.output_tokens,
"n_web_searches": n_searches,
"tools_declared": tools,
"stop_reason": stop_reason,
"latency_s": latency,
"ts": time.time(),
})
p_total += msg.usage.input_tokens
c_total += msg.usage.output_tokens
n_searches_total += n_searches
if text:
last_text = text
# If the model wants a client-side tool we don't dispatch
# here — break and let the caller (or future loop variant)
# handle it. Only ``server_tool_use`` blocks (web_search)
# are auto-continued by Anthropic itself.
client_tool_use = any(
b.get("type") == "tool_use" for b in content_blocks
)
if client_tool_use:
break
if stop_reason == "end_turn" or stop_reason is None:
break
# Otherwise: ``stop_reason`` like "max_tokens" or "tool_use"
# (server side) — Anthropic returned mid-thought. Append the
# assistant turn and ask it to continue.
messages.append({"role": "assistant", "content": msg.content})
messages.append({
"role": "user",
"content": "Continue.",
})
return last_text, p_total, c_total, n_searches_total, turns
def _call_cloud(
self,
*,
@@ -460,6 +674,23 @@ class LocalCloudAgent(BaseAgent):
temperature=temperature,
**kwargs,
)
if self._cloud_endpoint == "gemini":
# Gemini helper doesn't accept tools / response_format kwargs.
# Drop them rather than letting them surface as a TypeError so
# paradigms that opportunistically pass these for OpenAI /
# Anthropic still work when routed to Gemini.
kwargs.pop("tools", None)
kwargs.pop("tool_choice", None)
kwargs.pop("response_format", None)
kwargs.pop("output_config", None)
return self._call_gemini(
self._cloud_model,
user=user,
system=system,
max_tokens=max_tokens,
temperature=temperature,
**kwargs,
)
raise ValueError(f"unsupported cloud endpoint: {self._cloud_endpoint!r}")
# ------------------------------------------------------------------
@@ -609,7 +840,9 @@ __all__ = [
"LocalCloudAgent",
"NO_TEMP_PREFIXES",
"WEB_SEARCH_COST_PER_CALL",
"build_web_search_tool",
"estimate_cost",
"is_gpt5_family",
"supports_temperature",
"web_search_cfg",
]
src/openjarvis/agents/hybrid/advisors.py
+54 -14
View File
@@ -28,7 +28,12 @@ import urllib.request
from typing import Any, Dict, Optional, Tuple
from openjarvis.agents._stubs import AgentContext
from openjarvis.agents.hybrid._base import LocalCloudAgent
from openjarvis.agents.hybrid._base import (
WEB_SEARCH_COST_PER_CALL,
LocalCloudAgent,
build_web_search_tool,
web_search_cfg,
)
from openjarvis.agents.hybrid.mini_swe_agent import (
run_swe_agent_loop,
)
@@ -112,13 +117,32 @@ class AdvisorsAgent(LocalCloudAgent):
advisor_max_tokens = int(cfg.get("advisor_max_tokens", 1024))
advisor_temperature = float(cfg.get("advisor_temperature", 0.2))
# 1. Initial executor pass
initial_resp, e1_in, e1_out = self._call_cloud(
user=f"Question:\n{question}",
system=EXECUTOR_INITIAL_SYS,
max_tokens=executor_max_tokens,
temperature=0.0,
)
ws_enabled, ws_max_uses = web_search_cfg(cfg)
use_ws = ws_enabled and self._cloud_endpoint == "anthropic"
gaia_max_turns = int(cfg.get("gaia_max_turns", 8))
ws_tool = [build_web_search_tool(ws_max_uses)] if use_ws else None
n_searches_total = 0
# 1. Initial executor pass — advisor (Qwen) doesn't get tools;
# only the cloud executor passes do.
if use_ws:
initial_resp, e1_in, e1_out, n_s1, _ = self._call_anthropic_agent(
self._cloud_model,
user=f"Question:\n{question}",
system=EXECUTOR_INITIAL_SYS,
max_tokens=executor_max_tokens,
temperature=0.0,
tools=ws_tool,
max_turns=gaia_max_turns,
)
n_searches_total += n_s1
else:
initial_resp, e1_in, e1_out = self._call_cloud(
user=f"Question:\n{question}",
system=EXECUTOR_INITIAL_SYS,
max_tokens=executor_max_tokens,
temperature=0.0,
)
# 2. Advisor pass (local)
if not self._local_endpoint or not self._local_model:
@@ -147,25 +171,41 @@ class AdvisorsAgent(LocalCloudAgent):
f"Produce your best final answer now, respecting the question's "
f"answer-format rules."
)
final_answer, e2_in, e2_out = self._call_cloud(
user=final_user,
system=EXECUTOR_FINAL_SYS,
max_tokens=executor_max_tokens,
temperature=0.0,
)
if use_ws:
final_answer, e2_in, e2_out, n_s2, _ = self._call_anthropic_agent(
self._cloud_model,
user=final_user,
system=EXECUTOR_FINAL_SYS,
max_tokens=executor_max_tokens,
temperature=0.0,
tools=ws_tool,
max_turns=gaia_max_turns,
)
n_searches_total += n_s2
else:
final_answer, e2_in, e2_out = self._call_cloud(
user=final_user,
system=EXECUTOR_FINAL_SYS,
max_tokens=executor_max_tokens,
temperature=0.0,
)
tokens_local = adv_in + adv_out
tokens_cloud = e1_in + e1_out + e2_in + e2_out
cost = self.cost_usd(self._cloud_model, e1_in + e2_in, e1_out + e2_out)
cost += n_searches_total * WEB_SEARCH_COST_PER_CALL
meta: Dict[str, Any] = {
"tokens_local": tokens_local,
"tokens_cloud": tokens_cloud,
"cost_usd": cost,
"turns": 3,
"web_search_uses": n_searches_total,
"traces": {
"initial_response": initial_resp,
"advisor_feedback": advisor_text,
"web_search_enabled": use_ws,
"n_web_searches": n_searches_total,
"note": "inference-only advisor (untrained); lower bound on the technique.",
},
}
src/openjarvis/agents/hybrid/archon.py
+48 -1
View File
@@ -42,7 +42,13 @@ import types
from typing import Any, Dict, List, Optional, Tuple
from openjarvis.agents._stubs import AgentContext
from openjarvis.agents.hybrid._base import LocalCloudAgent, _record_event
from openjarvis.agents.hybrid._base import (
WEB_SEARCH_COST_PER_CALL,
LocalCloudAgent,
_record_event,
build_web_search_tool,
web_search_cfg,
)
from openjarvis.agents.hybrid._prices import (
NO_TEMP_PREFIXES,
)
@@ -125,8 +131,11 @@ def _tally() -> Dict[str, int]:
counts = {
"cloud_prompt": 0, "cloud_completion": 0,
"local_prompt": 0, "local_completion": 0,
"n_web_searches": 0,
}
_TALLY_LOCAL.counts = counts
# Backfill for older threads that started before we added the key.
counts.setdefault("n_web_searches", 0)
return counts
@@ -134,9 +143,24 @@ def _reset_tally() -> None:
_TALLY_LOCAL.counts = {
"cloud_prompt": 0, "cloud_completion": 0,
"local_prompt": 0, "local_completion": 0,
"n_web_searches": 0,
}
# Per-thread web_search tool: when set, the Anthropic generator declares
# it on every call. Set inside ``_run_paradigm`` before invoking Archon
# so the ranker/fuser passes pick it up; cleared in ``finally``.
_WS_LOCAL = threading.local()
def _set_anthropic_web_search(tool: Optional[Dict[str, Any]]) -> None:
_WS_LOCAL.tool = tool
def _get_anthropic_web_search() -> Optional[Dict[str, Any]]:
return getattr(_WS_LOCAL, "tool", None)
def _make_local_generator(local_endpoint: str, local_model: str):
"""Archon custom-generator signature: (model, messages, max_tokens, temperature) -> str."""
from openai import OpenAI
@@ -237,6 +261,9 @@ def _wrap_archon_cloud_generators() -> None:
)
if not model.startswith(NO_TEMP_PREFIXES):
kwargs["temperature"] = temperature
ws_tool = _get_anthropic_web_search()
if ws_tool is not None:
kwargs["tools"] = [ws_tool]
t0 = _time.time()
resp = client.messages.create(**kwargs)
text = "".join(b.text for b in resp.content if hasattr(b, "text"))
@@ -244,6 +271,9 @@ def _wrap_archon_cloud_generators() -> None:
if u:
_tally()["cloud_prompt"] += getattr(u, "input_tokens", 0) or 0
_tally()["cloud_completion"] += getattr(u, "output_tokens", 0) or 0
srv = getattr(u, "server_tool_use", None) if u else None
n_searches = getattr(srv, "web_search_requests", 0) if srv else 0
_tally()["n_web_searches"] += int(n_searches)
_record_event({
"kind": "archon_cloud_anthropic",
"model": model,
@@ -252,6 +282,8 @@ def _wrap_archon_cloud_generators() -> None:
"response": text.strip(),
"tokens_in": getattr(u, "input_tokens", 0) if u else 0,
"tokens_out": getattr(u, "output_tokens", 0) if u else 0,
"n_web_searches": int(n_searches),
"tools_declared": kwargs.get("tools"),
"latency_s": _time.time() - t0,
"ts": _time.time(),
})
@@ -430,6 +462,14 @@ class ArchonAgent(LocalCloudAgent):
archon_cfg = {"name": f"hybrid-archon-{arch}", "layers": layers}
_reset_tally()
# Web_search opt-in: when enabled, declare the native server-side
# tool on Anthropic ranker/fuser passes (via thread-local). The
# local proposers run on vLLM and don't see it.
ws_enabled, ws_max_uses = web_search_cfg(cfg)
if ws_enabled:
_set_anthropic_web_search(build_web_search_tool(ws_max_uses))
else:
_set_anthropic_web_search(None)
archon = Archon(archon_cfg)
try:
@@ -439,6 +479,8 @@ class ArchonAgent(LocalCloudAgent):
])
except Exception as e:
answer = f"[archon error: {e!r}]"
finally:
_set_anthropic_web_search(None)
if isinstance(answer, list):
answer = answer[-1] if answer else ""
@@ -446,16 +488,19 @@ class ArchonAgent(LocalCloudAgent):
cp = _tally()["cloud_prompt"]
cc = _tally()["cloud_completion"]
n_searches = _tally().get("n_web_searches", 0)
cost = _cost_cloud(ranker_model, cp, cc)
if fuser_model != ranker_model:
# Conservative: charge both at the more expensive of the two.
cost = max(cost, _cost_cloud(fuser_model, cp, cc))
cost += n_searches * WEB_SEARCH_COST_PER_CALL
meta = {
"tokens_local": _tally()["local_prompt"] + _tally()["local_completion"],
"tokens_cloud": cp + cc,
"cost_usd": cost,
"turns": (K + 2) if arch == "ensemble_rank_fuse" else 1,
"web_search_uses": n_searches,
"traces": {
"architecture": arch,
"n_samples": K,
@@ -463,6 +508,8 @@ class ArchonAgent(LocalCloudAgent):
"fuser_model": fuser_model,
"local_model": self._local_model,
"tokens_breakdown": dict(_tally()),
"web_search_enabled": ws_enabled,
"n_web_searches": n_searches,
},
}
return answer, meta
src/openjarvis/agents/hybrid/baseline_cloud.py
+167
View File
@@ -0,0 +1,167 @@
"""BaselineCloudAgent — cloud-only reference for the hybrid ablation.
Used as the "what does the cloud do alone?" row in the n=100 ablation
matrix (see ``.openjarvis/experiments/hybrid/docs/results-table.md``).
No local model is involved — ``local_*`` settings are ignored.
On GAIA the agent makes one cloud call with the formatted prompt (which
already carries the ``FINAL ANSWER:`` format reminder from
``_prompts.format_gaia``) and returns the text. On SWE-bench-Verified
the agent delegates to :func:`run_swe_agent_loop` with ``backbone="cloud"``
so the model gets to run bash and read the repo — same wiring as the
``mini-swe-agent-swebenchverified-opus-*`` cells. As of 2026-05-15
``_loop_cloud`` dispatches to per-endpoint loops for Anthropic, OpenAI,
and Gemini so all three cloud backbones get the proper bash-agent loop
on SWE (previously OpenAI / Gemini SWE cells silently fell back to a
one-shot blind patch — fixed).
Construction args mirror :class:`LocalCloudAgent`. The ``cloud`` block
in the cell registry determines the cloud model + endpoint; ``local``
is accepted for schema compatibility but unused.
"""
from __future__ import annotations
from typing import Any, Dict, Optional, Tuple
from openjarvis.agents._stubs import AgentContext
from openjarvis.agents.hybrid._base import (
WEB_SEARCH_COST_PER_CALL,
LocalCloudAgent,
build_web_search_tool,
web_search_cfg,
)
from openjarvis.agents.hybrid._prices import cost as estimate_cost
from openjarvis.agents.hybrid.mini_swe_agent import run_swe_agent_loop
from openjarvis.core.registry import AgentRegistry
@AgentRegistry.register("baseline_cloud")
class BaselineCloudAgent(LocalCloudAgent):
"""Cloud-only baseline used as a reference in the n=100 ablation.
Configurable knobs via ``cfg``:
- ``cloud_max_tokens`` (int, default 4096): max_tokens per GAIA call
and per turn of the SWE agent loop.
- ``swe_max_turns`` (int, default 30): SWE-bench loop turn cap.
- ``swe_bash_timeout_s`` (int, default 120): SWE-bench bash timeout.
"""
agent_id = "baseline_cloud"
def _run_paradigm(
self,
input: str,
context: Optional[AgentContext],
**kwargs: Any,
) -> Tuple[str, Dict[str, Any]]:
cfg = self._cfg
task: Dict[str, Any] = {}
if context is not None:
task = context.metadata.get("task") or {}
is_swe = bool(
task.get("problem_statement")
and task.get("repo")
and task.get("base_commit")
)
if is_swe:
out = run_swe_agent_loop(
task,
backbone="cloud",
backbone_model=self._cloud_model,
cloud_endpoint=self._cloud_endpoint,
initial_prompt=input,
max_turns=int(cfg.get("swe_max_turns", 30)),
bash_timeout=int(cfg.get("swe_bash_timeout_s", 120)),
output_cap=int(cfg.get("swe_output_cap", 10_000)),
turn_max_tokens=int(cfg.get("cloud_max_tokens", 4096)),
trace_prefix="baseline_cloud",
)
meta = {
"tokens_local": 0,
"tokens_cloud": out["tokens_in"] + out["tokens_out"],
"cost_usd": out["cost_usd"],
"turns": out["turns"],
"traces": {
"backbone": "cloud",
"max_turns_hit": out["max_turns_hit"],
"patch_chars": len(out["patch"]),
"final_summary": out["final_summary"],
},
}
return out["answer"], meta
# GAIA branch. If `web_search.enabled` is true AND we're on
# Anthropic, run the multi-turn agent loop with the native
# server-side web_search tool. Otherwise fall back to the
# legacy one-shot call (preserves behavior of every existing
# non-opted-in cell).
ws_enabled, ws_max_uses = web_search_cfg(cfg)
gaia_max_turns = int(cfg.get("gaia_max_turns", 8))
if ws_enabled and self._cloud_endpoint == "anthropic":
text, p_tok, c_tok, n_searches, turns = self._call_anthropic_agent(
self._cloud_model,
user=input,
max_tokens=int(cfg.get("cloud_max_tokens", 4096)),
temperature=0.0,
tools=[build_web_search_tool(ws_max_uses)],
max_turns=gaia_max_turns,
)
cost = (
estimate_cost(self._cloud_model, p_tok, c_tok)
+ n_searches * WEB_SEARCH_COST_PER_CALL
)
meta = {
"tokens_local": 0,
"tokens_cloud": p_tok + c_tok,
"cost_usd": cost,
"turns": turns,
"web_search_uses": n_searches,
"traces": {
"mode": "anthropic_agent_loop",
"is_swe": is_swe,
"cloud_endpoint": self._cloud_endpoint,
"web_search_enabled": True,
"web_search_max_uses": ws_max_uses,
"n_web_searches": n_searches,
},
}
return text, meta
if ws_enabled and self._cloud_endpoint != "anthropic":
# OpenAI / Gemini don't have a parity native web_search tool
# wired here. Skip cleanly rather than fake one. Cells that
# want web_search must run on Anthropic until those backends
# are wired.
self.record_trace_event({
"kind": "web_search_skipped",
"reason": "non_anthropic_endpoint",
"endpoint": self._cloud_endpoint,
})
# One-shot direct cloud call. GAIA only — SWE goes through the
# mini-SWE-agent loop above (now supports anthropic/openai/gemini).
text, p_tok, c_tok = self._call_cloud(
user=input,
max_tokens=int(cfg.get("cloud_max_tokens", 4096)),
temperature=0.0,
)
meta = {
"tokens_local": 0,
"tokens_cloud": p_tok + c_tok,
"cost_usd": estimate_cost(self._cloud_model, p_tok, c_tok),
"turns": 1,
"web_search_uses": 0,
"traces": {
"mode": "one_shot",
"is_swe": is_swe,
"cloud_endpoint": self._cloud_endpoint,
},
}
return text, meta
__all__ = ["BaselineCloudAgent"]
src/openjarvis/agents/hybrid/conductor.py
+178 -15
View File
@@ -39,7 +39,12 @@ from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from openjarvis.agents._stubs import AgentContext
from openjarvis.agents.hybrid._base import LocalCloudAgent
from openjarvis.agents.hybrid._base import (
WEB_SEARCH_COST_PER_CALL,
LocalCloudAgent,
build_web_search_tool,
web_search_cfg,
)
from openjarvis.agents.hybrid._prices import (
PRICES,
is_gpt5_family,
@@ -340,6 +345,130 @@ def _resolve_worker_pool(
return resolved
# Endpoints conductor's `_call_worker` actually knows how to dispatch to.
# Web-search and gemini are NOT supported here — toolorchestra has the
# web-search dispatcher; gemini isn't wired into _call_worker.
_CONDUCTOR_VALID_ENDPOINTS = ("vllm", "openai", "anthropic")
def _resolve_worker_pool(
cfg: Dict[str, Any],
local_model: Optional[str],
local_endpoint: Optional[str],
cloud_model: str,
) -> List[Dict[str, Any]]:
"""Return the worker pool for this run.
Strict replace, not merge: if ``cfg["worker_pool"]`` is set, the
default pool is ignored entirely. Falls back to ``_default_pool`` when
the override is absent.
Each user-supplied entry must be a dict with keys ``id``, ``name``,
``endpoint``, and ``model``. ``endpoint`` must be one of
``vllm`` / ``openai`` / ``anthropic`` — conductor does not wire
web-search or gemini workers.
Substitution: ``model = "$local"`` (or ``"<local>"``) resolves to
``local_model``; ``model = "$cloud"`` / ``"<cloud>"`` to ``cloud_model``.
On any validation failure, raises ``ValueError`` with the message
``"Invalid worker_pool entry [<id>]: <reason>"``. Fails fast at agent
init rather than mid-task.
"""
override = cfg.get("worker_pool")
if override is None:
return _default_pool(local_model, local_endpoint)
if not isinstance(override, list) or not override:
raise ValueError(
"Invalid worker_pool entry [-]: worker_pool must be a non-empty list"
)
resolved: List[Dict[str, Any]] = []
seen_ids: set = set()
has_non_search = False
for raw in override:
wid_repr = raw.get("id", "?") if isinstance(raw, dict) else "?"
if not isinstance(raw, dict):
raise ValueError(
f"Invalid worker_pool entry [{wid_repr}]: entry must be a dict"
)
entry = dict(raw)
wid = entry.get("id")
if not isinstance(wid, int):
raise ValueError(
f"Invalid worker_pool entry [{wid_repr}]: 'id' must be an int"
)
if wid in seen_ids:
raise ValueError(
f"Invalid worker_pool entry [{wid}]: duplicate id"
)
seen_ids.add(wid)
if not entry.get("name") or not isinstance(entry["name"], str):
raise ValueError(
f"Invalid worker_pool entry [{wid}]: 'name' must be a non-empty string"
)
endpoint = entry.get("endpoint") or entry.get("type")
if not isinstance(endpoint, str) or endpoint.lower() not in _CONDUCTOR_VALID_ENDPOINTS:
raise ValueError(
f"Invalid worker_pool entry [{wid}]: 'endpoint' must be one of "
f"{_CONDUCTOR_VALID_ENDPOINTS} (got {endpoint!r})"
)
endpoint = endpoint.lower()
entry["endpoint"] = endpoint
# Substitute $local / $cloud placeholders.
model = entry.get("model")
if isinstance(model, str) and model in ("$local", "<local>"):
if not local_model:
raise ValueError(
f"Invalid worker_pool entry [{wid}]: model='{model}' "
"requires a local_model to be configured for this cell"
)
model = local_model
entry["model"] = model
elif isinstance(model, str) and model in ("$cloud", "<cloud>"):
model = cloud_model
entry["model"] = model
if not isinstance(model, str) or not model:
raise ValueError(
f"Invalid worker_pool entry [{wid}]: 'model' must be a non-empty string"
)
if endpoint == "vllm":
if not entry.get("base_url"):
# Default to the local endpoint if not specified — matches
# how _default_pool wires it.
if not local_endpoint:
raise ValueError(
f"Invalid worker_pool entry [{wid}]: vllm worker needs "
"'base_url' (or a configured local_endpoint to fall back to)"
)
entry["base_url"] = local_endpoint
entry.setdefault("api_key", "EMPTY")
# Local also counts as a non-search worker for the
# "must have at least one solver" check.
has_non_search = True
else:
# Cloud workers: model must be priced (any unknown model would
# silently cost $0, which masks billing mistakes downstream).
if model not in PRICES:
raise ValueError(
f"Invalid worker_pool entry [{wid}]: model {model!r} is "
f"not in PRICES (known: {sorted(PRICES)})"
)
has_non_search = True
entry.setdefault(
"description",
f"User-supplied {endpoint} worker ({model}).",
)
resolved.append(entry)
if not has_non_search:
raise ValueError(
"Invalid worker_pool entry [-]: worker_pool must contain at least "
"one non-search worker (vllm / openai / anthropic)"
)
return resolved
def _format_worker_pool(workers: List[Dict[str, Any]]) -> str:
return "\n".join(
f"Model {w['id']} ({w['name']}): {w['description']}" for w in workers
@@ -376,9 +505,19 @@ def _build_step_prompt(
# ---------- Worker invocation ----------
def _call_worker(
worker: Dict[str, Any], prompt: str, cfg: Dict[str, Any]
) -> Tuple[str, int, int, bool]:
"""Returns (text, p_tok, c_tok, is_local)."""
worker: Dict[str, Any],
prompt: str,
cfg: Dict[str, Any],
*,
web_search_tool: Optional[Dict[str, Any]] = None,
) -> Tuple[str, int, int, bool, int]:
"""Returns (text, p_tok, c_tok, is_local, n_web_searches).
``web_search_tool``: if set AND the worker endpoint is Anthropic, the
server-side web_search tool is declared on the call. ``n_web_searches``
is the actual count Anthropic ran. Non-Anthropic workers ignore it
and return 0.
"""
ep = (worker.get("endpoint") or "openai").lower()
max_tok = int(cfg.get("worker_max_tokens", 4096))
temp = float(cfg.get("worker_temperature", 0.2))
@@ -392,7 +531,7 @@ def _call_worker(
temperature=temp,
enable_thinking=False,
)
return text, p, c, True
return text, p, c, True, 0
if ep == "openai":
text, p, c = LocalCloudAgent._call_openai(
worker["model"],
@@ -400,16 +539,20 @@ def _call_worker(
max_tokens=max_tok,
temperature=(1.0 if is_gpt5_family(worker["model"]) else temp),
)
return text, p, c, False
return text, p, c, False, 0
if ep == "anthropic":
eff_temp = temp if supports_temperature(worker["model"]) else 0.0
text, p, c, _ = LocalCloudAgent._call_anthropic(
worker["model"],
anthropic_kwargs: Dict[str, Any] = dict(
user=prompt,
max_tokens=max_tok,
temperature=eff_temp,
)
return text, p, c, False
if web_search_tool is not None:
anthropic_kwargs["tools"] = [web_search_tool]
text, p, c, n_searches = LocalCloudAgent._call_anthropic(
worker["model"], **anthropic_kwargs
)
return text, p, c, False, n_searches
raise ValueError(f"unsupported worker endpoint: {ep!r}")
@@ -420,10 +563,11 @@ def _swe_worker_step(
cfg: Dict[str, Any],
workdir: Path,
step_idx: int,
) -> Tuple[str, int, int, bool]:
) -> Tuple[str, int, int, bool, int]:
"""Run one Conductor worker step as a mini-SWE-agent subloop on a shared
workdir. Returns (final_summary_or_diff, tokens_in, tokens_out, is_local)
in the same shape as ``_call_worker``."""
workdir. Returns (final_summary_or_diff, tokens_in, tokens_out, is_local,
n_web_searches) in the same shape as ``_call_worker``. SWE workers
don't use web_search (the bash tool is the only tool they need)."""
ep = (worker.get("endpoint") or "openai").lower()
if ep == "vllm":
backbone, model, endpoint, is_local = (
@@ -455,7 +599,10 @@ def _swe_worker_step(
trace_prefix=f"conductor_step{step_idx}",
workdir=workdir,
)
return out["final_summary"] or out["answer"], out["tokens_in"], out["tokens_out"], is_local
return (
out["final_summary"] or out["answer"],
out["tokens_in"], out["tokens_out"], is_local, 0,
)
@AgentRegistry.register("conductor")
@@ -566,6 +713,15 @@ class ConductorAgent(LocalCloudAgent):
cost = 0.0
final_answer = ""
shared_workdir: Optional[Path] = None
n_web_searches_total = 0
# Web_search opt-in: when enabled, declare the native server-side
# tool on Anthropic worker calls. GAIA-only — SWE workers use bash
# not web_search. OpenAI / vLLM workers ignore it.
ws_enabled, ws_max_uses = web_search_cfg(cfg)
ws_tool = (
build_web_search_tool(ws_max_uses) if ws_enabled else None
)
try:
if swe_mode:
@@ -598,17 +754,21 @@ class ConductorAgent(LocalCloudAgent):
})
if swe_mode:
text, w_in, w_out, is_local = _swe_worker_step(
text, w_in, w_out, is_local, n_searches = _swe_worker_step(
worker, task_meta, prompt, cfg, shared_workdir, i,
)
else:
text, w_in, w_out, is_local = _call_worker(worker, prompt, cfg)
text, w_in, w_out, is_local, n_searches = _call_worker(
worker, prompt, cfg, web_search_tool=ws_tool,
)
if is_local:
tokens_local += w_in + w_out
else:
tokens_cloud += w_in + w_out
cost += self.cost_usd(worker["model"], w_in, w_out)
cost += n_searches * WEB_SEARCH_COST_PER_CALL
n_web_searches_total += n_searches
steps.append({
"step_idx": i,
"model_id": mid,
@@ -650,10 +810,13 @@ class ConductorAgent(LocalCloudAgent):
"tokens_cloud": tokens_cloud,
"cost_usd": cost,
"turns": len(steps) + 1, # planner + N execution steps
"web_search_uses": n_web_searches_total,
"traces": {
"steps": traces,
"plan": plan,
"fallback_used": fallback_used,
"web_search_enabled": ws_enabled,
"n_web_searches": n_web_searches_total,
"parse_attempts": parse_attempts,
"workers": [
{k: v for k, v in w.items() if k != "api_key"}
src/openjarvis/agents/hybrid/minions.py
+36 -4
View File
@@ -48,6 +48,8 @@ from openjarvis.agents.hybrid._base import (
ANTHROPIC_WEB_SEARCH_TOOL,
WEB_SEARCH_COST_PER_CALL,
LocalCloudAgent,
build_web_search_tool,
web_search_cfg,
)
from openjarvis.agents.hybrid._prices import NO_TEMP_PREFIXES
from openjarvis.agents.hybrid.mini_swe_agent import run_swe_agent_loop
@@ -200,6 +202,13 @@ def _patch_anthropic_globally() -> None:
def make_patched(orig): # type: ignore[no-untyped-def]
def patched(self, **kwargs): # type: ignore[no-untyped-def]
# External Minions's AnthropicClient.chat passes
# `cache_control={"type":"ephemeral"}` as a top-level kwarg
# (clients/anthropic.py:207). Newer Anthropic SDKs reject that
# — cache_control belongs on individual content blocks, not on
# Messages.create itself. Strip it; we're not relying on the
# ephemeral hint for correctness in these short minions turns.
kwargs.pop("cache_control", None)
model = kwargs.get("model", "")
if model.startswith(NO_TEMP_PREFIXES):
kwargs.pop("temperature", None)
@@ -251,7 +260,12 @@ def _apply_patches_once() -> None:
# ---------- Pre-fetch helper (GAIA only) ----------
def _prefetch_context(question: str, cloud_endpoint: str, cloud_model: str) -> Dict[str, Any]:
def _prefetch_context(
question: str,
cloud_endpoint: str,
cloud_model: str,
max_uses: int = 8,
) -> Dict[str, Any]:
"""Use Anthropic web_search to fetch real source material the worker can read.
Minions's premise is "worker reads a doc, asks cloud for help" — but GAIA
@@ -278,7 +292,7 @@ def _prefetch_context(question: str, cloud_endpoint: str, cloud_model: str) -> D
cloud_model,
user=prompt,
max_tokens=8192,
tools=[ANTHROPIC_WEB_SEARCH_TOOL],
tools=[build_web_search_tool(max_uses)],
tool_choice={"type": "any"},
)
from openjarvis.agents.hybrid._prices import cost as _cost_usd
@@ -412,12 +426,29 @@ class MinionsAgent(LocalCloudAgent):
# GAIA-shape only: prefetch a web_search digest so the worker has
# something real to read. SWE-bench (problem_statement only) already
# ships its own doc.
#
# Honors the new ``method_cfg.web_search`` schema:
# - omitted → prefetch ON (legacy default for minions GAIA)
# - enabled = false → prefetch OFF
# - enabled = true → prefetch ON (honors max_uses)
prefetch: Dict[str, Any] = {
"text": "", "tokens": 0, "cost_usd": 0.0, "n_searches": 0,
}
if task_meta.get("question"):
ws_block = cfg.get("web_search") if isinstance(cfg.get("web_search"), dict) else None
ws_enabled, ws_max_uses = web_search_cfg(cfg)
# If the cell explicitly set web_search.enabled = false, honor that.
# If it set web_search.enabled = true, honor max_uses. If it didn't
# set a web_search block at all, keep legacy prefetch ON.
prefetch_on = (
ws_block is None # legacy default
or ws_enabled
)
if task_meta.get("question") and prefetch_on:
prefetch = _prefetch_context(
task_meta["question"], self._cloud_endpoint, self._cloud_model
task_meta["question"],
self._cloud_endpoint,
self._cloud_model,
max_uses=ws_max_uses,
)
if prefetch.get("text"):
@@ -462,6 +493,7 @@ class MinionsAgent(LocalCloudAgent):
"tokens_cloud": (rp + rc) + prefetch["tokens"],
"cost_usd": self.cost_usd(self._cloud_model, rp, rc) + prefetch["cost_usd"],
"turns": cfg.get("max_rounds", 3),
"web_search_uses": prefetch["n_searches"],
"traces": {
"mode": mode,
"supervisor_messages": out.get("supervisor_messages"),
src/openjarvis/agents/hybrid/runner.py
+155 -12
View File
@@ -43,13 +43,45 @@ DEFAULT_REGISTRY_DIR = PACKAGE_DIR / "registry"
DEFAULT_EXPERIMENTS_DIR = Path(
os.environ.get(
"OPENJARVIS_HYBRID_EXPERIMENTS_DIR",
str(Path.home() / ".openjarvis-hybrid" / "experiments"),
"/matx/u/aspark/.openjarvis/experiments/hybrid",
)
)
DEFAULT_SUBSETS_DIR = DEFAULT_EXPERIMENTS_DIR / "subsets"
# ---------- Registry ----------
_SWE_BENCHES = {"swebench-verified", "swebench_verified", "swebench"}
def _validate_cells(cells: Dict[str, Dict[str, Any]]) -> None:
"""Catch registry mistakes that would silently degrade behaviour.
Currently: skillorchestra on a SWE bench MUST have
``method_cfg.swe_use_agent_loop = true``. Without the flag,
skillorchestra.py falls back to a one-shot cloud call even for
SWE-bench tasks, which is rarely what the experimenter wants and is
invisible at runtime (Bug 5, 2026-05-15).
"""
bad: List[str] = []
for name, cell in cells.items():
if cell.get("method") != "skillorchestra":
continue
if cell.get("bench") not in _SWE_BENCHES:
continue
mcfg = cell.get("method_cfg") or {}
if not bool(mcfg.get("swe_use_agent_loop")):
bad.append(name)
if bad:
raise ValueError(
"skillorchestra SWE cells missing required "
"`method_cfg.swe_use_agent_loop = true`: "
+ ", ".join(sorted(bad))
+ ". Without this flag the cell silently falls back to a "
"one-shot cloud call for SWE-bench tasks."
)
def load_registry(registry_dir: Optional[Path] = None) -> Dict[str, Dict[str, Any]]:
"""Merge every ``<registry_dir>/*.toml``. Cell names must be unique."""
base = registry_dir or DEFAULT_REGISTRY_DIR
@@ -67,6 +99,7 @@ def load_registry(registry_dir: Optional[Path] = None) -> Dict[str, Dict[str, An
f"duplicate cell {name!r} (already defined before {p.name})"
)
cells[name] = cell
_validate_cells(cells)
return cells
@@ -81,12 +114,17 @@ def _load_gaia_tasks(n: Optional[int]) -> List[Dict[str, Any]]:
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.
# the GAIA-specific fields including any reference answer. Prefer the
# upstream GAIA `task_id` field (bare uuid) over rec.record_id (which
# OpenJarvis prefixes with `gaia-`) so subsets keyed by the upstream
# id round-trip.
md = rec.metadata or {}
task_id = md.get("task_id") or rec.record_id
out.append({
"task_id": rec.record_id,
"question": rec.metadata.get("question", rec.problem),
"task_id": task_id,
"question": md.get("question", rec.problem),
"reference": rec.reference,
"metadata": dict(rec.metadata),
"metadata": dict(md),
})
return out
@@ -95,7 +133,7 @@ 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 = SWEBenchDataset(variant="verified")
ds.load(max_samples=n)
out: List[Dict[str, Any]] = []
for rec in ds.iter_records():
@@ -124,6 +162,72 @@ def load_tasks(bench: str, n: Optional[int]) -> List[Dict[str, Any]]:
raise ValueError(f"unknown bench: {bench!r}")
def _load_subset_file(subset_path: str) -> Dict[str, Any]:
"""Resolve a cell's ``subset`` field to a parsed JSON dict.
Resolution order:
1. Absolute path → use as-is.
2. Bare filename / relative path → look up under
``<experiments>/subsets/`` (matches where ``make_subset.py``
writes its output).
Accepts both list-of-ids and dict-with-task_ids shapes; the legacy
harness wrote the dict shape and we preserve that. Returns a dict
with at least a ``task_ids`` list so callers don't have to branch.
"""
p = Path(subset_path)
if not p.is_absolute():
p = DEFAULT_SUBSETS_DIR / p
if not p.exists():
raise FileNotFoundError(f"subset file not found: {p}")
data = json.loads(p.read_text())
if isinstance(data, list):
return {"task_ids": list(data)}
if isinstance(data, dict):
if "task_ids" not in data:
raise ValueError(
f"subset {p.name} has no 'task_ids' field; got keys {list(data.keys())}"
)
return data
raise ValueError(f"subset {p.name} must be a list or dict; got {type(data).__name__}")
def _apply_subset(
tasks: List[Dict[str, Any]],
subset: Dict[str, Any],
cell: Dict[str, Any],
) -> List[Dict[str, Any]]:
"""Filter ``tasks`` to the subset's task IDs, preserving subset order.
Hard-errors if the cell's ``n`` doesn't equal ``len(task_ids)`` so a
typo in the registry can't silently shrink the eval. Also errors if
any subset ID is missing from the dataset (caller's bench wiring is
broken).
"""
ids: List[str] = list(subset["task_ids"])
cell_n = int(cell["n"])
if cell_n != len(ids):
raise ValueError(
f"subset n={len(ids)} ≠ cell n={cell_n} — refusing to silently "
"change scope. Fix the registry's `n` to match the subset file."
)
if "bench" in subset and subset["bench"] != cell["bench"]:
raise ValueError(
f"subset bench={subset['bench']!r} ≠ cell bench={cell['bench']!r}"
)
order = {tid: i for i, tid in enumerate(ids)}
allow = set(ids)
kept = [t for t in tasks if t["task_id"] in allow]
kept.sort(key=lambda t: order[t["task_id"]])
missing = allow - {t["task_id"] for t in kept}
if missing:
raise ValueError(
f"subset references {len(missing)} task_ids not in dataset "
f"(e.g. {next(iter(missing))!r})"
)
return kept
# ---------- Scoring ----------
def _score_gaia(task: Dict[str, Any], answer: str) -> Dict[str, Any]:
@@ -275,6 +379,7 @@ def _run_one(agent, bench: str, task: Dict[str, Any], log_dir: str) -> Dict[str,
"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)),
"web_search_uses": int(meta.get("web_search_uses", 0)),
"traces": meta.get("traces", {}),
}
if "soft_error" in meta:
@@ -286,6 +391,7 @@ def _run_one(agent, bench: str, task: Dict[str, Any], log_dir: str) -> Dict[str,
"answer": "",
"tokens_local": 0, "tokens_cloud": 0,
"cost_usd": 0.0, "latency_s": time.time() - t0,
"web_search_uses": 0,
"traces": {},
"error": f"{type(e).__name__}: {e}\n{traceback.format_exc()}",
}
@@ -312,6 +418,7 @@ def _write_summary(
cell: Dict[str, Any],
tasks: List[Dict[str, Any]],
t_start: float,
n_processed: int = -1,
) -> None:
results_path = out_dir / "results.jsonl"
rows = [
@@ -326,8 +433,29 @@ def _write_summary(
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)
total_web_searches = sum(int(r.get("web_search_uses", 0) or 0) for r in rows)
elapsed = time.time() - t_start
# Preserve prior wall_time_s on no-op resume so we don't clobber the
# original run's runtime. A resume that did zero work (everything was
# already cached in results.jsonl) records ~seconds of elapsed time;
# writing that as wall_time_s makes the cell look 20× faster than it
# really was. If we did process anything this session, accumulate so
# partial-resumes still report total wall time honestly.
summary_path = out_dir / "summary.json"
prior_wall = 0.0
if summary_path.exists():
try:
prior_wall = float(
json.loads(summary_path.read_text()).get("wall_time_s", 0.0) or 0.0
)
except Exception:
prior_wall = 0.0
if n_processed == 0 and prior_wall > 0:
wall = prior_wall
else:
wall = prior_wall + elapsed
summary = {
"cell": cell_name,
"method": cell["method"],
@@ -338,14 +466,16 @@ def _write_summary(
"accuracy": acc,
"tokens_local_total": total_local,
"tokens_cloud_total": total_cloud,
"web_search_uses_total": total_web_searches,
"cost_usd_total": total_cost,
"wall_time_s": elapsed,
"wall_time_s": wall,
"task_count": len(tasks),
}
(out_dir / "summary.json").write_text(json.dumps(summary, indent=2))
summary_path.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",
f"acc={acc:.3f} cost=${total_cost:.2f} time={wall/60:.1f}m "
f"(session +{elapsed/60:.1f}m, processed={n_processed})",
flush=True,
)
@@ -398,8 +528,21 @@ def _run_cell_locked(
flush=True,
)
tasks = load_tasks(cell["bench"], n=cell["n"])
print(f"[load] {cell['bench']} → {len(tasks)} tasks", flush=True)
subset_path = cell.get("subset")
if subset_path:
subset = _load_subset_file(subset_path)
# Load the full bench (n=None) so we can pick the exact subset IDs.
# The dataset providers are cached, so this isn't a re-fetch.
all_tasks = load_tasks(cell["bench"], n=None)
tasks = _apply_subset(all_tasks, subset, cell)
print(
f"[load] {cell['bench']} subset={Path(subset_path).name} "
f"→ {len(tasks)} tasks",
flush=True,
)
else:
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)))
@@ -440,7 +583,7 @@ def _run_cell_locked(
for fut in as_completed(futures):
fut.result()
_write_summary(out_dir, cell_name, cell, tasks, t_start)
_write_summary(out_dir, cell_name, cell, tasks, t_start, n_processed=len(pending))
# ---------- CLI ----------
src/openjarvis/agents/hybrid/skillorchestra.py
+32 -6
View File
@@ -33,7 +33,12 @@ import json
from typing import Any, Dict, List, Optional, Tuple
from openjarvis.agents._stubs import AgentContext
from openjarvis.agents.hybrid._base import LocalCloudAgent
from openjarvis.agents.hybrid._base import (
WEB_SEARCH_COST_PER_CALL,
LocalCloudAgent,
build_web_search_tool,
web_search_cfg,
)
from openjarvis.agents.hybrid._prices import supports_temperature
from openjarvis.agents.hybrid.mini_swe_agent import run_swe_agent_loop
from openjarvis.core.registry import AgentRegistry
@@ -454,6 +459,7 @@ class SkillOrchestraAgent(LocalCloudAgent):
tokens_local = 0
tokens_cloud = r_in + r_out
run_cost = self.cost_usd(router_model, r_in, r_out)
self._web_search_uses_last = 0
# 2. Execute via chosen agent
task_meta = (context.metadata.get("task") if context is not None else {}) or {}
@@ -515,13 +521,31 @@ class SkillOrchestraAgent(LocalCloudAgent):
tokens_cloud += out["tokens_in"] + out["tokens_out"]
run_cost += out["cost_usd"]
else:
ans, w_in, w_out = self._call_cloud(
user=question,
max_tokens=int(cfg.get("cloud_max_tokens", 4096)),
temperature=0.0,
)
# GAIA cloud worker: opt-in native web_search via the new
# method_cfg.web_search schema. Only fires when the chosen
# worker is on Anthropic (server-side tool).
ws_enabled, ws_max_uses = web_search_cfg(cfg)
gaia_max_turns = int(cfg.get("gaia_max_turns", 8))
n_searches = 0
if ws_enabled and self._cloud_endpoint == "anthropic":
ans, w_in, w_out, n_searches, _ = self._call_anthropic_agent(
self._cloud_model,
user=question,
max_tokens=int(cfg.get("cloud_max_tokens", 4096)),
temperature=0.0,
tools=[build_web_search_tool(ws_max_uses)],
max_turns=gaia_max_turns,
)
else:
ans, w_in, w_out = self._call_cloud(
user=question,
max_tokens=int(cfg.get("cloud_max_tokens", 4096)),
temperature=0.0,
)
tokens_cloud += w_in + w_out
run_cost += self.cost_usd(self._cloud_model, w_in, w_out)
run_cost += n_searches * WEB_SEARCH_COST_PER_CALL
self._web_search_uses_last = n_searches
worker_model = self._cloud_model
meta = {
@@ -529,12 +553,14 @@ class SkillOrchestraAgent(LocalCloudAgent):
"tokens_cloud": tokens_cloud,
"cost_usd": run_cost,
"turns": 2, # router + worker
"web_search_uses": self._web_search_uses_last,
"traces": {
"chosen_agent": chosen,
"worker_model": worker_model,
"skill_weights": skill_weights,
"agent_scores": scored,
"reasoning": decision.get("reasoning", ""),
"n_web_searches": self._web_search_uses_last,
},
}
return ans, meta
tests/agents/hybrid/test_web_search_wiring.py
+196
View File
@@ -0,0 +1,196 @@
"""Tests for the GAIA web_search wiring (2026-05-17).
Confirms that when a GAIA cell sets ``method_cfg.web_search.enabled = true``
and the cloud endpoint is Anthropic, every paradigm declares the native
``web_search_20250305`` server-side tool on its Anthropic call. Uses mocks
— no real API calls.
Also confirms the default (``web_search`` absent / disabled) stays
one-shot and does NOT declare the tool, preserving back-compat with the
currently running n=100 cells.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from openjarvis.agents._stubs import AgentContext
from openjarvis.agents.hybrid._base import (
LocalCloudAgent,
build_web_search_tool,
web_search_cfg,
)
# ---------------------------------------------------------------------------
# 1. Schema parser
# ---------------------------------------------------------------------------
def test_web_search_cfg_default_off() -> None:
assert web_search_cfg(None) == (False, 8)
assert web_search_cfg({}) == (False, 8)
assert web_search_cfg({"web_search": None}) == (False, 8)
def test_web_search_cfg_enabled() -> None:
assert web_search_cfg({"web_search": {"enabled": True}}) == (True, 8)
assert web_search_cfg({"web_search": {"enabled": True, "max_uses": 3}}) == (True, 3)
assert web_search_cfg({"web_search": {"enabled": False, "max_uses": 5}}) == (False, 5)
def test_build_web_search_tool_shape() -> None:
tool = build_web_search_tool(5)
assert tool == {
"type": "web_search_20250305",
"name": "web_search",
"max_uses": 5,
}
# ---------------------------------------------------------------------------
# 2. Baseline cloud GAIA cell declares the tool when opted in
# ---------------------------------------------------------------------------
def _fake_anthropic_response(
text: str = "FINAL ANSWER: 42",
n_searches: int = 2,
input_tokens: int = 100,
output_tokens: int = 50,
):
"""Build a minimal Anthropic-shaped response object."""
return SimpleNamespace(
content=[SimpleNamespace(
type="text", text=text,
# the serializer probes a fixed attr list; only `text` is
# needed for the text block path.
)],
usage=SimpleNamespace(
input_tokens=input_tokens,
output_tokens=output_tokens,
server_tool_use=SimpleNamespace(web_search_requests=n_searches),
),
stop_reason="end_turn",
)
class _FakeMessages:
"""Captures the create() kwargs so the test can assert what was sent."""
def __init__(self):
self.last_kwargs = None
self.responses = [_fake_anthropic_response()]
self.calls = 0
def create(self, **kwargs):
self.last_kwargs = kwargs
idx = min(self.calls, len(self.responses) - 1)
self.calls += 1
return self.responses[idx]
class _FakeAnthropic:
"""anthropic.Anthropic() stub. Stores messages on a class attr so the
test can pull it out after the agent's call."""
last_messages = None
def __init__(self, *args, **kwargs):
self.messages = _FakeMessages()
type(self).last_messages = self.messages
@pytest.fixture
def fake_anthropic(monkeypatch):
"""Patch anthropic.Anthropic at the SDK level so every paradigm's
raw client construction picks up the fake. Importing the real
library is fine — the agent uses ``anthropic.Anthropic()`` only.
"""
import anthropic
monkeypatch.setattr(anthropic, "Anthropic", _FakeAnthropic)
yield _FakeAnthropic
def test_baseline_cloud_declares_web_search_on_gaia_when_enabled(fake_anthropic):
from openjarvis.agents.hybrid.baseline_cloud import BaselineCloudAgent
agent = BaselineCloudAgent(
engine=None,
model="claude-opus-4-7",
cloud_endpoint="anthropic",
cfg={
"cloud_max_tokens": 1024,
"web_search": {"enabled": True, "max_uses": 4},
"gaia_max_turns": 2,
},
)
ctx = AgentContext(metadata={
"task": {"task_id": "t1", "question": "What is X?"},
"task_id": "t1",
})
result = agent.run("What is X?", ctx)
sent = fake_anthropic.last_messages.last_kwargs
assert sent is not None, "Anthropic client was never called"
assert "tools" in sent, "web_search tool was not declared"
assert sent["tools"] == [
{"type": "web_search_20250305", "name": "web_search", "max_uses": 4}
]
# Per-row web_search_uses surfaced through meta.
assert result.metadata["web_search_uses"] == 2
def test_baseline_cloud_does_not_declare_tool_when_disabled(fake_anthropic):
from openjarvis.agents.hybrid.baseline_cloud import BaselineCloudAgent
agent = BaselineCloudAgent(
engine=None,
model="claude-opus-4-7",
cloud_endpoint="anthropic",
cfg={"cloud_max_tokens": 1024}, # no web_search block — default OFF
)
ctx = AgentContext(metadata={
"task": {"task_id": "t2", "question": "What is X?"},
"task_id": "t2",
})
agent.run("What is X?", ctx)
sent = fake_anthropic.last_messages.last_kwargs
assert sent is not None
assert "tools" not in sent, (
"Default behavior MUST be one-shot with no tools declared — "
"tools must only appear when method_cfg.web_search.enabled = true."
)
def test_baseline_cloud_web_search_skipped_on_non_anthropic(monkeypatch):
"""Non-Anthropic endpoint with web_search.enabled=true must not crash
and must not invoke a fake local web_search — it falls back to the
one-shot path (logged as web_search_skipped)."""
from openjarvis.agents.hybrid import baseline_cloud as bc_mod
from openjarvis.agents.hybrid.baseline_cloud import BaselineCloudAgent
# Stub the cloud call so we don't hit OpenAI.
def fake_call_cloud(self, *, user, system=None, max_tokens=4096,
temperature=0.0, **kwargs):
return "FINAL ANSWER: x", 10, 5
monkeypatch.setattr(
bc_mod.BaselineCloudAgent, "_call_cloud", fake_call_cloud, raising=True,
)
agent = BaselineCloudAgent(
engine=None,
model="gpt-5",
cloud_endpoint="openai",
cfg={"cloud_max_tokens": 1024, "web_search": {"enabled": True}},
)
ctx = AgentContext(metadata={
"task": {"task_id": "t3", "question": "X?"},
"task_id": "t3",
})
res = agent.run("X?", ctx)
# Falls through to one-shot path; no crash, no fake web_search.
assert res.metadata["web_search_uses"] == 0