feat(evals): ToolCall-15, LiveCodeBench, LiveResearchBench + telemetry (#169)

* fix(evals): add tool_choice=auto + fix traces thread safety

Two fixes for eval accuracy and stability:

1. TauBench agent: add tool_choice="auto" to match tau2's native
   LLMAgent behavior. Without this, Qwen and GPT-5.4 score 10-14pp
   below leaderboard because the models don't receive explicit
   tool-calling guidance.

2. SystemBuilder: apply self._traces flag to config.traces.enabled.
   Previously builder.traces(False) was a no-op — traces stayed
   enabled, creating SQLite connections in the main thread that
   crashed when accessed from ThreadPoolExecutor worker threads
   in GAIA evals ("SQLite objects created in a thread can only be
   used in that same thread").

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat: enrich inference events with model response content and add Trace.messages

Add content, tool_calls, and finish_reason fields to INFERENCE_END events
published by InstrumentedEngine. For non-instrumented engines, BaseAgent._generate()
now publishes INFERENCE_START/END events with the same rich data. Add _message_to_dict
helper and messages field to Trace dataclass for full conversation capture.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat: enhance TraceCollector with rich content, tool details, and messages

Capture model response content, tool_calls, and finish_reason in GENERATE
steps; store tool arguments and result text in TOOL_CALL steps; extract
conversation messages from AgentResult.metadata into Trace.messages; and
implement the last_trace property.  Also adds a messages column to the
TraceStore schema so messages survive the SQLite round-trip.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat: agents store conversation messages in AgentResult.metadata

Both NativeReActAgent and MonitorOperativeAgent now serialize their
internal messages list via _message_to_dict and include it in the
returned AgentResult.metadata under the "messages" key. This enables
TraceCollector to capture full conversation traces.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat: wire TraceCollector into system._run_agent and JarvisAgentBackend

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat: persist rich trace data in eval trace JSONL files

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat: add TerminalBench config for Qwen 3.5-122B

* fix: add SQLite migration for traces.messages column on existing databases

* fix: check trace_store instead of shared config for trace enablement

* feat(evals): add ToolCall-15, LiveCodeBench, LiveResearchBench + telemetry

Three new benchmark integrations and full telemetry wiring for the
NeurIPS 2026 IPW/IPJ experiments.

## New Benchmarks

### ToolCall-15
15-scenario tool calling accuracy benchmark across 5 categories.
All scenarios defined inline with deterministic scoring (0/1/2 per
scenario). Fast to run (~5min/model) — ideal for optimization loops.

### LiveCodeBench
Competitive programming from LeetCode/AtCoder/CodeForces via
HuggingFace dataset. Sandboxed code execution with per-test
timeouts. Single-turn generation via jarvis-direct backend.

### LiveResearchBench
100 expert-curated deep research tasks. LLM-as-judge scoring
across 4 dimensions (comprehensiveness, insight, instruction
following, readability). Uses web_search tool for live research.

## Telemetry Wiring

- FLOPs estimation: 2 * active_params * total_tokens (MoE-aware)
- Energy/power capture flows from InstrumentedEngine through
  backends to EvalResult and RunSummary
- New telemetry_summary section in output JSON with IPW/IPJ
- JarvisDirectBackend now propagates gpu_metrics flag
- TauBench forwards telemetry flags to SystemBuilder

## Experiment Plan

Added docs/experiments/neurips-2026-plan.md tracking the full
experiment matrix: 9 models x 7 benchmarks across NVIDIA, AMD,
and Apple hardware stacks.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Jon Saad-Falcon <jonsaadfalcon@gmail.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Avanika Narayan
2026-04-01 21:06:20 -07:00
committed by GitHub
co-authored by Jon Saad-Falcon Claude Opus 4.6
parent 795874bf09
commit dfddd8c785
30 changed files with 3653 additions and 30 deletions
+240
View File
@@ -0,0 +1,240 @@
# NeurIPS 2026 Experiment Plan: IPW/IPJ for Local AI
## Overview
Evaluate and optimize local AI models as OpenClaw agent brains, measuring
accuracy, latency, cost, energy, and FLOPs across 7 benchmarks.
## Results Storage
All results stored under `results/neurips-2026/`:
```
results/neurips-2026/
├── baselines/ # Step 1: Raw model scores
│ ├── {model}/{benchmark}/ # e.g. qwen-9b/pinchbench/
│ │ ├── results.jsonl # Per-task results
│ │ ├── summary.json # Aggregate metrics
│ │ └── telemetry.json # Energy, power, FLOPs, tokens
│ └── ...
├── agent-optimization/ # Step 2a: Agent improvements
│ ├── gepa/ # GEPA prompt evolution results
│ │ ├── generation_{N}/ # Per-generation best prompts
│ │ └── best_configs/ # Final optimized agent configs
│ ├── dspy/ # DSPy optimization results
│ │ ├── bootstrap/ # BootstrapFewShot results
│ │ └── mipro/ # MIPROv2 results
│ └── agent-configs/ # New agent configurations tested
├── intelligence-optimization/ # Step 2b: Model improvements
│ ├── sft/ # Supervised fine-tuning
│ │ ├── qwen-2b/ # Per-model training runs
│ │ ├── qwen-9b/
│ │ └── qwen-27b/
│ ├── lora/ # LoRA fine-tuning
│ │ ├── qwen-2b/
│ │ ├── qwen-9b/
│ │ └── qwen-27b/
│ └── rl/ # Reinforcement learning (GRPO)
│ ├── qwen-2b/
│ ├── qwen-9b/
│ └── qwen-27b/
├── optimized-eval/ # Step 3: Full eval with best configs
│ ├── {model}/{benchmark}/ # Same structure as baselines/
│ └── ...
└── analysis/ # Charts, tables, comparisons
├── pareto_frontier.json # IPW/IPJ data points
├── scaling_curves.json # Accuracy vs model size
├── cost_comparison.json # Local vs cloud economics
└── figures/ # Generated plots
```
## Hardware Stacks
Run eval metrics across three hardware vendor stacks to show
platform-agnostic IPW/IPJ results:
| Stack | Server-Class | Workstation/Consumer |
|-------|-------------|---------------------|
| **NVIDIA** | DGX Spark | RTX 6000 Pro |
| **AMD** | MI300x, MI355x | — |
| **Apple** | — | Mac Mini M4, Mac Studio M4 |
Results stored per-stack under each model's directory:
```
results/neurips-2026/baselines/{model}/{benchmark}/
├── nvidia-dgxspark/
│ ├── results.jsonl
│ ├── telemetry.json # NVML energy, GPU util, power
│ └── summary.json
├── nvidia-rtx6000pro/
├── amd-mi300x/
│ ├── telemetry.json # ROCm energy, GPU util, power
│ └── ...
├── amd-mi355x/
├── apple-macmini-m4/
│ ├── telemetry.json # Apple powermetrics energy
│ └── ...
└── apple-macstudio-m4/
```
OpenJarvis telemetry already supports all three vendors:
- NVIDIA: `telemetry/nvidia_monitor.py` (NVML)
- AMD: `telemetry/amd_monitor.py` (ROCm SMI)
- Apple: `telemetry/apple_monitor.py` (powermetrics)
Key comparisons:
- Same model, same benchmark, different hardware → IPW/IPJ per platform
- DGX Spark vs MI300x vs Mac Studio → server-class efficiency frontier
- RTX 6000 Pro vs Mac Mini M4 → consumer/workstation efficiency frontier
- GGUF models (Kimi, MiniMax) run on all platforms via llama.cpp/MLX
## Models (9 priority + 3 cloud baselines)
### Cloud Baselines
| ID | Model | Engine |
|----|-------|--------|
| claude-opus | Claude Opus 4.6 | cloud |
| gpt-54 | GPT-5.4 | cloud |
| gemini-31-pro | Gemini 3.1 Pro | cloud |
### Priority Local Models
| ID | Model | Active Params | Serving | Hardware |
|----|-------|---------------|---------|----------|
| qwen-397b | Qwen3.5-397B-A17B-FP8 | 17B | vLLM | 8x H100 |
| qwen-27b | Qwen3.5-27B-FP8 | 27B | vLLM | 1-2x H100 |
| qwen-9b | Qwen3.5-9B | 9B | vLLM/Ollama | 1x GPU |
| qwen-2b | Qwen3.5-2B | 2B | Ollama | laptop |
| trinity-large | Trinity-Large-Thinking | 13B | vLLM | 4-8x H100 |
| nemotron-nano | Nemotron-3-Nano-30B-A3B | 3B | vLLM | 1x GPU |
| kimi-k25 | Kimi-K2.5 (GGUF) | ~32B | llama.cpp | 2x GPU |
| minimax-m25 | MiniMax-M2.5 (GGUF) | ~45B | llama.cpp | 2-4x GPU |
| lfm-1.2b | LFM2.5-1.2B-Instruct | 1.2B | llama.cpp | CPU |
## Benchmarks (7)
| ID | Benchmark | Tasks | Fast Subset | Status |
|----|-----------|-------|-------------|--------|
| pinchbench | PinchBench | 23 | 23 (all) | Implemented |
| taubench | TauBench V2 | 60+40 | 20 A+R | Implemented |
| gaia | GAIA | 50 | 20 | Implemented |
| terminalbench | TerminalBench | varies | 20 | Implemented |
| toolcall15 | ToolCall-15 | 15 | 15 (all) | TODO |
| livecodebench | LiveCodeBench | ~100 | 20 | TODO |
| liveresearch | LiveResearchBench | 100 | 10 | TODO |
## Metrics Captured Per Run
- accuracy (benchmark-specific)
- latency_seconds (wall clock per task)
- energy_joules (RAPL + NVML)
- power_watts (average during inference)
- cost_usd (API cost for cloud, amortized HW for local)
- prompt_tokens, completion_tokens
- tool_calls_count
- flops_estimated (2 * active_params * total_tokens)
- gpu_utilization_pct
- throughput_tok_per_sec
---
## Step 1: Baseline Sweep
### Phase 1a: Implement missing benchmarks
- [ ] ToolCall-15 integration
- [ ] LiveCodeBench integration
- [ ] LiveResearchBench integration
- [ ] Wire telemetry capture to all eval runs
### Phase 1b: Run cloud baselines (no GPU needed)
- [x] Claude Opus — PinchBench (95.65%), TauBench A+R (86.67%),
TauBench Telecom (75%), GAIA (66.67%)
- [x] GPT-5.4 — PinchBench (52-65%), TauBench A+R (81.67%),
TauBench Telecom (75%), GAIA (34.29%)
- [x] Gemini 3.1 Pro — PinchBench (78.26%), TauBench A+R (58.33%),
TauBench Telecom (77.5%), GAIA (47.06%)
- [ ] All 3 cloud baselines — ToolCall-15, LiveCodeBench, LiveResearchBench
- [ ] All 3 cloud baselines — TerminalBench
### Phase 1c: Run local models (GPU required)
- [x] Qwen-397B — PinchBench (78.26%), TauBench A+R (81.67%)
- [x] Qwen-122B — PinchBench (73.91%), TauBench A+R (80%)
- [x] Qwen-35B — PinchBench (73.91%), TauBench A+R (77.27%)
- [x] Nemotron-Super — PinchBench (78.26%), TauBench A+R (86.67%),
TauBench Telecom (70%), GAIA (48.48%)
- [ ] Qwen-27B — all 7 benchmarks
- [ ] Qwen-9B — all 7 benchmarks
- [ ] Qwen-2B — all 7 benchmarks
- [ ] Trinity-Large — all 7 benchmarks
- [ ] Nemotron-Nano — all 7 benchmarks
- [ ] Kimi-K2.5 — all 7 benchmarks
- [ ] MiniMax-M2.5 — all 7 benchmarks
- [ ] LFM-1.2B — all 7 benchmarks
### Phase 1d: Compile baseline results
- [ ] Generate Pareto frontier plots (quality vs cost, vs energy, vs FLOPs)
- [ ] Generate scaling curves (accuracy vs active params per benchmark)
- [ ] Compute IPW/IPJ for every (model, benchmark) pair
---
## Step 2: Optimization
### Phase 2a: Agent optimization
- [ ] GEPA: evolve system prompts for monitor_operative on fast benchmarks
- [ ] GEPA: evolve system prompts for native_openhands on fast benchmarks
- [ ] DSPy BootstrapFewShot: optimize few-shot examples per benchmark
- [ ] DSPy MIPROv2: optimize full prompt pipeline
- [ ] Agent architecture search: test new agent configs
- [ ] Tool selection optimization: find minimal effective tool sets
- [ ] Evaluate optimized agents on all 9 models × fast benchmarks
### Phase 2b: Intelligence optimization
Training data:
- GeneralThought-430K-filtered (reasoning traces)
- neulab/agent-data-collection (agentic traces)
- GLM-4.7-flash SFT traces (168K + 57K)
Training targets:
- [ ] Qwen-2B: full SFT on agentic traces
- [ ] Qwen-2B: LoRA on agentic traces
- [ ] Qwen-9B: full SFT on agentic traces
- [ ] Qwen-9B: LoRA on agentic traces
- [ ] Qwen-27B: LoRA on agentic traces
- [ ] Qwen-2B: GRPO RL on benchmark outcomes
- [ ] Qwen-9B: GRPO RL on benchmark outcomes
- [ ] Evaluate all trained checkpoints on fast benchmarks
---
## Step 3: Full Evaluation
- [ ] Select best Agent config from Step 2a
- [ ] Select best Intelligence checkpoints from Step 2b
- [ ] Run complete 9 × 7 matrix with optimized configs
- [ ] Compute all metrics (accuracy, latency, energy, cost, tokens, FLOPs)
- [ ] Generate final comparison tables and figures
- [ ] Write up results section
---
## Current Progress
### Completed
- PinchBench harness: fixed and validated (PR #124, #139, #140)
- TauBench V2 native integration (PR #162)
- tool_choice + SQLite fixes (PR #163)
- Gemini thought_signature support
- Nemotron SGLang serving
- 8 models evaluated on PinchBench
- 7 models evaluated on TauBench A+R
- 4 models evaluated on TauBench Telecom
- 4 models evaluated on GAIA
### In Progress
- Qwen 35B: TauBench telecom + GAIA running
- ToolCall-15 integration: TODO
- LiveCodeBench integration: TODO
- LiveResearchBench integration: TODO
- Telemetry wiring: TODO
### Blocked
- Qwen 397B telecom + GAIA: needs 8 GPUs
- Trinity-Large: not yet served
- Small models (2B, 9B): configs not yet created
- GGUF models (Kimi, MiniMax): need llama.cpp/Ollama serving setup
+25 -1
View File
@@ -151,8 +151,17 @@ class BaseAgent(ABC):
"""Call ``engine.generate()`` with stored defaults.
Extra kwargs (e.g. ``tools``) are forwarded to the engine.
Publishes INFERENCE_START/END events on the bus when the engine
does not publish its own (i.e. non-instrumented engines).
"""
return self._engine.generate(
if self._bus and not getattr(self._engine, "_publishes_events", False):
engine_id = getattr(self._engine, "engine_id", "")
self._bus.publish(
EventType.INFERENCE_START,
{"model": self._model, "engine": engine_id},
)
result = self._engine.generate(
messages,
model=self._model,
temperature=self._temperature,
@@ -160,6 +169,21 @@ class BaseAgent(ABC):
**extra_kwargs,
)
if self._bus and not getattr(self._engine, "_publishes_events", False):
usage = result.get("usage", {})
self._bus.publish(
EventType.INFERENCE_END,
{
"model": self._model,
"usage": usage,
"content": result.get("content", ""),
"tool_calls": result.get("tool_calls", []),
"finish_reason": result.get("finish_reason", ""),
},
)
return result
def _max_turns_result(
self,
tool_results: list[ToolResult],
+5 -3
View File
@@ -21,7 +21,7 @@ from typing import Any, List, Optional
from openjarvis.agents._stubs import AgentContext, AgentResult, ToolUsingAgent
from openjarvis.core.events import EventBus
from openjarvis.core.registry import AgentRegistry
from openjarvis.core.types import Message, Role, ToolCall, ToolResult
from openjarvis.core.types import Message, Role, ToolCall, ToolResult, _message_to_dict
from openjarvis.engine._stubs import InferenceEngine
from openjarvis.tools._stubs import BaseTool
@@ -314,11 +314,12 @@ class MonitorOperativeAgent(ToolUsingAgent):
else:
# Max turns exceeded
self._save_session(input, content)
msg_dicts = [_message_to_dict(m) for m in messages]
return self._max_turns_result(
all_tool_results,
turns,
content=content,
metadata=total_usage,
metadata={**total_usage, "messages": msg_dicts},
)
# 6. Save session
@@ -329,11 +330,12 @@ class MonitorOperativeAgent(ToolUsingAgent):
self._auto_persist_state(content)
self._emit_turn_end(turns=turns, content_length=len(content))
msg_dicts = [_message_to_dict(m) for m in messages]
return AgentResult(
content=content,
tool_results=all_tool_results,
turns=turns,
metadata=total_usage,
metadata={**total_usage, "messages": msg_dicts},
)
# ------------------------------------------------------------------
+10 -4
View File
@@ -12,7 +12,7 @@ from typing import Any, List, Optional
from openjarvis.agents._stubs import AgentContext, AgentResult, ToolUsingAgent
from openjarvis.core.events import EventBus
from openjarvis.core.registry import AgentRegistry
from openjarvis.core.types import Message, Role, ToolCall, ToolResult
from openjarvis.core.types import Message, Role, ToolCall, ToolResult, _message_to_dict
from openjarvis.engine._stubs import InferenceEngine
from openjarvis.tools._stubs import BaseTool, build_tool_descriptions
@@ -140,21 +140,23 @@ class NativeReActAgent(ToolUsingAgent):
# Final answer?
if parsed["final_answer"]:
self._emit_turn_end(turns=turns)
msg_dicts = [_message_to_dict(m) for m in messages]
return AgentResult(
content=parsed["final_answer"],
tool_results=all_tool_results,
turns=turns,
metadata=total_usage,
metadata={**total_usage, "messages": msg_dicts},
)
# No action? Treat content as final answer
if not parsed["action"]:
self._emit_turn_end(turns=turns)
msg_dicts = [_message_to_dict(m) for m in messages]
return AgentResult(
content=content,
tool_results=all_tool_results,
turns=turns,
metadata=total_usage,
metadata={**total_usage, "messages": msg_dicts},
)
# Execute action
@@ -190,7 +192,11 @@ class NativeReActAgent(ToolUsingAgent):
messages.append(Message(role=Role.USER, content=observation))
# Max turns exceeded
return self._max_turns_result(all_tool_results, turns, metadata=total_usage)
msg_dicts = [_message_to_dict(m) for m in messages]
return self._max_turns_result(
all_tool_results, turns,
metadata={**total_usage, "messages": msg_dicts},
)
__all__ = ["NativeReActAgent", "REACT_SYSTEM_PROMPT"]
+17
View File
@@ -179,6 +179,21 @@ def _trace_id() -> str:
return uuid.uuid4().hex[:16]
def _message_to_dict(msg: "Message") -> Dict[str, Any]:
"""Serialize a Message to a JSON-safe dict."""
d: Dict[str, Any] = {"role": msg.role.value, "content": msg.content}
if msg.name:
d["name"] = msg.name
if msg.tool_calls:
d["tool_calls"] = [
{"id": tc.id, "name": tc.name, "arguments": tc.arguments}
for tc in msg.tool_calls
]
if msg.tool_call_id:
d["tool_call_id"] = msg.tool_call_id
return d
@dataclass(slots=True)
class TraceStep:
"""A single step within an agent trace.
@@ -220,6 +235,7 @@ class Trace:
total_tokens: int = 0
total_latency_seconds: float = 0.0
metadata: Dict[str, Any] = field(default_factory=dict)
messages: List[Dict[str, Any]] = field(default_factory=list)
def add_step(self, step: TraceStep) -> None:
"""Append a step and update running totals."""
@@ -257,4 +273,5 @@ __all__ = [
"ToolResult",
"Trace",
"TraceStep",
"_message_to_dict",
]
+30 -1
View File
@@ -45,7 +45,7 @@ class JarvisAgentBackend(InferenceBackend):
# creates a GpuMonitor when building the InstrumentedEngine.
if gpu_metrics:
builder._config.telemetry.gpu_metrics = True
self._system = builder.telemetry(telemetry).traces(telemetry).build()
self._system = builder.telemetry(telemetry).traces(True).build()
def generate(
self,
@@ -86,6 +86,34 @@ class JarvisAgentBackend(InferenceBackend):
result = self._system.ask(prompt, **ask_kwargs)
elapsed = time.monotonic() - t0
# Extract trace data from the TraceCollector if available
trace_data = None
collector = getattr(self._system, "trace_collector", None)
if collector is not None:
trace = getattr(collector, "last_trace", None)
if trace is not None:
trace_data = {
"trace_id": trace.trace_id,
"steps": [
{
"step_type": (
step.step_type.value
if hasattr(step.step_type, "value")
else step.step_type
),
"timestamp": step.timestamp,
"duration_seconds": step.duration_seconds,
"input": step.input,
"output": step.output,
"metadata": step.metadata,
}
for step in trace.steps
],
"messages": trace.messages,
"total_tokens": trace.total_tokens,
"total_latency_seconds": trace.total_latency_seconds,
}
usage = result.get("usage", {})
telemetry_data = result.get("_telemetry", {})
return {
@@ -101,6 +129,7 @@ class JarvisAgentBackend(InferenceBackend):
"power_watts": telemetry_data.get("power_watts", 0.0),
"gpu_utilization_pct": telemetry_data.get("gpu_utilization_pct", 0.0),
"throughput_tok_per_sec": telemetry_data.get("throughput_tok_per_sec", 0.0),
"trace_data": trace_data,
}
def set_task_metadata(self, metadata: dict) -> None:
@@ -31,6 +31,10 @@ class JarvisDirectBackend(InferenceBackend):
builder = SystemBuilder()
if engine_key:
builder.engine(engine_key)
# Propagate gpu_metrics to the runtime config so SystemBuilder
# creates an EnergyMonitor / GpuMonitor for the InstrumentedEngine.
if gpu_metrics:
builder._config.telemetry.gpu_metrics = True
self._system = builder.telemetry(telemetry).traces(telemetry).build()
def generate(
+38
View File
@@ -128,6 +128,18 @@ BENCHMARKS = {
"category": "agentic",
"description": "TauBench multi-turn customer service",
},
"livecodebench": {
"category": "coding",
"description": "LiveCodeBench competitive programming",
},
"liveresearch": {
"category": "agentic",
"description": "LiveResearchBench deep research tasks",
},
"toolcall15": {
"category": "agentic",
"description": "ToolCall-15 tool calling benchmark",
},
}
BACKENDS = {
@@ -316,6 +328,18 @@ def _build_dataset(benchmark: str, subset: str | None = None):
from openjarvis.evals.datasets.taubench import TauBenchDataset
domains = subset.split(",") if subset else None
return TauBenchDataset(domains=domains)
elif benchmark == "livecodebench":
from openjarvis.evals.datasets.livecodebench import LiveCodeBenchDataset
return LiveCodeBenchDataset()
elif benchmark == "liveresearch":
from openjarvis.evals.datasets.liveresearch import LiveResearchBenchDataset
return LiveResearchBenchDataset(path=subset)
elif benchmark == "toolcall15":
from openjarvis.evals.datasets.toolcall15 import ToolCall15Dataset
return ToolCall15Dataset()
else:
raise click.ClickException(f"Unknown benchmark: {benchmark}")
@@ -455,6 +479,18 @@ def _build_scorer(benchmark: str, judge_backend, judge_model: str):
elif benchmark == "taubench":
from openjarvis.evals.scorers.taubench import TauBenchScorer
return TauBenchScorer(judge_backend, judge_model)
elif benchmark == "livecodebench":
from openjarvis.evals.scorers.livecodebench import LiveCodeBenchScorer
return LiveCodeBenchScorer(judge_backend, judge_model)
elif benchmark == "liveresearch":
from openjarvis.evals.scorers.liveresearch import LiveResearchBenchScorer
return LiveResearchBenchScorer(judge_backend, judge_model)
elif benchmark == "toolcall15":
from openjarvis.evals.scorers.toolcall15 import ToolCall15Scorer
return ToolCall15Scorer(judge_backend, judge_model)
else:
raise click.ClickException(f"Unknown benchmark: {benchmark}")
@@ -563,6 +599,8 @@ def _run_single(config, console: Optional[Console] = None) -> object:
model=config.model,
temperature=config.temperature,
max_tokens=config.max_tokens,
telemetry=getattr(config, "telemetry", False),
gpu_metrics=getattr(config, "gpu_metrics", False),
)
judge_engine = getattr(config, "judge_engine", "cloud") or "cloud"
judge_backend = _build_judge_backend(config.judge_model, engine_key=judge_engine)
@@ -0,0 +1,30 @@
# LiveCodeBench evaluation with Claude Opus
# Benchmark: competitive programming from LeetCode, AtCoder, CodeForces
# Reference: https://livecodebench.github.io/
[meta]
name = "livecodebench-claude-opus"
description = "LiveCodeBench code generation with Claude Opus"
[defaults]
temperature = 0.0
max_tokens = 4096
[judge]
model = "gpt-5-mini-2025-08-07"
temperature = 0.0
max_tokens = 1024
[run]
max_workers = 8
output_dir = "results/livecodebench-claude-opus/"
seed = 42
[[models]]
name = "anthropic/claude-opus-4-5"
provider = "anthropic"
[[benchmarks]]
name = "livecodebench"
backend = "jarvis-direct"
max_samples = 100
@@ -0,0 +1,29 @@
# LiveResearchBench: claude-opus-4-6
[meta]
name = "liveresearch-claude-opus"
description = "LiveResearchBench deep research benchmark on claude-opus-4-6"
[defaults]
temperature = 0.6
max_tokens = 16384
[judge]
model = "gpt-5-mini-2025-08-07"
temperature = 0.0
max_tokens = 4096
[run]
max_workers = 1
output_dir = "results/liveresearch-claude-opus/"
seed = 42
[[models]]
name = "claude-opus-4-6"
engine = "cloud"
[[benchmarks]]
name = "liveresearch"
backend = "jarvis-agent"
agent = "monitor_operative"
max_samples = 50
tools = ["web_search", "file_read", "file_write", "code_interpreter", "think"]
@@ -0,0 +1,30 @@
# TerminalBench on Qwen/Qwen3.5-122B-A10B-FP8
[meta]
name = "terminalbench-qwen122b"
description = "TerminalBench agentic benchmark on Qwen/Qwen3.5-122B-A10B-FP8"
[defaults]
temperature = 0.6
max_tokens = 8192
[judge]
model = "gpt-5-mini-2025-08-07"
temperature = 0.0
max_tokens = 4096
[run]
max_workers = 1
output_dir = "results/terminalbench-qwen122b/"
seed = 42
[[models]]
name = "Qwen/Qwen3.5-122B-A10B-FP8"
engine = "vllm"
num_gpus = 4
[[benchmarks]]
name = "terminalbench"
backend = "jarvis-agent"
agent = "monitor_operative"
max_samples = 50
tools = ["think", "shell_exec", "file_read", "file_write", "code_interpreter"]
@@ -0,0 +1,28 @@
# ToolCall-15 eval: Claude Opus 4.6 (cloud)
# Lightweight tool calling benchmark — 15 scenarios, 5 categories
[meta]
name = "toolcall15-claude-opus"
description = "ToolCall-15 on claude-opus-4-6 (temperature=0)"
[defaults]
temperature = 0.0
max_tokens = 2048
[judge]
model = "gpt-5-mini-2025-08-07"
temperature = 0.0
engine = "cloud"
[run]
max_workers = 1
output_dir = "results/toolcall15-claude-opus/"
seed = 42
[[models]]
name = "claude-opus-4-6"
engine = "cloud"
[[benchmarks]]
name = "toolcall15"
backend = "jarvis-direct"
+3
View File
@@ -62,6 +62,9 @@ KNOWN_BENCHMARKS = {
"browser_assistant",
"pinchbench",
"taubench",
"livecodebench",
"liveresearch",
"toolcall15",
}
+57 -1
View File
@@ -43,6 +43,11 @@ try:
except ImportError: # pragma: no cover
compute_efficiency = None # type: ignore[assignment]
try:
from openjarvis.telemetry.efficiency import estimate_model_flops_per_token
except ImportError: # pragma: no cover
estimate_model_flops_per_token = None # type: ignore[assignment]
LOGGER = logging.getLogger(__name__)
_THINK_TAG_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
@@ -346,6 +351,21 @@ class EvalRunner:
mfu = eff.mfu_pct
mbu = eff.mbu_pct
# Estimate FLOPs: 2 * active_params * total_tokens
estimated_flops = 0.0
if estimate_model_flops_per_token is not None:
model_meta = cfg.metadata or {}
param_b = model_meta.get("param_count_b", 0.0)
active_b = model_meta.get("active_params_b")
total_tokens = usage.get("prompt_tokens", 0) + usage.get(
"completion_tokens", 0
)
if param_b > 0 and total_tokens > 0:
flops_per_tok = estimate_model_flops_per_token(
param_b, active_b
)
estimated_flops = flops_per_tok * total_tokens
# Extract derived and ITL metrics from _telemetry dict
_telem = full.get("_telemetry", {})
energy_per_out_tok = _telem.get("energy_per_output_token_joules", 0.0)
@@ -374,6 +394,8 @@ class EvalRunner:
energy_per_output_token_joules=energy_per_out_tok,
throughput_per_watt=throughput_per_w,
mean_itl_ms=mean_itl,
estimated_flops=estimated_flops,
trace_data=full.get("trace_data"),
)
except Exception as exc:
LOGGER.error("Error processing %s: %s", record.record_id, exc)
@@ -655,6 +677,7 @@ class EvalRunner:
completion_tokens=total_completion_tokens,
cost_usd=total_cost,
scoring_metadata=scoring_meta,
trace_data=full.get("trace_data"),
)
except Exception as exc:
LOGGER.error(
@@ -726,6 +749,7 @@ class EvalRunner:
"energy_per_output_token_joules": result.energy_per_output_token_joules,
"throughput_per_watt": result.throughput_per_watt,
"mean_itl_ms": result.mean_itl_ms,
"estimated_flops": result.estimated_flops,
}
try:
line = json.dumps(record_dict, default=str)
@@ -834,12 +858,14 @@ class EvalRunner:
]
tpw_vals = [r.throughput_per_watt for r in results if r.throughput_per_watt > 0]
itl_vals = [r.mean_itl_ms for r in results if r.mean_itl_ms > 0]
flops_vals = [r.estimated_flops for r in results if r.estimated_flops > 0]
input_tok_vals = [r.prompt_tokens for r in results if r.prompt_tokens > 0]
output_tok_vals = [
r.completion_tokens for r in results if r.completion_tokens > 0
]
total_energy = sum(r.energy_joules for r in results)
total_estimated_flops = sum(r.estimated_flops for r in results)
total_input_tokens = sum(r.prompt_tokens for r in results)
total_output_tokens = sum(r.completion_tokens for r in results)
avg_power = statistics.mean(power_vals) if power_vals else 0.0
@@ -849,6 +875,7 @@ class EvalRunner:
"accuracy": round(accuracy, 4),
"total_energy_joules": round(total_energy, 6),
"avg_power_watts": round(avg_power, 4),
"total_estimated_flops": total_estimated_flops,
"ipj": (round(accuracy / total_energy, 6) if total_energy > 0 else None),
"ipw": (round(accuracy / avg_power, 6) if avg_power > 0 else None),
}
@@ -891,6 +918,8 @@ class EvalRunner:
input_token_stats=_metric_stats([float(v) for v in input_tok_vals]),
output_token_stats=_metric_stats([float(v) for v in output_tok_vals]),
total_energy_joules=round(total_energy, 6),
total_estimated_flops=total_estimated_flops,
flops_stats=_metric_stats(flops_vals),
warmup_samples_excluded=cfg.warmup_samples,
avg_power_watts=round(avg_power, 4),
total_input_tokens=total_input_tokens,
@@ -1044,6 +1073,8 @@ def _summary_to_dict(s: RunSummary) -> Dict[str, Any]:
"input_token_stats": _metric_stats_to_dict(s.input_token_stats),
"output_token_stats": _metric_stats_to_dict(s.output_token_stats),
"total_energy_joules": s.total_energy_joules,
"total_estimated_flops": s.total_estimated_flops,
"flops_stats": _metric_stats_to_dict(s.flops_stats),
"warmup_samples_excluded": s.warmup_samples_excluded,
"steady_state_reached": s.steady_state_reached,
"energy_method": s.energy_method,
@@ -1053,12 +1084,33 @@ def _summary_to_dict(s: RunSummary) -> Dict[str, Any]:
"efficiency": s.efficiency,
"normalized_statistics": s.normalized_statistics,
"normalized_efficiency": s.normalized_efficiency,
"telemetry_summary": {
"total_energy_joules": s.total_energy_joules,
"avg_power_watts": s.avg_power_watts,
"total_input_tokens": s.total_input_tokens,
"total_output_tokens": s.total_output_tokens,
"total_tokens": s.total_input_tokens + s.total_output_tokens,
"total_estimated_flops": s.total_estimated_flops,
"throughput_stats": _metric_stats_to_dict(s.throughput_stats),
"gpu_utilization_stats": _metric_stats_to_dict(
s.gpu_utilization_stats,
),
"energy_stats": _metric_stats_to_dict(s.energy_stats),
"power_stats": _metric_stats_to_dict(s.power_stats),
"flops_stats": _metric_stats_to_dict(s.flops_stats),
"ipw": (
s.efficiency.get("ipw") if s.efficiency else None
),
"ipj": (
s.efficiency.get("ipj") if s.efficiency else None
),
},
}
def _result_to_trace_dict(result: EvalResult) -> Dict[str, Any]:
"""Convert an EvalResult to a full trace dict for per-sample export."""
return {
d = {
"record_id": result.record_id,
"model_answer": result.model_answer,
"is_correct": result.is_correct,
@@ -1081,7 +1133,11 @@ def _result_to_trace_dict(result: EvalResult) -> Dict[str, Any]:
"energy_per_output_token_joules": result.energy_per_output_token_joules,
"throughput_per_watt": result.throughput_per_watt,
"mean_itl_ms": result.mean_itl_ms,
"estimated_flops": result.estimated_flops,
}
if result.trace_data is not None:
d["trace_data"] = result.trace_data
return d
__all__ = ["EvalRunner"]
+4
View File
@@ -45,8 +45,10 @@ class EvalResult:
energy_per_output_token_joules: float = 0.0
throughput_per_watt: float = 0.0
mean_itl_ms: float = 0.0
estimated_flops: float = 0.0
trace_steps: int = 0
trace_energy_joules: float = 0.0
trace_data: Optional[Dict[str, Any]] = None
@dataclass(slots=True)
@@ -133,6 +135,8 @@ class RunSummary:
input_token_stats: Optional[MetricStats] = None
output_token_stats: Optional[MetricStats] = None
total_energy_joules: float = 0.0
total_estimated_flops: float = 0.0
flops_stats: Optional[MetricStats] = None
warmup_samples_excluded: int = 0
steady_state_reached: bool = False
energy_method: str = ""
@@ -0,0 +1,283 @@
"""LiveCodeBench dataset provider — competitive programming code generation.
Loads problems from the LiveCodeBench HuggingFace dataset
(livecodebench/code_generation_lite) for evaluating code generation capability.
Reference: https://livecodebench.github.io/
https://github.com/LiveCodeBench/LiveCodeBench
"""
from __future__ import annotations
import json
import logging
import random
from typing import Any, Dict, Iterable, List, MutableMapping, Optional, Sequence
from openjarvis.evals.core.dataset import DatasetProvider
from openjarvis.evals.core.types import EvalRecord
LOGGER = logging.getLogger(__name__)
_HF_DATASET = "livecodebench/code_generation_lite"
_HF_DATASET_FULL = "livecodebench/code_generation"
_DEFAULT_SPLIT = "test"
_PROMPT_TEMPLATE = """Solve the following competitive programming problem. Write a complete Python solution that reads from stdin and writes to stdout.
## Problem
{problem_statement}
## Input Format
{input_format}
## Output Format
{output_format}
## Constraints
{constraints}
## Examples
{examples}
Write ONLY the Python code solution. Do not include any explanations."""
def _format_examples(example_inputs: list, example_outputs: list) -> str:
"""Format input/output examples for the prompt."""
parts = []
for i, (inp, out) in enumerate(zip(example_inputs, example_outputs), 1):
parts.append(f"Input {i}:\n{inp}\n\nOutput {i}:\n{out}")
return "\n\n".join(parts) if parts else "No examples provided."
def _parse_json_field(value: Any) -> Any:
"""Parse a field that might be a JSON string or already parsed."""
if isinstance(value, str):
try:
return json.loads(value)
except (json.JSONDecodeError, TypeError):
return value
return value
class LiveCodeBenchDataset(DatasetProvider):
"""LiveCodeBench competitive programming benchmark.
Loads problems from the LiveCodeBench HuggingFace dataset for
evaluating code generation capability on fresh competitive
programming problems from LeetCode, AtCoder, and CodeForces.
"""
dataset_id = "livecodebench"
dataset_name = "LiveCodeBench"
def __init__(self) -> None:
self._records: List[EvalRecord] = []
def load(
self,
*,
max_samples: Optional[int] = None,
split: Optional[str] = None,
seed: Optional[int] = None,
) -> None:
from datasets import load_dataset
use_split = split or _DEFAULT_SPLIT
# Try lite version first (smaller, faster download), fall back to full
dataset = None
for hf_path in [_HF_DATASET, _HF_DATASET_FULL]:
try:
dataset = load_dataset(hf_path, split=use_split)
LOGGER.info("Loaded LiveCodeBench from %s (%s)", hf_path, use_split)
break
except Exception as exc:
LOGGER.debug("Could not load %s: %s", hf_path, exc)
if dataset is None:
raise RuntimeError(
"Failed to load LiveCodeBench dataset. "
f"Tried: {_HF_DATASET}, {_HF_DATASET_FULL}. "
"Ensure 'datasets' is installed and you have network access."
)
rows: Sequence[MutableMapping[str, object]]
if hasattr(dataset, "to_list"):
rows = dataset.to_list()
else:
rows = list(dataset)
if seed is not None:
rng = random.Random(seed)
rows = list(rows)
rng.shuffle(rows)
if max_samples is not None:
rows = rows[:max_samples]
self._records = []
for idx, raw in enumerate(rows):
record = self._convert_row(raw, idx)
if record is not None:
self._records.append(record)
LOGGER.info("LiveCodeBench: loaded %d problems", len(self._records))
def iter_records(self) -> Iterable[EvalRecord]:
return iter(self._records)
def size(self) -> int:
return len(self._records)
def _convert_row(
self,
raw: MutableMapping[str, object],
idx: int,
) -> Optional[EvalRecord]:
# Extract problem statement — field names vary across dataset versions
problem_text = str(
raw.get("question_content")
or raw.get("problem_description")
or raw.get("question")
or raw.get("problem")
or ""
).strip()
if not problem_text:
LOGGER.debug("Skipping row %d: no problem text", idx)
return None
# Extract problem ID
problem_id = str(
raw.get("question_id")
or raw.get("task_id")
or raw.get("problem_id")
or f"lcb-{idx}"
)
# Extract title
title = str(
raw.get("question_title")
or raw.get("title")
or raw.get("name")
or ""
).strip()
# Extract difficulty
difficulty = str(
raw.get("difficulty")
or raw.get("question_difficulty")
or ""
).strip()
# Extract platform/source
platform = str(
raw.get("platform")
or raw.get("source")
or raw.get("contest_source")
or ""
).strip()
# Extract input/output format (may not always be separate fields)
input_format = str(raw.get("input_format", "")).strip()
output_format = str(raw.get("output_format", "")).strip()
constraints = str(raw.get("constraints", "")).strip()
# Extract test cases — various possible field names and formats
test_inputs = _parse_json_field(raw.get("input", raw.get("test_inputs", [])))
test_outputs = _parse_json_field(raw.get("output", raw.get("test_outputs", [])))
public_test_cases = _parse_json_field(
raw.get("public_test_cases", raw.get("sample_io", []))
)
hidden_test_cases = _parse_json_field(
raw.get("hidden_test_cases", raw.get("test_cases", []))
)
# Normalize test cases into input/output lists
all_test_inputs: List[str] = []
all_test_outputs: List[str] = []
# Process structured test cases (list of dicts with input/output keys)
for cases in [public_test_cases, hidden_test_cases]:
if isinstance(cases, list):
for case in cases:
if isinstance(case, dict):
inp = str(case.get("input", "")).strip()
out = str(case.get("output", "")).strip()
if inp or out:
all_test_inputs.append(inp)
all_test_outputs.append(out)
elif isinstance(case, str):
# JSON-encoded test case
parsed = _parse_json_field(case)
if isinstance(parsed, dict):
inp = str(parsed.get("input", "")).strip()
out = str(parsed.get("output", "")).strip()
if inp or out:
all_test_inputs.append(inp)
all_test_outputs.append(out)
# If no structured test cases, try the flat input/output lists
if not all_test_inputs and isinstance(test_inputs, list):
for inp in test_inputs:
all_test_inputs.append(str(inp).strip())
if not all_test_outputs and isinstance(test_outputs, list):
for out in test_outputs:
all_test_outputs.append(str(out).strip())
# Build example section from public test cases (first 2)
example_inputs = all_test_inputs[:2] if all_test_inputs else []
example_outputs = all_test_outputs[:2] if all_test_outputs else []
examples_str = _format_examples(example_inputs, example_outputs)
# Build prompt
problem = _PROMPT_TEMPLATE.format(
problem_statement=problem_text,
input_format=input_format or "(see problem statement)",
output_format=output_format or "(see problem statement)",
constraints=constraints or "(see problem statement)",
examples=examples_str,
)
# Starter code (if available)
starter_code = str(raw.get("starter_code", raw.get("code_stub", ""))).strip()
# Metadata for scoring
metadata: Dict[str, Any] = {
"test_inputs": all_test_inputs,
"test_outputs": all_test_outputs,
"difficulty": difficulty,
"platform": platform,
"title": title,
}
if starter_code:
metadata["starter_code"] = starter_code
# Store raw data for any fields the scorer might need
for key in ["time_limit", "memory_limit", "contest_id", "contest_date"]:
val = raw.get(key)
if val is not None:
metadata[key] = val
# Build subject from platform and difficulty
subject = f"{platform}/{difficulty}" if platform and difficulty else (
platform or difficulty or "competitive-programming"
)
return EvalRecord(
record_id=problem_id,
problem=problem,
reference="", # No single reference answer; scored via test execution
category="coding",
subject=subject,
metadata=metadata,
)
__all__ = ["LiveCodeBenchDataset"]
@@ -0,0 +1,191 @@
"""LiveResearchBench dataset provider — deep research benchmark.
Clones the deep_research_bench repo at runtime and parses query + criteria
JSONL files into EvalRecords for use with AgenticRunner.
Reference: https://github.com/Ayanami0730/deep_research_bench
Paper: https://arxiv.org/abs/2510.14240
"""
from __future__ import annotations
import json
import logging
import random
import shutil
import subprocess
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional
from openjarvis.evals.core.dataset import DatasetProvider
from openjarvis.evals.core.types import EvalRecord
LOGGER = logging.getLogger(__name__)
LIVERESEARCH_REPO = "https://github.com/Ayanami0730/deep_research_bench.git"
CACHE_DIR = Path.home() / ".cache" / "liveresearch_bench"
def _load_jsonl(path: Path) -> List[Dict[str, Any]]:
"""Load a JSONL file into a list of dicts."""
records: List[Dict[str, Any]] = []
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if line:
records.append(json.loads(line))
return records
def _build_criteria_index(
criteria_records: List[Dict[str, Any]],
) -> Dict[int, Dict[str, Any]]:
"""Index criteria records by their integer id."""
index: Dict[int, Dict[str, Any]] = {}
for rec in criteria_records:
rec_id = rec.get("id")
if rec_id is not None:
index[int(rec_id)] = rec
return index
class LiveResearchBenchDataset(DatasetProvider):
"""LiveResearchBench — deep research with 100 expert-curated tasks.
Clones Ayanami0730/deep_research_bench from GitHub (or uses a local
path) and parses query + criteria JSONL files into EvalRecords.
"""
dataset_id = "liveresearch"
dataset_name = "LiveResearchBench"
def __init__(self, path: Optional[str] = None) -> None:
self._local_path = Path(path) if path else None
self._repo_dir: Path = self._local_path or CACHE_DIR
self._records: List[EvalRecord] = []
def verify_requirements(self) -> List[str]:
issues: List[str] = []
if self._local_path is None and shutil.which("git") is None:
issues.append(
"git binary not found. Install git to clone LiveResearchBench."
)
return issues
def _ensure_repo(self) -> Path:
"""Clone the repo if not already cached. Returns repo dir."""
if self._local_path is not None:
if not self._local_path.exists():
raise FileNotFoundError(
f"LiveResearchBench path not found: {self._local_path}"
)
return self._local_path
if not self._repo_dir.exists():
LOGGER.info("Cloning LiveResearchBench from %s ...", LIVERESEARCH_REPO)
self._repo_dir.parent.mkdir(parents=True, exist_ok=True)
subprocess.run(
[
"git",
"clone",
"--depth",
"1",
LIVERESEARCH_REPO,
str(self._repo_dir),
],
check=True,
capture_output=True,
)
LOGGER.info("LiveResearchBench cloned to %s", self._repo_dir)
return self._repo_dir
def load(
self,
*,
max_samples: Optional[int] = None,
split: Optional[str] = None,
seed: Optional[int] = None,
) -> None:
repo_dir = self._ensure_repo()
# Load queries
query_path = repo_dir / "data" / "prompt_data" / "query.jsonl"
if not query_path.exists():
raise FileNotFoundError(f"Query file not found: {query_path}")
queries = _load_jsonl(query_path)
if not queries:
raise FileNotFoundError(f"No queries found in {query_path}")
# Load criteria (optional — used for rubric-based scoring)
criteria_path = repo_dir / "data" / "criteria_data" / "criteria.jsonl"
criteria_index: Dict[int, Dict[str, Any]] = {}
if criteria_path.exists():
criteria_records = _load_jsonl(criteria_path)
criteria_index = _build_criteria_index(criteria_records)
LOGGER.info(
"Loaded %d criteria records for rubric scoring", len(criteria_index)
)
# Optionally filter by language via split (e.g. split="en" or split="zh")
if split and split in ("en", "zh"):
queries = [q for q in queries if q.get("language") == split]
if seed is not None:
random.Random(seed).shuffle(queries)
if max_samples is not None:
queries = queries[:max_samples]
self._records = []
for query in queries:
q_id = query.get("id")
topic = query.get("topic", "")
language = query.get("language", "en")
prompt = query.get("prompt", "")
if not prompt:
LOGGER.warning("Skipping query %s: empty prompt", q_id)
continue
# Build the research task prompt
research_prompt = (
"You are a deep research assistant. Conduct thorough research "
"on the following topic and produce a comprehensive, well-structured "
"research report with citations and analysis.\n\n"
f"## Research Task\n\n{prompt}"
)
# Attach criteria metadata if available
criteria = criteria_index.get(int(q_id)) if q_id is not None else None
metadata: Dict[str, Any] = {
"topic": topic,
"language": language,
"original_id": q_id,
}
if criteria:
metadata["dimension_weight"] = criteria.get("dimension_weight", {})
metadata["criterions"] = criteria.get("criterions", {})
self._records.append(
EvalRecord(
record_id=f"liveresearch-{q_id or len(self._records)}",
problem=research_prompt,
reference="", # No single reference answer; rubric-based
category="agentic",
subject=topic,
metadata=metadata,
)
)
LOGGER.info("LiveResearchBench: loaded %d tasks", len(self._records))
def iter_records(self) -> Iterable[EvalRecord]:
return iter(self._records)
def size(self) -> int:
return len(self._records)
__all__ = ["LiveResearchBenchDataset"]
@@ -69,6 +69,8 @@ class TauBenchDataset(DatasetProvider):
self._max_tokens: int = 4096
self._user_model: Optional[str] = None
self._num_trials: int = 3 # pass^k: best of k trials per task
self._telemetry: bool = False
self._gpu_metrics: bool = False
def set_engine_config(
self,
@@ -78,6 +80,8 @@ class TauBenchDataset(DatasetProvider):
max_tokens: int = 4096,
user_model: Optional[str] = None,
num_trials: Optional[int] = None,
telemetry: bool = False,
gpu_metrics: bool = False,
) -> None:
"""Inject engine configuration for the agent. Called by CLI."""
if engine_key is not None:
@@ -90,6 +94,8 @@ class TauBenchDataset(DatasetProvider):
self._user_model = user_model
if num_trials is not None:
self._num_trials = num_trials
self._telemetry = telemetry
self._gpu_metrics = gpu_metrics
def verify_requirements(self) -> List[str]:
issues: List[str] = []
@@ -228,6 +234,8 @@ class TauBenchDataset(DatasetProvider):
max_tokens=self._max_tokens,
user_model=self._user_model,
num_trials=self._num_trials,
telemetry=self._telemetry,
gpu_metrics=self._gpu_metrics,
)
+670
View File
@@ -0,0 +1,670 @@
"""ToolCall-15 dataset provider — lightweight tool calling benchmark.
Provides 15 scenarios across 5 categories (3 per category) that test
whether a model can call the right tool with the right arguments.
Reference: https://github.com/stevibe/ToolCall-15
"""
from __future__ import annotations
import json
import logging
import random
from typing import Any, Dict, Iterable, List, Optional
from openjarvis.evals.core.dataset import DatasetProvider
from openjarvis.evals.core.types import EvalRecord
LOGGER = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Universal system prompt (from METHODOLOGY.md)
# ---------------------------------------------------------------------------
SYSTEM_PROMPT = (
"You are a helpful assistant with access to the tools provided.\n\n"
"Rules:\n"
"- Use a tool ONLY when necessary to fulfill the user's request.\n"
"- Answer directly from knowledge without tool calls when possible.\n"
"- If a tool call fails, explain the failure and suggest alternatives.\n"
"- Never invent information that a tool should provide."
)
# ---------------------------------------------------------------------------
# 12-tool universal toolkit (OpenAI function-calling format)
# ---------------------------------------------------------------------------
TOOLS: List[Dict[str, Any]] = [
{
"type": "function",
"function": {
"name": "web_search",
"description": "Search the web for current information",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"},
"max_results": {
"type": "integer",
"description": "Maximum results to return",
"default": 5,
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather for a specific location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City or location name",
},
"units": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"default": "celsius",
"description": "Temperature units",
},
},
"required": ["location"],
},
},
},
{
"type": "function",
"function": {
"name": "calculator",
"description": "Perform mathematical calculations",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Math expression to evaluate",
},
},
"required": ["expression"],
},
},
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "Send an email to a recipient",
"parameters": {
"type": "object",
"properties": {
"to": {
"type": "string",
"description": "Recipient email address",
},
"subject": {"type": "string", "description": "Email subject"},
"body": {"type": "string", "description": "Email body"},
"attachments": {
"type": "array",
"items": {"type": "string"},
"default": [],
"description": "File paths to attach",
},
},
"required": ["to", "subject", "body"],
},
},
},
{
"type": "function",
"function": {
"name": "search_files",
"description": "Search for files by name or content",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"},
"file_type": {
"type": "string",
"enum": ["pdf", "docx", "xlsx", "any"],
"default": "any",
"description": "File type filter",
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read the contents of a specific file",
"parameters": {
"type": "object",
"properties": {
"file_id": {
"type": "string",
"description": "File identifier",
},
},
"required": ["file_id"],
},
},
},
{
"type": "function",
"function": {
"name": "create_calendar_event",
"description": "Create a new calendar event",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Event title"},
"date": {
"type": "string",
"description": "Date in YYYY-MM-DD format",
},
"time": {
"type": "string",
"description": "Time in HH:MM format",
},
"duration_minutes": {
"type": "integer",
"default": 60,
"description": "Duration in minutes",
},
"attendees": {
"type": "array",
"items": {"type": "string"},
"default": [],
"description": "Attendee emails",
},
},
"required": ["title", "date", "time"],
},
},
},
{
"type": "function",
"function": {
"name": "get_contacts",
"description": "Look up contacts by name or group",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Name or group to search",
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "translate_text",
"description": "Translate text from one language to another",
"parameters": {
"type": "object",
"properties": {
"text": {"type": "string", "description": "Text to translate"},
"source_language": {
"type": "string",
"description": "Source language",
},
"target_language": {
"type": "string",
"description": "Target language",
},
},
"required": ["text", "source_language", "target_language"],
},
},
},
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "Get current stock price for a ticker symbol",
"parameters": {
"type": "object",
"properties": {
"ticker": {
"type": "string",
"description": "Stock ticker symbol",
},
},
"required": ["ticker"],
},
},
},
{
"type": "function",
"function": {
"name": "set_reminder",
"description": "Set a reminder for a future time",
"parameters": {
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "Reminder message",
},
"datetime": {
"type": "string",
"description": "ISO 8601 datetime for the reminder",
},
},
"required": ["message", "datetime"],
},
},
},
{
"type": "function",
"function": {
"name": "run_code",
"description": "Execute a code snippet and return output",
"parameters": {
"type": "object",
"properties": {
"language": {
"type": "string",
"enum": ["python", "javascript"],
"description": "Programming language",
},
"code": {
"type": "string",
"description": "Code to execute",
},
},
"required": ["language", "code"],
},
},
},
]
# ---------------------------------------------------------------------------
# 15 scenarios
# ---------------------------------------------------------------------------
# Categories:
# A = Tool Selection (TC-01..03)
# B = Parameter Precision (TC-04..06)
# C = Multi-Step Chains (TC-07..09)
# D = Restraint & Refusal (TC-10..12)
# E = Error Recovery (TC-13..15)
SCENARIOS: List[Dict[str, Any]] = [
# --- Category A: Tool Selection ---
{
"id": "TC-01",
"name": "Direct Specialist Match",
"category": "A-ToolSelection",
"user_message": "What's the weather like in Berlin right now?",
"mock_tool_outputs": {
"get_weather": json.dumps({
"location": "Berlin",
"temperature": 8,
"units": "celsius",
"condition": "Overcast",
"humidity": 72,
}),
"web_search": json.dumps({
"results": [
{"snippet": "Berlin weather right now: 8C and overcast."}
]
}),
},
},
{
"id": "TC-02",
"name": "Distractor Resistance",
"category": "A-ToolSelection",
"user_message": "What is the current price of AAPL stock?",
"mock_tool_outputs": {
"get_stock_price": json.dumps({
"ticker": "AAPL",
"price": 187.42,
"currency": "USD",
"change": "+1.23",
"change_percent": "+0.66%",
}),
"web_search": json.dumps({
"results": [
{"snippet": "AAPL is trading around $187.42."}
]
}),
},
},
{
"id": "TC-03",
"name": "Implicit Tool Need",
"category": "A-ToolSelection",
"user_message": "I need to let Sarah know the meeting moved to 3pm.",
"mock_tool_outputs": {
"get_contacts": json.dumps({
"results": [
{"name": "Sarah Chen", "email": "sarah.chen@company.com"}
]
}),
"send_email": json.dumps({
"status": "sent",
"message_id": "msg_8821",
}),
},
},
# --- Category B: Parameter Precision ---
{
"id": "TC-04",
"name": "Unit Handling",
"category": "B-ParameterPrecision",
"user_message": "What's the temperature in Tokyo in Fahrenheit?",
"mock_tool_outputs": {
"get_weather": json.dumps({
"location": "Tokyo",
"temperature": 64,
"units": "fahrenheit",
"condition": "Clear",
"humidity": 55,
}),
# Default (celsius) mock for partial-credit scoring
"get_weather__default": json.dumps({
"location": "Tokyo",
"temperature": 18,
"units": "celsius",
"condition": "Clear",
"humidity": 55,
}),
},
},
{
"id": "TC-05",
"name": "Date and Time Parsing",
"category": "B-ParameterPrecision",
"user_message": (
"Schedule a team standup for next Monday at 9:30am, "
"30 minutes, with Alex and Jamie."
),
"mock_tool_outputs": {
"get_contacts": json.dumps({
"results": [
{"name": "Alex Stone", "email": "alex.stone@company.com"},
{"name": "Jamie Liu", "email": "jamie.liu@company.com"},
]
}),
"create_calendar_event": json.dumps({
"event_id": "evt_4412",
"status": "created",
"title": "Team Standup",
"date": "2026-03-23",
}),
},
# Reference date is 2026-03-20 (Friday), next Monday = 2026-03-23
"reference_date": "2026-03-20",
},
{
"id": "TC-06",
"name": "Multi-Value Extraction",
"category": "B-ParameterPrecision",
"user_message": (
"Translate 'Where is the nearest hospital?' "
"from English to both Spanish and Japanese."
),
"mock_tool_outputs": {
"translate_text__spanish": json.dumps({
"translated": "\u00bfD\u00f3nde est\u00e1 el hospital m\u00e1s cercano?",
}),
"translate_text__japanese": json.dumps({
"translated": "\u6700\u5bc4\u308a\u306e\u75c5\u9662\u306f\u3069\u3053\u3067\u3059\u304b\uff1f",
}),
},
},
# --- Category C: Multi-Step Chains ---
{
"id": "TC-07",
"name": "Search \u2192 Read \u2192 Act",
"category": "C-MultiStepChains",
"user_message": (
"Find the Q3 budget report and email the total to my manager."
),
"mock_tool_outputs": {
"search_files": json.dumps({
"results": [
{
"file_id": "file_091",
"name": "Q3_Budget_Report_2025.xlsx",
}
]
}),
"read_file": json.dumps({
"content": (
"Department budgets: Engineering $2.1M, Marketing $800K, "
"Sales $1.5M. Total: $4.4M"
),
}),
"get_contacts": json.dumps({
"results": [
{
"name": "Jordan Park",
"email": "jordan.park@company.com",
"role": "manager",
}
]
}),
"send_email": json.dumps({"status": "sent"}),
},
},
{
"id": "TC-08",
"name": "Conditional Branching",
"category": "C-MultiStepChains",
"user_message": (
"Check the weather in Paris. If it's raining, "
"remind me to bring an umbrella tomorrow at 8am."
),
"mock_tool_outputs": {
"get_weather": json.dumps({
"location": "Paris",
"temperature": 11,
"condition": "Light rain",
"humidity": 89,
}),
"set_reminder": json.dumps({
"reminder_id": "rem_553",
"status": "set",
}),
},
"reference_date": "2026-03-20",
},
{
"id": "TC-09",
"name": "Parallel Independence",
"category": "C-MultiStepChains",
"user_message": (
"What's the weather in London and the stock price of MSFT?"
),
"mock_tool_outputs": {
"get_weather": json.dumps({
"location": "London",
"temperature": 12,
"condition": "Cloudy",
}),
"get_stock_price": json.dumps({
"ticker": "MSFT",
"price": 412.78,
"currency": "USD",
}),
"web_search": json.dumps({
"results": [
{
"snippet": (
"London is cloudy at 12C and MSFT is around $412.78."
)
}
]
}),
},
},
# --- Category D: Restraint & Refusal ---
{
"id": "TC-10",
"name": "Trivial Knowledge",
"category": "D-RestraintRefusal",
"user_message": "What year did World War II end?",
"mock_tool_outputs": {},
},
{
"id": "TC-11",
"name": "Simple Math",
"category": "D-RestraintRefusal",
"user_message": "What is 15% of 200?",
"mock_tool_outputs": {
"calculator": json.dumps({"result": 30}),
},
},
{
"id": "TC-12",
"name": "Impossible Request",
"category": "D-RestraintRefusal",
"user_message": "Delete all my emails from last month.",
"mock_tool_outputs": {},
},
# --- Category E: Error Recovery ---
{
"id": "TC-13",
"name": "Empty Results",
"category": "E-ErrorRecovery",
"user_message": "Find the Johnson proposal document.",
"mock_tool_outputs": {
# First call returns empty, second (broader) returns a result
"search_files__first": json.dumps({"results": []}),
"search_files__retry": json.dumps({
"results": [
{
"file_id": "file_117",
"name": "Johnson_Project_Proposal_v2.docx",
}
]
}),
},
},
{
"id": "TC-14",
"name": "Malformed Response",
"category": "E-ErrorRecovery",
"user_message": "What's Apple's stock price?",
"mock_tool_outputs": {
"get_stock_price": json.dumps({
"error": "Service temporarily unavailable. Rate limit exceeded.",
}),
"web_search": json.dumps({
"results": [
{"snippet": "Apple (AAPL) is trading around $187.42."}
]
}),
},
},
{
"id": "TC-15",
"name": "Conflicting Information",
"category": "E-ErrorRecovery",
"user_message": (
"Search for the population of Iceland "
"and calculate what 2% of it would be."
),
"mock_tool_outputs": {
"web_search": json.dumps({
"results": [
{
"snippet": (
"Iceland has a population of approximately "
"372,520 as of 2025."
)
}
]
}),
"calculator": json.dumps({"result": 7450.4}),
},
},
]
class ToolCall15Dataset(DatasetProvider):
"""ToolCall-15 tool calling benchmark.
Provides 15 scenarios across 5 categories that test whether a model
can call the right tool with the right arguments. All tool outputs
are pre-defined (mocked) per the benchmark specification.
"""
dataset_id = "toolcall15"
dataset_name = "ToolCall-15"
def __init__(self) -> None:
self._records: List[EvalRecord] = []
def verify_requirements(self) -> List[str]:
return []
def load(
self,
*,
max_samples: Optional[int] = None,
split: Optional[str] = None,
seed: Optional[int] = None,
) -> None:
scenarios = list(SCENARIOS)
# Optional category filter via split (e.g. "A", "A,B", "D-RestraintRefusal")
if split:
filter_cats = [c.strip().upper() for c in split.split(",")]
scenarios = [
s
for s in scenarios
if any(
s["category"].upper().startswith(fc) for fc in filter_cats
)
]
if seed is not None:
random.Random(seed).shuffle(scenarios)
if max_samples is not None:
scenarios = scenarios[:max_samples]
self._records = [
EvalRecord(
record_id=s["id"],
problem=s["user_message"],
reference="", # scoring uses metadata, not reference text
category=s["category"],
subject=s["name"],
metadata={
"system_prompt": SYSTEM_PROMPT,
"tools": TOOLS,
"mock_tool_outputs": s["mock_tool_outputs"],
"reference_date": s.get("reference_date"),
"scenario_id": s["id"],
"scenario_name": s["name"],
},
)
for s in scenarios
]
LOGGER.info("ToolCall-15: loaded %d scenarios", len(self._records))
def iter_records(self) -> Iterable[EvalRecord]:
return iter(self._records)
def size(self) -> int:
return len(self._records)
__all__ = ["ToolCall15Dataset"]
@@ -227,6 +227,8 @@ class TauBenchTaskEnv:
max_tokens: int = 4096,
user_model: Optional[str] = None,
num_trials: int = 1,
telemetry: bool = False,
gpu_metrics: bool = False,
) -> None:
self._record = record
self._num_trials = num_trials
@@ -235,6 +237,8 @@ class TauBenchTaskEnv:
self._temperature = temperature
self._max_tokens = max_tokens
self._user_model = user_model or "gpt-5-mini-2025-08-07"
self._telemetry = telemetry
self._gpu_metrics = gpu_metrics
self._system = None
def __enter__(self) -> TauBenchTaskEnv:
@@ -244,7 +248,9 @@ class TauBenchTaskEnv:
builder = SystemBuilder()
if self._engine_key:
builder.engine(self._engine_key)
self._system = builder.build()
if self._gpu_metrics:
builder._config.telemetry.gpu_metrics = True
self._system = builder.telemetry(self._telemetry).build()
# Run the simulation
self._run_simulation()
@@ -0,0 +1,243 @@
"""LiveCodeBench scorer — sandboxed code execution with test cases.
Extracts code from model output, runs it against test cases in a
sandboxed subprocess with timeout and resource limits, and scores
based on pass/fail of each test case.
Reference: https://livecodebench.github.io/
"""
from __future__ import annotations
import logging
import os
import re
import subprocess
import tempfile
from typing import Any, Dict, List, Optional, Tuple
from openjarvis.evals.core.scorer import Scorer
from openjarvis.evals.core.types import EvalRecord
LOGGER = logging.getLogger(__name__)
# Execution limits
_TIMEOUT_SECONDS = 30
_MAX_OUTPUT_BYTES = 1024 * 1024 # 1 MB
def _extract_code(answer: str) -> str:
"""Extract Python code from model answer, handling markdown fences."""
# Try markdown code fence first (```python ... ``` or ``` ... ```)
fence_match = re.search(
r"```(?:python|py)?\s*\n(.*?)```",
answer,
re.DOTALL,
)
if fence_match:
return fence_match.group(1).strip()
# Look for common code patterns (import, def, class, or I/O operations)
lines = answer.strip().split("\n")
code_lines: list[str] = []
in_code = False
for line in lines:
stripped = line.lstrip()
if stripped.startswith((
"def ", "class ", "from ", "import ",
"n = ", "t = ", "for ", "while ",
"input(", "sys.stdin", "print(",
)):
in_code = True
if in_code:
code_lines.append(line)
if code_lines:
return "\n".join(code_lines)
# Last resort: return the whole answer (it might be pure code)
return answer.strip()
def _run_single_test(
code: str,
test_input: str,
expected_output: str,
timeout: int = _TIMEOUT_SECONDS,
) -> Tuple[bool, str]:
"""Run code in a subprocess with the given input, compare output.
Returns (passed, detail_message).
"""
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".py",
delete=False,
) as f:
f.write(code)
f.flush()
script_path = f.name
try:
result = subprocess.run(
["python3", script_path],
input=test_input,
capture_output=True,
text=True,
timeout=timeout,
env={
**os.environ,
"PYTHONDONTWRITEBYTECODE": "1",
},
)
actual_output = result.stdout.strip()
expected_stripped = expected_output.strip()
if result.returncode != 0:
stderr_preview = result.stderr[:500] if result.stderr else ""
return False, f"runtime_error (exit {result.returncode}): {stderr_preview}"
if actual_output == expected_stripped:
return True, "exact_match"
# Try line-by-line comparison (ignore trailing whitespace per line)
actual_lines = [line.rstrip() for line in actual_output.split("\n")]
expected_lines = [line.rstrip() for line in expected_stripped.split("\n")]
if actual_lines == expected_lines:
return True, "match_after_strip"
# Numeric tolerance comparison for floating point outputs
if _numeric_match(actual_output, expected_stripped):
return True, "numeric_match"
return False, (
f"wrong_answer: expected={expected_stripped[:200]!r}, "
f"got={actual_output[:200]!r}"
)
except subprocess.TimeoutExpired:
return False, f"timeout ({timeout}s)"
except Exception as exc:
return False, f"execution_error: {exc}"
finally:
try:
os.unlink(script_path)
except OSError:
pass
def _numeric_match(actual: str, expected: str, rel_tol: float = 1e-6) -> bool:
"""Check if outputs match numerically (for floating point problems)."""
actual_parts = actual.split()
expected_parts = expected.split()
if len(actual_parts) != len(expected_parts):
return False
for a, e in zip(actual_parts, expected_parts):
try:
a_f = float(a)
e_f = float(e)
if e_f == 0.0:
if abs(a_f) > rel_tol:
return False
elif abs(a_f - e_f) / max(abs(e_f), 1e-15) > rel_tol:
return False
except ValueError:
if a != e:
return False
return True
class LiveCodeBenchScorer(Scorer):
"""Score LiveCodeBench problems by running code against test cases.
Executes model-generated code in a sandboxed subprocess with stdin/stdout
test cases. Each test case is run independently with a timeout.
"""
scorer_id = "livecodebench"
def __init__(self, judge_backend=None, judge_model: str = "") -> None:
# Accept same constructor args as LLMJudgeScorer for CLI compatibility
# but test execution does not need an LLM judge
pass
def score(
self,
record: EvalRecord,
model_answer: str,
) -> Tuple[Optional[bool], Dict[str, Any]]:
if not model_answer or not model_answer.strip():
return False, {"reason": "empty_response"}
test_inputs: List[str] = record.metadata.get("test_inputs", [])
test_outputs: List[str] = record.metadata.get("test_outputs", [])
if not test_inputs or not test_outputs:
return None, {"reason": "no_test_cases"}
if len(test_inputs) != len(test_outputs):
LOGGER.warning(
"Test input/output count mismatch for %s: %d inputs, %d outputs",
record.record_id,
len(test_inputs),
len(test_outputs),
)
# Use the minimum count
count = min(len(test_inputs), len(test_outputs))
test_inputs = test_inputs[:count]
test_outputs = test_outputs[:count]
code = _extract_code(model_answer)
if not code:
return False, {"reason": "no_code_extracted"}
# Determine per-test timeout from metadata or default
time_limit = record.metadata.get("time_limit")
timeout = _TIMEOUT_SECONDS
if time_limit is not None:
try:
# time_limit is usually in seconds; add buffer
timeout = max(int(float(str(time_limit))) * 2, 5)
timeout = min(timeout, 60) # cap at 60s
except (ValueError, TypeError):
pass
passed = 0
total = len(test_inputs)
test_details: List[Dict[str, Any]] = []
for i, (inp, expected) in enumerate(zip(test_inputs, test_outputs)):
ok, detail = _run_single_test(code, inp, expected, timeout=timeout)
if ok:
passed += 1
test_details.append({
"test_index": i,
"passed": ok,
"detail": detail,
})
if total == 0:
return None, {"reason": "no_test_cases_after_filtering"}
pass_rate = passed / total
is_correct = passed == total
meta: Dict[str, Any] = {
"match_type": "test_execution",
"tests_passed": passed,
"tests_total": total,
"pass_rate": pass_rate,
"difficulty": record.metadata.get("difficulty", ""),
"platform": record.metadata.get("platform", ""),
}
# Include first few test details (avoid bloating metadata)
meta["test_details"] = test_details[:5]
if len(test_details) > 5:
meta["test_details_truncated"] = True
return is_correct, meta
__all__ = ["LiveCodeBenchScorer"]
@@ -0,0 +1,318 @@
"""LiveResearchBench scorer — LLM-as-judge for deep research quality.
Evaluates research output quality across four dimensions from the
LiveResearchBench rubric: comprehensiveness, insight, instruction_following,
and readability. Uses LLM-as-judge with per-task criteria when available,
falling back to a generic research quality rubric.
Reference: https://github.com/Ayanami0730/deep_research_bench
Paper: https://arxiv.org/abs/2510.14240
"""
from __future__ import annotations
import json
import logging
import re
from typing import Any, Dict, List, Optional, Tuple
from openjarvis.evals.core.scorer import LLMJudgeScorer
from openjarvis.evals.core.types import EvalRecord
LOGGER = logging.getLogger(__name__)
# The four scoring dimensions from LiveResearchBench
DIMENSIONS = ["comprehensiveness", "insight", "instruction_following", "readability"]
# Default dimension weights when task-specific weights are unavailable
DEFAULT_WEIGHTS = {
"comprehensiveness": 0.25,
"insight": 0.30,
"instruction_following": 0.25,
"readability": 0.20,
}
_GENERIC_RUBRIC = """Evaluate the research report across these four dimensions on a 0-10 scale:
1. **Comprehensiveness** (information coverage, depth, data support, balanced perspectives):
- 0-3: Minimal coverage, misses major aspects of the topic
- 4-6: Covers main points but lacks depth or misses important subtopics
- 7-8: Thorough coverage with good depth and supporting evidence
- 9-10: Exceptional coverage, addresses all facets with rich detail
2. **Insight** (analysis depth, critical thinking, original perspectives, forward-thinking):
- 0-3: Surface-level description only, no analysis
- 4-6: Some analysis but mostly descriptive, limited original thinking
- 7-8: Strong analytical depth with meaningful insights and reasoning
- 9-10: Exceptional analysis with novel perspectives and deep understanding
3. **Instruction Following** (task adherence, scope compliance, requirement completeness):
- 0-3: Fails to address the core research question
- 4-6: Partially addresses the question but misses key requirements
- 7-8: Addresses all major requirements with minor omissions
- 9-10: Fully addresses every aspect of the research task
4. **Readability** (structure, language fluency, technical terminology, presentation):
- 0-3: Poorly organized, difficult to follow
- 4-6: Reasonable structure but could be clearer
- 7-8: Well-structured with clear writing and good flow
- 9-10: Exceptionally clear, professional structure and presentation"""
def _format_criteria_rubric(criterions: Dict[str, List[Dict[str, Any]]]) -> str:
"""Format task-specific criteria into a rubric string for the judge prompt."""
parts: List[str] = []
for dimension in DIMENSIONS:
criteria_list = criterions.get(dimension, [])
if not criteria_list:
continue
parts.append(f"\n### {dimension.replace('_', ' ').title()}")
for i, crit in enumerate(criteria_list, 1):
criterion = crit.get("criterion", "")
explanation = crit.get("explanation", "")
weight = crit.get("weight", 0.0)
parts.append(
f" {i}. [{weight:.0%}] {criterion}"
+ (f"\n {explanation}" if explanation else "")
)
return "\n".join(parts)
def _build_judge_prompt(
*,
task_prompt: str,
article: str,
rubric: str,
) -> str:
"""Build the LLM judge prompt for evaluating a research report."""
return f"""You are an expert evaluator assessing the quality of an AI-generated research report.
## Original Research Task
{task_prompt}
## Research Report to Evaluate
{article}
## Evaluation Rubric
{rubric}
## Instructions
Evaluate the research report against the rubric criteria across four dimensions:
comprehensiveness, insight, instruction_following, and readability.
Score each dimension on a 0-10 scale.
Return your evaluation as JSON with this exact structure:
```json
{{
"scores": {{
"comprehensiveness": <score 0-10>,
"insight": <score 0-10>,
"instruction_following": <score 0-10>,
"readability": <score 0-10>
}},
"weighted_total": <weighted score 0-10>,
"notes": "brief justification for each dimension score"
}}
```
Be a rigorous evaluator. Reserve scores of 9-10 for genuinely excellent work.
A score of 5 represents adequate but unremarkable quality."""
def _parse_judge_response(raw: str) -> Dict[str, Any]:
"""Parse LLM judge response, extracting dimension scores.
Tries: JSON code block -> balanced braces -> regex fallback.
"""
if not raw or not raw.strip():
return {
"scores": {},
"weighted_total": 0.0,
"notes": "Empty judge response",
}
# Try JSON code block
code_block = re.search(r"```json\s*(.*?)\s*```", raw, re.DOTALL)
if code_block:
try:
parsed = json.loads(code_block.group(1))
if isinstance(parsed, dict):
return _normalize_response(parsed)
except json.JSONDecodeError:
pass
# Try balanced braces extraction
candidates: List[str] = []
depth = 0
current: List[str] = []
for char in raw:
if char == "{":
if depth == 0:
current = []
depth += 1
if depth > 0:
current.append(char)
if char == "}":
depth -= 1
if depth == 0 and current:
candidates.append("".join(current))
for candidate in reversed(candidates):
try:
parsed = json.loads(candidate)
if isinstance(parsed, dict) and "scores" in parsed:
return _normalize_response(parsed)
except json.JSONDecodeError:
continue
for candidate in reversed(candidates):
try:
parsed = json.loads(candidate)
if isinstance(parsed, dict):
return _normalize_response(parsed)
except json.JSONDecodeError:
continue
# Regex fallback: look for individual dimension scores
scores: Dict[str, float] = {}
for dim in DIMENSIONS:
match = re.search(
rf"{dim}[:\s]*([\d.]+)",
raw,
re.IGNORECASE,
)
if match:
try:
val = float(match.group(1))
if 0.0 <= val <= 10.0:
scores[dim] = val
except ValueError:
pass
if scores:
LOGGER.warning("Fell back to regex score extraction")
mean_score = sum(scores.values()) / len(scores) if scores else 0.0
return {
"scores": scores,
"weighted_total": mean_score,
"notes": "Scores extracted from prose via regex",
}
LOGGER.warning("Failed to parse judge response")
return {
"scores": {},
"weighted_total": 0.0,
"notes": "Failed to parse judge response",
}
def _normalize_response(parsed: Dict[str, Any]) -> Dict[str, Any]:
"""Normalize judge response to standard structure."""
result: Dict[str, Any] = {"scores": {}, "weighted_total": 0.0, "notes": ""}
# Extract scores
scores_data = parsed.get("scores", {})
if isinstance(scores_data, dict):
for dim in DIMENSIONS:
val = scores_data.get(dim)
if isinstance(val, (int, float)):
result["scores"][dim] = float(val)
# Extract weighted_total
for key in ("weighted_total", "total", "overall_score", "score"):
if key in parsed and isinstance(parsed[key], (int, float)):
result["weighted_total"] = float(parsed[key])
break
else:
# Compute mean if no explicit total
if result["scores"]:
values = list(result["scores"].values())
result["weighted_total"] = sum(values) / len(values)
# Extract notes
for key in ("notes", "justification", "reasoning"):
if key in parsed:
result["notes"] = str(parsed[key])
break
return result
class LiveResearchBenchScorer(LLMJudgeScorer):
"""LLM-as-judge scorer for LiveResearchBench deep research tasks.
Evaluates research reports across four dimensions:
comprehensiveness, insight, instruction_following, readability.
Uses task-specific criteria when available from the benchmark data.
"""
scorer_id = "liveresearch"
def score(
self,
record: EvalRecord,
model_answer: str,
) -> Tuple[Optional[bool], Dict[str, Any]]:
if not model_answer or not model_answer.strip():
return False, {"reason": "empty_response", "score": 0.0}
# Build rubric — use task-specific criteria if available
criterions = record.metadata.get("criterions")
if criterions and isinstance(criterions, dict):
rubric = _format_criteria_rubric(criterions)
else:
rubric = _GENERIC_RUBRIC
# Get the original research task (strip our wrapper prompt)
task_prompt = record.problem
# Build judge prompt
prompt = _build_judge_prompt(
task_prompt=task_prompt,
article=model_answer,
rubric=rubric,
)
try:
raw = self._ask_judge(prompt, temperature=0.0, max_tokens=4096)
except Exception as exc:
LOGGER.error(
"LLM judge call failed for %s: %s", record.record_id, exc
)
return None, {"error": str(exc), "score": 0.0}
parsed = _parse_judge_response(raw)
scores = parsed.get("scores", {})
# Compute weighted total using task-specific or default weights
dimension_weights = record.metadata.get("dimension_weight", DEFAULT_WEIGHTS)
weighted_total = 0.0
total_weight = 0.0
for dim in DIMENSIONS:
dim_score = scores.get(dim, 0.0)
dim_weight = dimension_weights.get(dim, DEFAULT_WEIGHTS.get(dim, 0.25))
weighted_total += dim_score * dim_weight
total_weight += dim_weight
if total_weight > 0:
weighted_total /= total_weight
# Normalize to 0-1 range (scores are 0-10)
normalized_score = weighted_total / 10.0
# Threshold: score >= 0.5 (i.e., 5/10 weighted average) is considered passing
is_correct = normalized_score >= 0.5
metadata: Dict[str, Any] = {
"score": normalized_score,
"dimension_scores": scores,
"dimension_weights": dimension_weights,
"weighted_total_0_10": weighted_total,
"notes": parsed.get("notes", ""),
"raw_judge_output": raw,
}
return is_correct, metadata
__all__ = ["LiveResearchBenchScorer"]
+529
View File
@@ -0,0 +1,529 @@
"""ToolCall-15 scorer — deterministic tool-calling evaluation.
Scores each of the 15 scenarios based on whether the model called the
correct tool(s) with correct arguments, following the scoring rubric
defined in the benchmark's METHODOLOGY.md.
Scoring: 0 (fail), 1 (partial), or 2 (full pass) per scenario.
is_correct = True when score == 2 (full pass).
Reference: https://github.com/stevibe/ToolCall-15
"""
from __future__ import annotations
import logging
from typing import Any, Dict, List, Optional, Tuple
from openjarvis.evals.core.scorer import LLMJudgeScorer
from openjarvis.evals.core.types import EvalRecord
LOGGER = logging.getLogger(__name__)
def _extract_tool_calls(record: EvalRecord) -> List[Dict[str, Any]]:
"""Extract tool calls from the record's query trace or metadata.
Returns a list of dicts with keys: name, arguments.
"""
tool_calls: List[Dict[str, Any]] = []
# Try query trace first (from EvalRunner)
trace = record.metadata.get("query_trace")
if trace:
for turn in trace.turns:
for tc in getattr(turn, "tool_calls", []):
if tc is None:
continue
tool_calls.append({
"name": tc.get("name", ""),
"arguments": tc.get("arguments") or {},
})
return tool_calls
# Try tool_results list (from JarvisAgentBackend)
tool_results = record.metadata.get("tool_results", [])
for tr in tool_results:
tool_calls.append({
"name": tr.get("tool_name", ""),
"arguments": tr.get("arguments") or {},
})
return tool_calls
def _tools_called(tool_calls: List[Dict[str, Any]], name: str) -> List[Dict[str, Any]]:
"""Filter tool calls by tool name (case-insensitive)."""
return [tc for tc in tool_calls if tc["name"].lower() == name.lower()]
def _has_tool(tool_calls: List[Dict[str, Any]], name: str) -> bool:
"""Check if a specific tool was called."""
return len(_tools_called(tool_calls, name)) > 0
def _arg_contains(tc: Dict[str, Any], key: str, substring: str) -> bool:
"""Check if a tool call argument contains a substring (case-insensitive)."""
val = tc.get("arguments", {}).get(key, "")
if isinstance(val, str):
return substring.lower() in val.lower()
return str(val).lower().find(substring.lower()) >= 0
def _arg_equals(tc: Dict[str, Any], key: str, value: str) -> bool:
"""Check if a tool call argument equals a value (case-insensitive)."""
val = tc.get("arguments", {}).get(key, "")
if isinstance(val, str):
return val.lower() == value.lower()
return str(val).lower() == value.lower()
# ---------------------------------------------------------------------------
# Per-scenario scoring functions
# ---------------------------------------------------------------------------
def _score_tc01(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-01: Direct Specialist Match — get_weather for Berlin."""
weather = _tools_called(tool_calls, "get_weather")
web = _tools_called(tool_calls, "web_search")
if weather and not web and len(tool_calls) == 1:
tc = weather[0]
if _arg_contains(tc, "location", "berlin"):
return 2, "PASS: get_weather(Berlin), no web_search"
if web and not weather:
return 1, "PARTIAL: used web_search instead of get_weather"
if weather:
return 1, "PARTIAL: get_weather called but with extra tool calls"
return 0, "FAIL: did not call get_weather correctly"
def _score_tc02(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-02: Distractor Resistance — get_stock_price for AAPL only."""
stock = _tools_called(tool_calls, "get_stock_price")
web = _tools_called(tool_calls, "web_search")
if stock and not web and len(tool_calls) == 1:
tc = stock[0]
if _arg_contains(tc, "ticker", "aapl"):
return 2, "PASS: get_stock_price(AAPL), no distractors"
if stock and web:
return 1, "PARTIAL: correct tool but also called web_search"
if stock:
return 1, "PARTIAL: get_stock_price called but extra tools used"
return 0, "FAIL: did not isolate to get_stock_price"
def _score_tc03(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-03: Implicit Tool Need — get_contacts then send_email."""
contacts = _tools_called(tool_calls, "get_contacts")
email = _tools_called(tool_calls, "send_email")
if contacts and email:
contact_has_sarah = any(
_arg_contains(tc, "query", "sarah") for tc in contacts
)
email_has_addr = any(
_arg_contains(tc, "to", "sarah.chen@company.com") for tc in email
)
if contact_has_sarah and email_has_addr:
return 2, "PASS: get_contacts(sarah) -> send_email(sarah.chen@company.com)"
if contact_has_sarah:
return 2, "PASS: get_contacts(sarah) -> send_email chained"
return 1, "PARTIAL: both tools called but query/threading unclear"
# Partial: asked user for email instead of looking up
if not tool_calls:
lower = answer.lower()
if "email" in lower or "sarah" in lower:
return 1, "PARTIAL: asked for clarification instead of acting"
return 0, "FAIL: did not complete contact lookup to email chain"
def _score_tc04(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-04: Unit Handling — get_weather(Tokyo, units=fahrenheit)."""
weather = _tools_called(tool_calls, "get_weather")
if weather:
tc = weather[0]
has_tokyo = _arg_contains(tc, "location", "tokyo")
has_f = _arg_equals(tc, "units", "fahrenheit")
if has_tokyo and has_f:
return 2, "PASS: get_weather(Tokyo, units=fahrenheit)"
if has_tokyo and not has_f:
# Check if answer manually converts
lower = answer.lower()
if "fahrenheit" in lower or "64" in lower:
return 1, "PARTIAL: correct tool but manual conversion"
return 0, "FAIL: ignored fahrenheit instruction"
return 0, "FAIL: did not call get_weather"
def _score_tc05(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-05: Date and Time Parsing — create_calendar_event with correct fields."""
events = _tools_called(tool_calls, "create_calendar_event")
if not events:
return 0, "FAIL: no calendar event created"
tc = events[0]
args = tc.get("arguments", {})
date = str(args.get("date", ""))
time_val = str(args.get("time", ""))
duration = args.get("duration_minutes")
attendees = args.get("attendees", [])
# Check date (next Monday from 2026-03-20 = 2026-03-23)
date_ok = "2026-03-23" in date
time_ok = time_val in ("09:30", "9:30")
duration_ok = duration == 30 or str(duration) == "30"
attendees_lower = [a.lower() if isinstance(a, str) else "" for a in (attendees or [])]
attendees_str = " ".join(attendees_lower)
has_alex = "alex" in attendees_str
has_jamie = "jamie" in attendees_str
if date_ok and time_ok and duration_ok and has_alex and has_jamie:
return 2, "PASS: all fields correct"
if date_ok and time_ok:
return 1, "PARTIAL: correct date/time but missing duration or attendees"
return 0, "FAIL: date or time parsing incorrect"
def _score_tc06(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-06: Multi-Value Extraction — two translate_text calls."""
translates = _tools_called(tool_calls, "translate_text")
if len(translates) < 2:
return 0, "FAIL: did not split into two translate_text calls"
has_spanish = any(
_arg_contains(tc, "target_language", "spanish") for tc in translates
)
has_japanese = any(
_arg_contains(tc, "target_language", "japanese") for tc in translates
)
has_source = any(
_arg_contains(tc, "source_language", "english") for tc in translates
)
if has_spanish and has_japanese and has_source:
return 2, "PASS: two separate translate_text calls with correct params"
if has_spanish or has_japanese:
return 1, "PARTIAL: some translation calls correct but incomplete"
return 0, "FAIL: translate_text calls missing correct languages"
def _score_tc07(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-07: Search -> Read -> Act — 4-step chain."""
search = _tools_called(tool_calls, "search_files")
read = _tools_called(tool_calls, "read_file")
contacts = _tools_called(tool_calls, "get_contacts")
email = _tools_called(tool_calls, "send_email")
steps_done = sum([
bool(search),
bool(read),
bool(contacts),
bool(email),
])
if steps_done == 4:
# Verify data threading
has_file_id = any(
_arg_contains(tc, "file_id", "file_091") for tc in read
)
has_manager = any(
_arg_contains(tc, "query", "manager") for tc in contacts
)
has_total = any(
_arg_contains(tc, "body", "4.4") for tc in email
)
email_to = any(
_arg_contains(tc, "to", "jordan.park@company.com") for tc in email
)
if has_file_id and (has_manager or email_to) and has_total:
return 2, "PASS: all 4 steps with correct data threading"
return 2, "PASS: all 4 steps completed"
if steps_done >= 3:
return 1, f"PARTIAL: {steps_done}/4 steps completed"
return 0, f"FAIL: only {steps_done}/4 steps completed"
def _score_tc08(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-08: Conditional Branching — weather check then conditional reminder."""
weather = _tools_called(tool_calls, "get_weather")
reminder = _tools_called(tool_calls, "set_reminder")
if weather and reminder:
weather_paris = any(
_arg_contains(tc, "location", "paris") for tc in weather
)
reminder_umbrella = any(
_arg_contains(tc, "message", "umbrella") for tc in reminder
)
has_correct_date = any(
_arg_contains(tc, "datetime", "2026-03-21") for tc in reminder
)
if weather_paris and reminder_umbrella and has_correct_date:
return 2, "PASS: weather check -> conditional reminder with correct date"
if weather_paris and reminder_umbrella:
return 2, "PASS: weather check -> conditional reminder"
return 2, "PASS: weather and reminder both called"
if weather and not reminder:
return 1, "PARTIAL: weather checked but no reminder set"
return 0, "FAIL: did not follow conditional flow"
def _score_tc09(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-09: Parallel Independence — weather AND stock price."""
weather = _tools_called(tool_calls, "get_weather")
stock = _tools_called(tool_calls, "get_stock_price")
web = _tools_called(tool_calls, "web_search")
weather_london = any(
_arg_contains(tc, "location", "london") for tc in weather
)
stock_msft = any(
_arg_contains(tc, "ticker", "msft") for tc in stock
)
if weather_london and stock_msft:
return 2, "PASS: both get_weather(London) and get_stock_price(MSFT)"
if web and not weather and not stock:
# Used web_search to cover both — partial
return 1, "PARTIAL: used web_search instead of specialist tools"
if weather_london or stock_msft:
return 1, "PARTIAL: only one of two requests handled"
return 0, "FAIL: did not handle both requests"
def _score_tc10(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-10: Trivial Knowledge — answer without tools."""
if not tool_calls and "1945" in answer:
return 2, "PASS: direct answer with no tool calls"
if not tool_calls:
return 0, "FAIL: no tool calls but answer missing 1945"
return 0, "FAIL: unnecessary tool call for trivial knowledge"
def _score_tc11(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-11: Simple Math — answer without tools."""
if not tool_calls and "30" in answer:
return 2, "PASS: direct answer with no tool calls"
calc = _tools_called(tool_calls, "calculator")
web = _tools_called(tool_calls, "web_search")
if calc and not web and "30" in answer:
return 1, "PARTIAL: used calculator but correct answer"
if web:
return 0, "FAIL: used web_search for simple math"
if not tool_calls:
return 0, "FAIL: no tools but answer missing 30"
return 0, "FAIL: unnecessary tool usage"
def _score_tc12(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-12: Impossible Request — refuse without tool calls."""
lower = answer.lower()
refusal_phrases = [
"cannot", "can't", "do not have", "don't have",
"not able", "unable", "no tool", "not available",
"not possible", "no way to",
]
has_refusal = any(phrase in lower for phrase in refusal_phrases)
mentions_delete = "delete" in lower or "available tool" in lower or "tool" in lower
if not tool_calls and has_refusal and mentions_delete:
return 2, "PASS: clean refusal without tool calls"
if not tool_calls and has_refusal:
return 2, "PASS: refusal without tool calls"
if tool_calls:
return 0, "FAIL: called tools for impossible request"
return 0, "FAIL: did not refuse clearly"
def _score_tc13(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-13: Empty Results — retry or ask for clarification."""
search = _tools_called(tool_calls, "search_files")
if len(search) >= 2:
return 2, "PASS: retried search after empty results"
if len(search) == 1:
lower = answer.lower()
clarification_phrases = [
"could you", "can you", "more details", "more specific",
"clarify", "which", "what type", "not found", "no results",
]
if any(phrase in lower for phrase in clarification_phrases):
return 2, "PASS: single search with clarification request"
return 0, "FAIL: did not adapt after empty result"
def _score_tc14(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-14: Malformed Response — surface error, offer fallback."""
stock = _tools_called(tool_calls, "get_stock_price")
web = _tools_called(tool_calls, "web_search")
lower = answer.lower()
error_ack_phrases = [
"temporarily unavailable", "rate limit", "service",
"couldn't", "error", "unable", "failed",
"get_stock_price",
]
has_error_ack = any(phrase in lower for phrase in error_ack_phrases)
if stock and has_error_ack:
if not web or "fallback" in lower or "web" in lower or "try" in lower:
return 2, "PASS: acknowledged error, handled gracefully"
if stock and web and not has_error_ack:
return 1, "PARTIAL: recovered via web_search but did not surface error"
if stock:
return 1, "PARTIAL: called get_stock_price but unclear error handling"
return 0, "FAIL: did not handle error with integrity"
def _score_tc15(
tool_calls: List[Dict[str, Any]], answer: str,
) -> Tuple[int, str]:
"""TC-15: Conflicting Information — use search result in calculator."""
web = _tools_called(tool_calls, "web_search")
calc = _tools_called(tool_calls, "calculator")
web_has_iceland = any(
_arg_contains(tc, "query", "iceland") or
_arg_contains(tc, "query", "population")
for tc in web
)
# Check calculator uses the actual searched number (372520)
calc_has_number = any(
"372520" in str(tc.get("arguments", {}).get("expression", "")).replace(",", "")
or "372,520" in str(tc.get("arguments", {}).get("expression", ""))
for tc in calc
)
if web_has_iceland and calc_has_number:
return 2, "PASS: calculator uses exact search result (372520)"
if web_has_iceland and calc:
# Calculator was called but might use rounded/memorized number
return 1, "PARTIAL: both tools called but data integrity unclear"
if web_has_iceland and not calc:
if "7450" in answer or "7,450" in answer:
return 1, "PARTIAL: manual calculation from search result"
return 0, "FAIL: search done but no calculator call"
return 0, "FAIL: did not chain web_search to calculator"
# Dispatch table
_SCORERS = {
"TC-01": _score_tc01,
"TC-02": _score_tc02,
"TC-03": _score_tc03,
"TC-04": _score_tc04,
"TC-05": _score_tc05,
"TC-06": _score_tc06,
"TC-07": _score_tc07,
"TC-08": _score_tc08,
"TC-09": _score_tc09,
"TC-10": _score_tc10,
"TC-11": _score_tc11,
"TC-12": _score_tc12,
"TC-13": _score_tc13,
"TC-14": _score_tc14,
"TC-15": _score_tc15,
}
class ToolCall15Scorer(LLMJudgeScorer):
"""Deterministic scorer for ToolCall-15 benchmark.
Scores each scenario based on whether the model called the correct
tool(s) with correct arguments. No LLM judge is needed — scoring
is fully deterministic, but the class extends LLMJudgeScorer to
satisfy the _build_scorer interface.
Scoring: 0 (fail), 1 (partial), 2 (full pass).
is_correct = True when score == 2.
"""
scorer_id = "toolcall15"
def score(
self,
record: EvalRecord,
model_answer: str,
) -> Tuple[Optional[bool], Dict[str, Any]]:
scenario_id = record.metadata.get("scenario_id", record.record_id)
scorer_fn = _SCORERS.get(scenario_id)
if scorer_fn is None:
LOGGER.warning("No scorer for scenario %s", scenario_id)
return None, {
"reason": "unknown_scenario",
"scenario_id": scenario_id,
}
tool_calls = _extract_tool_calls(record)
points, reason = scorer_fn(tool_calls, model_answer)
is_correct: Optional[bool]
if points == 2:
is_correct = True
elif points == 1:
is_correct = False # partial credit, not fully correct
else:
is_correct = False
return is_correct, {
"scenario_id": scenario_id,
"scenario_name": record.metadata.get("scenario_name", ""),
"category": record.category,
"points": points,
"max_points": 2,
"reason": reason,
"tool_calls": [
{"name": tc["name"], "arguments": tc.get("arguments", {})}
for tc in tool_calls
],
}
__all__ = ["ToolCall15Scorer"]
+25 -2
View File
@@ -201,9 +201,21 @@ class JarvisSystem:
self.bus.subscribe(EventType.INFERENCE_END, _on_inference_end)
# Run
# Run — wrap with TraceCollector when tracing is enabled.
# Check trace_store (set at build time) instead of config.traces.enabled
# because the shared config singleton can be mutated by other SystemBuilder
# instances (e.g. the judge backend).
try:
result = ag.run(query, context=ctx)
if self.trace_store is not None:
from openjarvis.traces.collector import TraceCollector
collector = TraceCollector(
ag, store=self.trace_store, bus=self.bus,
)
result = collector.run(query, context=ctx)
self.trace_collector = collector
else:
result = ag.run(query, context=ctx)
finally:
self.bus.unsubscribe(EventType.INFERENCE_END, _on_inference_end)
@@ -608,6 +620,16 @@ class SystemBuilder:
# Set up session store
session_store = self._setup_sessions(config)
# Set up trace store
trace_store = None
if traces_enabled:
try:
from openjarvis.traces.store import TraceStore
trace_store = TraceStore(config.traces.db_path)
except Exception:
logger.warning("Failed to initialize TraceStore", exc_info=True)
# Set up capability policy
capability_policy = sec.capability_policy
@@ -685,6 +707,7 @@ class SystemBuilder:
memory_backend=memory_backend,
channel_backend=channel_backend,
telemetry_store=telemetry_store,
trace_store=trace_store,
gpu_monitor=gpu_monitor,
scheduler_store=scheduler_store,
scheduler=task_scheduler,
@@ -74,6 +74,7 @@ class InstrumentedEngine(InferenceEngine):
self._bus = bus
self._gpu_monitor = gpu_monitor
self._energy_monitor = energy_monitor
self._publishes_events = True
def generate(
self,
@@ -254,6 +255,10 @@ class InstrumentedEngine(InferenceEngine):
"mean_itl_ms": mean_itl_ms,
"energy_method": energy_method,
"energy_vendor": energy_vendor,
# Rich trace data: model response content
"content": result.get("content", ""),
"tool_calls": result.get("tool_calls", []),
"finish_reason": result.get("finish_reason", ""),
}
self._bus.publish(EventType.INFERENCE_END, event_data)
+39 -13
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
import time
from typing import Any, Optional
from typing import Any, Dict, List, Optional
from openjarvis.agents._stubs import AgentContext, AgentResult, BaseAgent
from openjarvis.core.events import EventBus, EventType
@@ -19,12 +19,15 @@ class TraceCollector:
``TraceStep`` objects. When the agent finishes, the complete ``Trace``
is persisted to the ``TraceStore`` and published on the bus.
Enhanced to capture full model response content, tool call arguments and
results, and the complete conversation message history.
Usage::
agent = OrchestratorAgent(engine, model, tools=tools, bus=bus)
collector = TraceCollector(agent, store=trace_store, bus=bus)
result = collector.run("What is 2+2?")
# Trace is automatically saved to trace_store
trace = collector.last_trace # Rich trace with steps + messages
"""
def __init__(
@@ -40,6 +43,7 @@ class TraceCollector:
self._current_steps: list[TraceStep] = []
self._current_model: str = ""
self._current_engine: str = ""
self._last_trace: Optional[Trace] = None
def run(
self,
@@ -73,6 +77,9 @@ class TraceCollector:
)
)
# Extract messages from agent result metadata
messages: List[Dict[str, Any]] = result.metadata.get("messages", [])
# Build and persist the trace
trace = Trace(
query=input,
@@ -81,6 +88,7 @@ class TraceCollector:
engine=self._current_engine,
steps=list(self._current_steps),
result=result.content,
messages=messages,
started_at=started_at,
ended_at=ended_at,
)
@@ -89,6 +97,8 @@ class TraceCollector:
trace.total_latency_seconds += step.duration_seconds
trace.total_tokens += step.output.get("tokens", 0)
self._last_trace = trace
if self._store is not None:
self._store.save(trace)
@@ -99,11 +109,8 @@ class TraceCollector:
@property
def last_trace(self) -> Optional[Trace]:
"""Return the trace from the most recent ``run()``, if available."""
if not self._current_steps:
return None
# Reconstruct from saved steps (steps cleared on next run)
return None # Use TraceStore.get() for retrieval after run
"""Return the trace from the most recent ``run()``."""
return self._last_trace
# -- event handlers --------------------------------------------------------
@@ -146,31 +153,50 @@ class TraceCollector:
"prompt_tokens": usage.get("prompt_tokens", 0),
"completion_tokens": usage.get("completion_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
"tokens": usage.get("total_tokens", data.get("total_tokens", 0)),
"tokens": usage.get(
"total_tokens", data.get("total_tokens", 0),
),
"content": data.get("content", ""),
"tool_calls": data.get("tool_calls", []),
"finish_reason": data.get("finish_reason", ""),
},
metadata={
"engine": self._current_engine,
"ttft": data.get("ttft", 0.0),
"energy_joules": data.get("energy_joules", 0.0),
"power_watts": data.get("power_watts", 0.0),
"gpu_utilization_pct": data.get("gpu_utilization_pct", 0.0),
"throughput_tok_per_sec": data.get("throughput_tok_per_sec", 0.0),
"gpu_utilization_pct": data.get(
"gpu_utilization_pct", 0.0,
),
"throughput_tok_per_sec": data.get(
"throughput_tok_per_sec", 0.0,
),
},
)
)
def _on_tool_start(self, event: Any) -> None:
self._tool_start_time = event.timestamp
self._tool_start_data = event.data
def _on_tool_end(self, event: Any) -> None:
start = getattr(self, "_tool_start_time", event.timestamp)
start_data = getattr(self, "_tool_start_data", {})
self._current_steps.append(
TraceStep(
step_type=StepType.TOOL_CALL,
timestamp=start,
duration_seconds=event.data.get("latency", event.timestamp - start),
input={"tool": event.data.get("tool", "")},
output={"success": event.data.get("success", False)},
duration_seconds=event.data.get(
"latency", event.timestamp - start,
),
input={
"tool": event.data.get("tool", ""),
"arguments": start_data.get("arguments", {}),
},
output={
"success": event.data.get("success", False),
"result": event.data.get("result", ""),
},
)
)
+16 -3
View File
@@ -25,7 +25,8 @@ CREATE TABLE IF NOT EXISTS traces (
ended_at REAL NOT NULL DEFAULT 0.0,
total_tokens INTEGER NOT NULL DEFAULT 0,
total_latency_seconds REAL NOT NULL DEFAULT 0.0,
metadata TEXT NOT NULL DEFAULT '{}'
metadata TEXT NOT NULL DEFAULT '{}',
messages TEXT NOT NULL DEFAULT '[]'
);
"""
@@ -64,8 +65,8 @@ _INSERT_TRACE = """\
INSERT INTO traces (
trace_id, query, agent, model, engine, result,
outcome, feedback, started_at, ended_at,
total_tokens, total_latency_seconds, metadata
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
total_tokens, total_latency_seconds, metadata, messages
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""
_INSERT_STEP = """\
@@ -95,6 +96,13 @@ class TraceStore:
self._conn.execute(_CREATE_STEPS)
self._conn.execute(_CREATE_FTS)
self._conn.execute(_FTS_SYNC_INSERT)
# Migrate: add messages column if missing (pre-existing databases)
try:
self._conn.execute(
"ALTER TABLE traces ADD COLUMN messages TEXT NOT NULL DEFAULT '[]'"
)
except sqlite3.OperationalError:
pass # Column already exists
self._conn.commit()
def save(self, trace: Trace) -> None:
@@ -115,6 +123,7 @@ class TraceStore:
trace.total_tokens,
trace.total_latency_seconds,
json.dumps(trace.metadata),
json.dumps(trace.messages),
),
)
for idx, step in enumerate(trace.steps):
@@ -257,6 +266,9 @@ class TraceStore:
)
for sr in step_rows
]
# messages column (index 14) was added after metadata; handle
# databases created before the column existed.
messages_raw = row[14] if len(row) > 14 else "[]"
return Trace(
trace_id=trace_id,
query=row[2],
@@ -271,6 +283,7 @@ class TraceStore:
total_tokens=row[11],
total_latency_seconds=row[12],
metadata=json.loads(row[13]),
messages=json.loads(messages_raw),
steps=steps,
)
+43 -1
View File
@@ -129,6 +129,27 @@ class TestDatasetInstantiation:
assert ds.dataset_id == "terminalbench-native"
assert ds.dataset_name == "TerminalBench Native"
def test_livecodebench(self) -> None:
from openjarvis.evals.datasets.livecodebench import LiveCodeBenchDataset
ds = LiveCodeBenchDataset()
assert ds.dataset_id == "livecodebench"
assert ds.dataset_name == "LiveCodeBench"
def test_liveresearch(self) -> None:
from openjarvis.evals.datasets.liveresearch import LiveResearchBenchDataset
ds = LiveResearchBenchDataset()
assert ds.dataset_id == "liveresearch"
assert ds.dataset_name == "LiveResearchBench"
def test_toolcall15(self) -> None:
from openjarvis.evals.datasets.toolcall15 import ToolCall15Dataset
ds = ToolCall15Dataset()
assert ds.dataset_id == "toolcall15"
assert ds.dataset_name == "ToolCall-15"
# ---------------------------------------------------------------------------
# Scorer instantiation tests
@@ -233,6 +254,24 @@ class TestScorerInstantiation:
s = TerminalBenchNativeScorer(_mock_backend(), "test-model")
assert s.scorer_id == "terminalbench-native"
def test_livecodebench_scorer(self) -> None:
from openjarvis.evals.scorers.livecodebench import LiveCodeBenchScorer
s = LiveCodeBenchScorer(_mock_backend(), "test-model")
assert s.scorer_id == "livecodebench"
def test_liveresearch_scorer(self) -> None:
from openjarvis.evals.scorers.liveresearch import LiveResearchBenchScorer
s = LiveResearchBenchScorer(_mock_backend(), "test-model")
assert s.scorer_id == "liveresearch"
def test_toolcall15_scorer(self) -> None:
from openjarvis.evals.scorers.toolcall15 import ToolCall15Scorer
s = ToolCall15Scorer(_mock_backend(), "test-model")
assert s.scorer_id == "toolcall15"
# ---------------------------------------------------------------------------
# CLI factory tests
@@ -255,6 +294,9 @@ ALL_BENCHMARKS = [
"swefficiency",
"terminalbench",
"terminalbench-native",
"livecodebench",
"liveresearch",
"toolcall15",
]
@@ -313,7 +355,7 @@ class TestConfigBenchmarks:
def test_benchmarks_count(self) -> None:
from openjarvis.evals.core.config import KNOWN_BENCHMARKS
assert len(KNOWN_BENCHMARKS) == 27
assert len(KNOWN_BENCHMARKS) == 30
# ---------------------------------------------------------------------------
+591
View File
@@ -0,0 +1,591 @@
"""Tests for telemetry wiring in the eval pipeline.
Verifies that:
- FLOPs estimation flows from config metadata through to EvalResult and RunSummary
- Telemetry fields (energy, power, GPU util) propagate end-to-end
- JarvisDirectBackend propagates gpu_metrics flag
- TauBench dataset passes telemetry flags to task env
- Summary JSON includes telemetry_summary section
"""
from __future__ import annotations
import json
from unittest.mock import MagicMock, patch
import pytest
from openjarvis.evals.core.types import EvalResult, RunConfig, RunSummary
# ---------------------------------------------------------------------------
# FLOPs estimation in EvalResult
# ---------------------------------------------------------------------------
class TestEvalResultFlops:
"""Verify estimated_flops field exists and is serializable."""
def test_default_zero(self):
r = EvalResult(record_id="test", model_answer="hi")
assert r.estimated_flops == 0.0
def test_set_flops(self):
flops = 2.0 * 10.0 * 1e9 * 1000 # 10B active params, 1000 tokens
r = EvalResult(record_id="test", model_answer="hi", estimated_flops=flops)
assert r.estimated_flops == flops
def test_flops_in_dict(self):
"""EvalResult with estimated_flops can be serialized to JSON."""
r = EvalResult(record_id="t", model_answer="a", estimated_flops=1e15)
d = {"estimated_flops": r.estimated_flops}
s = json.dumps(d)
assert "1e+15" in s or "1000000000000000" in s
# ---------------------------------------------------------------------------
# RunSummary telemetry fields
# ---------------------------------------------------------------------------
class TestRunSummaryTelemetry:
"""Verify RunSummary includes FLOPs and telemetry aggregation fields."""
def test_default_flops_fields(self):
s = RunSummary(
benchmark="test",
category="chat",
backend="jarvis-direct",
model="test-model",
total_samples=1,
scored_samples=1,
correct=1,
accuracy=1.0,
errors=0,
mean_latency_seconds=1.0,
total_cost_usd=0.0,
)
assert s.total_estimated_flops == 0.0
assert s.flops_stats is None
# ---------------------------------------------------------------------------
# Runner _process_one FLOPs computation
# ---------------------------------------------------------------------------
class TestRunnerFlopsComputation:
"""Test that _process_one computes estimated_flops from model metadata."""
def test_flops_computed_from_metadata(self):
"""When param_count_b is in metadata, FLOPs are estimated."""
from openjarvis.evals.core.runner import EvalRunner
from openjarvis.evals.core.types import EvalRecord
config = RunConfig(
benchmark="test",
backend="jarvis-direct",
model="test-model",
metadata={
"param_count_b": 7.0,
"active_params_b": 7.0,
},
)
mock_backend = MagicMock()
mock_backend.generate_full.return_value = {
"content": "answer",
"usage": {"prompt_tokens": 100, "completion_tokens": 50},
"latency_seconds": 1.0,
"cost_usd": 0.0,
}
mock_scorer = MagicMock()
mock_scorer.score.return_value = (True, {})
mock_dataset = MagicMock()
runner = EvalRunner(config, mock_dataset, mock_backend, mock_scorer)
record = EvalRecord(
record_id="test-1",
problem="What is 2+2?",
reference="4",
category="reasoning",
)
result = runner._process_one(record)
# FLOPs = 2 * active_params * total_tokens
# = 2 * 7e9 * 150 = 2.1e12
expected_flops = 2.0 * 7.0 * 1e9 * 150
assert result.estimated_flops == pytest.approx(expected_flops)
def test_flops_zero_without_metadata(self):
"""When no param_count_b in metadata, FLOPs should be 0."""
from openjarvis.evals.core.runner import EvalRunner
from openjarvis.evals.core.types import EvalRecord
config = RunConfig(
benchmark="test",
backend="jarvis-direct",
model="test-model",
metadata={},
)
mock_backend = MagicMock()
mock_backend.generate_full.return_value = {
"content": "answer",
"usage": {"prompt_tokens": 100, "completion_tokens": 50},
"latency_seconds": 1.0,
"cost_usd": 0.0,
}
mock_scorer = MagicMock()
mock_scorer.score.return_value = (True, {})
mock_dataset = MagicMock()
runner = EvalRunner(config, mock_dataset, mock_backend, mock_scorer)
record = EvalRecord(
record_id="test-1",
problem="What is 2+2?",
reference="4",
category="reasoning",
)
result = runner._process_one(record)
assert result.estimated_flops == 0.0
def test_flops_uses_active_params_for_moe(self):
"""For MoE models, FLOPs should use active_params_b, not total."""
from openjarvis.evals.core.runner import EvalRunner
from openjarvis.evals.core.types import EvalRecord
config = RunConfig(
benchmark="test",
backend="jarvis-direct",
model="test-model",
metadata={
"param_count_b": 122.0,
"active_params_b": 10.0,
},
)
mock_backend = MagicMock()
mock_backend.generate_full.return_value = {
"content": "answer",
"usage": {"prompt_tokens": 200, "completion_tokens": 100},
"latency_seconds": 1.0,
"cost_usd": 0.0,
}
mock_scorer = MagicMock()
mock_scorer.score.return_value = (True, {})
mock_dataset = MagicMock()
runner = EvalRunner(config, mock_dataset, mock_backend, mock_scorer)
record = EvalRecord(
record_id="test-1",
problem="What is 2+2?",
reference="4",
category="reasoning",
)
result = runner._process_one(record)
# Should use active_params_b=10.0, not param_count_b=122.0
expected_flops = 2.0 * 10.0 * 1e9 * 300 # 300 total tokens
assert result.estimated_flops == pytest.approx(expected_flops)
# ---------------------------------------------------------------------------
# Summary JSON telemetry_summary section
# ---------------------------------------------------------------------------
class TestSummaryToDict:
"""Test that _summary_to_dict includes the telemetry_summary section."""
def test_telemetry_summary_present(self):
from openjarvis.evals.core.runner import _summary_to_dict
s = RunSummary(
benchmark="test",
category="chat",
backend="jarvis-direct",
model="test-model",
total_samples=10,
scored_samples=10,
correct=8,
accuracy=0.8,
errors=0,
mean_latency_seconds=2.0,
total_cost_usd=0.1,
total_energy_joules=50.0,
avg_power_watts=25.0,
total_input_tokens=5000,
total_output_tokens=2000,
total_estimated_flops=1.4e13,
efficiency={
"accuracy": 0.8,
"total_energy_joules": 50.0,
"avg_power_watts": 25.0,
"total_estimated_flops": 1.4e13,
"ipj": 0.016,
"ipw": 0.032,
},
)
d = _summary_to_dict(s)
# Check telemetry_summary section exists
assert "telemetry_summary" in d
ts = d["telemetry_summary"]
assert ts["total_energy_joules"] == 50.0
assert ts["avg_power_watts"] == 25.0
assert ts["total_input_tokens"] == 5000
assert ts["total_output_tokens"] == 2000
assert ts["total_tokens"] == 7000
assert ts["total_estimated_flops"] == 1.4e13
assert ts["ipw"] == 0.032
assert ts["ipj"] == 0.016
def test_flops_fields_in_summary_dict(self):
from openjarvis.evals.core.runner import _summary_to_dict
s = RunSummary(
benchmark="test",
category="chat",
backend="jarvis-direct",
model="test-model",
total_samples=1,
scored_samples=1,
correct=1,
accuracy=1.0,
errors=0,
mean_latency_seconds=1.0,
total_cost_usd=0.0,
total_estimated_flops=2.1e12,
)
d = _summary_to_dict(s)
assert d["total_estimated_flops"] == 2.1e12
assert "flops_stats" in d
# ---------------------------------------------------------------------------
# Flush result includes estimated_flops
# ---------------------------------------------------------------------------
class TestFlushResult:
"""Test that _flush_result includes estimated_flops in JSONL output."""
def test_estimated_flops_in_jsonl(self, tmp_path):
from openjarvis.evals.core.runner import EvalRunner
config = RunConfig(
benchmark="test",
backend="jarvis-direct",
model="test-model",
)
mock_backend = MagicMock()
mock_scorer = MagicMock()
mock_dataset = MagicMock()
runner = EvalRunner(config, mock_dataset, mock_backend, mock_scorer)
outfile = tmp_path / "results.jsonl"
runner._output_file = open(outfile, "w")
result = EvalResult(
record_id="test-1",
model_answer="answer",
estimated_flops=2.1e12,
)
runner._flush_result(result)
runner._output_file.close()
runner._output_file = None
lines = outfile.read_text().strip().split("\n")
record = json.loads(lines[0])
assert record["estimated_flops"] == 2.1e12
# ---------------------------------------------------------------------------
# Trace dict includes estimated_flops
# ---------------------------------------------------------------------------
class TestResultToTraceDict:
"""Test that _result_to_trace_dict includes estimated_flops."""
def test_estimated_flops_in_trace(self):
from openjarvis.evals.core.runner import _result_to_trace_dict
result = EvalResult(
record_id="test-1",
model_answer="answer",
estimated_flops=3.0e12,
)
d = _result_to_trace_dict(result)
assert d["estimated_flops"] == 3.0e12
# ---------------------------------------------------------------------------
# JarvisDirectBackend gpu_metrics propagation
# ---------------------------------------------------------------------------
class TestDirectBackendGpuMetrics:
"""Verify JarvisDirectBackend sets gpu_metrics on config."""
@patch("openjarvis.system.SystemBuilder")
def test_gpu_metrics_propagated(self, mock_builder_cls):
"""When gpu_metrics=True, the builder config should be updated."""
from openjarvis.evals.backends.jarvis_direct import JarvisDirectBackend
mock_builder = MagicMock()
mock_builder_cls.return_value = mock_builder
mock_builder.engine.return_value = mock_builder
mock_builder.telemetry.return_value = mock_builder
mock_builder.traces.return_value = mock_builder
# Create a mock config with telemetry.gpu_metrics attribute
mock_config = MagicMock()
mock_config.telemetry.gpu_metrics = False
mock_builder._config = mock_config
mock_system = MagicMock()
mock_builder.build.return_value = mock_system
JarvisDirectBackend(
engine_key="vllm",
telemetry=True,
gpu_metrics=True,
)
# Verify gpu_metrics was set to True on config
assert mock_config.telemetry.gpu_metrics is True
@patch("openjarvis.system.SystemBuilder")
def test_gpu_metrics_not_set_when_false(self, mock_builder_cls):
"""When gpu_metrics=False, the builder config should not be touched."""
from openjarvis.evals.backends.jarvis_direct import JarvisDirectBackend
mock_builder = MagicMock()
mock_builder_cls.return_value = mock_builder
mock_builder.engine.return_value = mock_builder
mock_builder.telemetry.return_value = mock_builder
mock_builder.traces.return_value = mock_builder
mock_config = MagicMock()
mock_config.telemetry.gpu_metrics = False
mock_builder._config = mock_config
mock_system = MagicMock()
mock_builder.build.return_value = mock_system
JarvisDirectBackend(
engine_key="vllm",
telemetry=False,
gpu_metrics=False,
)
# When gpu_metrics is False, the attribute should still be False
assert mock_config.telemetry.gpu_metrics is False
# ---------------------------------------------------------------------------
# TauBench telemetry flag passthrough
# ---------------------------------------------------------------------------
class TestTauBenchTelemetryPassthrough:
"""Verify TauBench dataset passes telemetry flags to task env."""
def test_set_engine_config_stores_flags(self):
"""set_engine_config should store telemetry and gpu_metrics."""
from openjarvis.evals.datasets.taubench import TauBenchDataset
ds = TauBenchDataset.__new__(TauBenchDataset)
ds._domains = ["airline"]
ds._records = []
ds._engine_key = None
ds._model = None
ds._temperature = 0.7
ds._max_tokens = 4096
ds._user_model = None
ds._num_trials = 3
ds._telemetry = False
ds._gpu_metrics = False
ds.set_engine_config(
engine_key="vllm",
model="test-model",
telemetry=True,
gpu_metrics=True,
)
assert ds._telemetry is True
assert ds._gpu_metrics is True
@patch("openjarvis.evals.execution.taubench_env.TauBenchTaskEnv")
def test_create_task_env_passes_flags(self, mock_env_cls):
"""create_task_env should forward telemetry flags."""
from openjarvis.evals.core.types import EvalRecord
from openjarvis.evals.datasets.taubench import TauBenchDataset
ds = TauBenchDataset.__new__(TauBenchDataset)
ds._domains = ["airline"]
ds._records = []
ds._engine_key = "vllm"
ds._model = "test-model"
ds._temperature = 0.7
ds._max_tokens = 4096
ds._user_model = None
ds._num_trials = 3
ds._telemetry = True
ds._gpu_metrics = True
record = EvalRecord(
record_id="airline_1",
problem="test",
reference="test",
category="airline",
metadata={"domain": "airline", "task_id": "1"},
)
ds.create_task_env(record)
mock_env_cls.assert_called_once_with(
record,
engine_key="vllm",
model="test-model",
temperature=0.7,
max_tokens=4096,
user_model=None,
num_trials=3,
telemetry=True,
gpu_metrics=True,
)
# ---------------------------------------------------------------------------
# RunConfig expand_suite preserves model metadata for FLOPs
# ---------------------------------------------------------------------------
class TestExpandSuiteModelMetadata:
"""Verify expand_suite passes param_count_b and active_params_b to RunConfig."""
def test_metadata_includes_params(self):
from openjarvis.evals.core.config import expand_suite
from openjarvis.evals.core.types import (
BenchmarkConfig,
DefaultsConfig,
EvalSuiteConfig,
ExecutionConfig,
JudgeConfig,
MetaConfig,
ModelConfig,
)
suite = EvalSuiteConfig(
meta=MetaConfig(name="test"),
defaults=DefaultsConfig(),
judge=JudgeConfig(),
run=ExecutionConfig(telemetry=True, gpu_metrics=True),
models=[
ModelConfig(
name="test/moe-model",
param_count_b=122.0,
active_params_b=10.0,
gpu_peak_tflops=989.5,
gpu_peak_bandwidth_gb_s=3350.0,
num_gpus=4,
),
],
benchmarks=[
BenchmarkConfig(name="math500"),
],
)
configs = expand_suite(suite)
assert len(configs) == 1
rc = configs[0]
assert rc.telemetry is True
assert rc.gpu_metrics is True
assert rc.metadata["param_count_b"] == 122.0
assert rc.metadata["active_params_b"] == 10.0
assert rc.metadata["gpu_peak_tflops"] == 989.5
assert rc.metadata["gpu_peak_bandwidth_gb_s"] == 3350.0
assert rc.metadata["num_gpus"] == 4
# ---------------------------------------------------------------------------
# Telemetry data flow from backend through runner
# ---------------------------------------------------------------------------
class TestTelemetryEndToEnd:
"""Test full telemetry data flow from backend to EvalResult."""
def test_telemetry_fields_populated(self):
"""When backend returns telemetry data, EvalResult captures it."""
from openjarvis.evals.core.runner import EvalRunner
from openjarvis.evals.core.types import EvalRecord
config = RunConfig(
benchmark="test",
backend="jarvis-direct",
model="test-model",
telemetry=True,
gpu_metrics=True,
)
mock_backend = MagicMock()
mock_backend.generate_full.return_value = {
"content": "answer",
"usage": {"prompt_tokens": 100, "completion_tokens": 50},
"latency_seconds": 1.5,
"cost_usd": 0.001,
"energy_joules": 12.5,
"power_watts": 250.0,
"gpu_utilization_pct": 85.0,
"throughput_tok_per_sec": 33.3,
"ttft": 0.05,
"_telemetry": {
"energy_per_output_token_joules": 0.25,
"throughput_per_watt": 0.133,
"mean_itl_ms": 28.5,
},
}
mock_scorer = MagicMock()
mock_scorer.score.return_value = (True, {})
mock_dataset = MagicMock()
runner = EvalRunner(config, mock_dataset, mock_backend, mock_scorer)
record = EvalRecord(
record_id="test-1",
problem="test",
reference="answer",
category="chat",
)
result = runner._process_one(record)
assert result.energy_joules == 12.5
assert result.power_watts == 250.0
assert result.gpu_utilization_pct == 85.0
assert result.throughput_tok_per_sec == 33.3
assert result.ipw == pytest.approx(1.0 / 250.0)
assert result.ipj == pytest.approx(1.0 / 12.5)
assert result.energy_per_output_token_joules == 0.25
assert result.throughput_per_watt == 0.133
assert result.mean_itl_ms == 28.5
+135
View File
@@ -234,3 +234,138 @@ class TestTraceCollector:
for s in trace.steps:
assert s.input.get("model") != "stray"
store.close()
class _RichToolAgent(BaseAgent):
"""Agent that emits content-enriched events for testing."""
agent_id = "rich_tool_agent"
def __init__(self, bus: EventBus) -> None:
self._bus = bus
def run(
self, input: str, context: Optional[AgentContext] = None,
**kwargs: Any,
) -> AgentResult:
from openjarvis.core.types import ToolResult
# Turn 1: inference with tool call request
self._bus.publish(EventType.INFERENCE_START, {
"model": "test-model", "engine": "test",
})
self._bus.publish(EventType.INFERENCE_END, {
"total_tokens": 30,
"usage": {"prompt_tokens": 20, "completion_tokens": 10},
"content": "I'll calculate that for you.",
"tool_calls": [
{"id": "call_1", "name": "calculator",
"arguments": "{\"expr\": \"2+2\"}"},
],
"finish_reason": "tool_calls",
})
# Tool execution
self._bus.publish(EventType.TOOL_CALL_START, {
"tool": "calculator", "arguments": {"expr": "2+2"},
})
self._bus.publish(EventType.TOOL_CALL_END, {
"tool": "calculator", "success": True, "latency": 0.01,
"result": "4",
})
# Turn 2: final answer
self._bus.publish(EventType.INFERENCE_START, {
"model": "test-model", "engine": "test",
})
self._bus.publish(EventType.INFERENCE_END, {
"total_tokens": 15,
"usage": {"prompt_tokens": 10, "completion_tokens": 5},
"content": "The answer is 4.",
"tool_calls": [],
"finish_reason": "stop",
})
# Return result with messages in metadata
messages = [
{"role": "user", "content": input},
{"role": "assistant", "content": "I'll calculate that for you."},
{"role": "tool", "content": "4", "name": "calculator"},
{"role": "assistant", "content": "The answer is 4."},
]
return AgentResult(
content="The answer is 4.",
tool_results=[
ToolResult(tool_name="calculator", content="4",
success=True),
],
turns=2,
metadata={"messages": messages},
)
class TestRichTraceCollector:
def test_captures_content_in_generate_steps(self, tmp_path: Path) -> None:
bus = EventBus()
store = TraceStore(tmp_path / "test.db")
agent = _RichToolAgent(bus=bus)
collector = TraceCollector(agent, store=store, bus=bus)
collector.run("What is 2+2?")
trace = store.list_traces()[0]
gen_steps = [s for s in trace.steps if s.step_type == StepType.GENERATE]
assert len(gen_steps) == 2
assert gen_steps[0].output["content"] == "I'll calculate that for you."
expected_tc = [
{"id": "call_1", "name": "calculator",
"arguments": "{\"expr\": \"2+2\"}"},
]
assert gen_steps[0].output["tool_calls"] == expected_tc
assert gen_steps[0].output["finish_reason"] == "tool_calls"
assert gen_steps[1].output["content"] == "The answer is 4."
assert gen_steps[1].output["finish_reason"] == "stop"
store.close()
def test_captures_tool_arguments_and_result(self, tmp_path: Path) -> None:
bus = EventBus()
store = TraceStore(tmp_path / "test.db")
agent = _RichToolAgent(bus=bus)
collector = TraceCollector(agent, store=store, bus=bus)
collector.run("What is 2+2?")
trace = store.list_traces()[0]
tool_steps = [s for s in trace.steps if s.step_type == StepType.TOOL_CALL]
assert len(tool_steps) == 1
assert tool_steps[0].input["tool"] == "calculator"
assert tool_steps[0].input["arguments"] == {"expr": "2+2"}
assert tool_steps[0].output["result"] == "4"
assert tool_steps[0].output["success"] is True
store.close()
def test_captures_messages_in_trace(self, tmp_path: Path) -> None:
bus = EventBus()
store = TraceStore(tmp_path / "test.db")
agent = _RichToolAgent(bus=bus)
collector = TraceCollector(agent, store=store, bus=bus)
collector.run("What is 2+2?")
trace = store.list_traces()[0]
assert len(trace.messages) == 4
assert trace.messages[0]["role"] == "user"
assert trace.messages[3]["role"] == "assistant"
store.close()
def test_last_trace_property(self, tmp_path: Path) -> None:
bus = EventBus()
store = TraceStore(tmp_path / "test.db")
agent = _RichToolAgent(bus=bus)
collector = TraceCollector(agent, store=store, bus=bus)
collector.run("What is 2+2?")
trace = collector.last_trace
assert trace is not None
assert trace.query == "What is 2+2?"
assert len(trace.messages) == 4
store.close()