test(benchmarks): rewrite benchmark runner and add brain scenarios
Fixes: 1. Rewrote run_competitive_benchmark.py to test BOTH capabilities: - Telepathic (JSON element extraction) - Brain (Free-text action extraction with format_json=False) 2. Normalizes scores by averaging per-scenario (fixes score inflation where models with more scenarios tested scored higher but were marked unsuitable). 3. Added 4 new brain_action scenarios to ensure the 'think=false' code path is actively benchmarked going forward. 4. Added test_benchmark_integrity.py to lock in scenario format rules. 5. Cleared stale llm_benchmarks.json data to force clean re-evaluations.
This commit is contained in:
@@ -9,11 +9,14 @@ from datetime import datetime
|
||||
# Add root project path so we can import internal modules safely
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from GramAddict.core.llm_provider import query_telepathic_llm
|
||||
from GramAddict.core.llm_provider import query_llm, query_telepathic_llm
|
||||
|
||||
BENCHMARKS_FILE = os.path.join(os.path.dirname(__file__), "data/llm_benchmarks.json")
|
||||
SCENARIOS_FILE = os.path.join(os.path.dirname(__file__), "data/benchmark_scenarios.json")
|
||||
|
||||
# Minimum iterations for statistical significance
|
||||
MIN_ITERATIONS = 5
|
||||
|
||||
|
||||
def load_json(path):
|
||||
if os.path.exists(path):
|
||||
@@ -31,35 +34,37 @@ def save_json(path, data):
|
||||
|
||||
|
||||
def normalize_scores(db):
|
||||
"""Normalize relative performance by AVERAGE score per scenario, not raw totals."""
|
||||
if not db.get("models"):
|
||||
return db
|
||||
|
||||
# 1. Find the highest raw score across all models
|
||||
max_raw = 0
|
||||
max_avg = 0
|
||||
leader_model = None
|
||||
|
||||
for name, data in db["models"].items():
|
||||
if data.get("is_unsuitable"):
|
||||
continue
|
||||
|
||||
raw = data.get("raw_score", 0)
|
||||
if raw > max_raw:
|
||||
max_raw = raw
|
||||
scenario_count = data.get("scenario_count", 1)
|
||||
avg = data.get("raw_score", 0) / max(scenario_count, 1)
|
||||
data["avg_score_per_scenario"] = round(avg, 1)
|
||||
|
||||
if avg > max_avg:
|
||||
max_avg = avg
|
||||
leader_model = name
|
||||
elif raw == max_raw and max_raw > 0:
|
||||
# Tie-breaker: Latency
|
||||
elif avg == max_avg and max_avg > 0:
|
||||
current_lat = data.get("latency_ms", 99999)
|
||||
leader_lat = db["models"][leader_model].get("latency_ms", 99999)
|
||||
if current_lat < leader_lat:
|
||||
leader_model = name
|
||||
|
||||
if max_raw == 0:
|
||||
if max_avg == 0:
|
||||
return db
|
||||
|
||||
# 2. Update relative performance
|
||||
for name, data in db["models"].items():
|
||||
raw = data.get("raw_score", 0)
|
||||
data["relative_performance_pct"] = round((raw / max_raw) * 100, 1)
|
||||
scenario_count = data.get("scenario_count", 1)
|
||||
avg = data.get("raw_score", 0) / max(scenario_count, 1)
|
||||
data["relative_performance_pct"] = round((avg / max_avg) * 100, 1)
|
||||
data["is_leader"] = name == leader_model
|
||||
|
||||
return db
|
||||
@@ -75,21 +80,15 @@ def get_installed_ollama_models():
|
||||
models = []
|
||||
for line in output.split("\n")[1:]:
|
||||
if line.strip():
|
||||
# Format: NAME, ID, SIZE, MODIFIED
|
||||
parts = line.split()
|
||||
if len(parts) >= 3:
|
||||
name = parts[0]
|
||||
size = parts[2]
|
||||
|
||||
# 1. Skip if size is '-' (remote/cloud model)
|
||||
if size == "-":
|
||||
continue
|
||||
|
||||
# 2. Skip ':cloud' tagged models explicitly
|
||||
if ":cloud" in name:
|
||||
continue
|
||||
|
||||
# 3. Filter out purely embedding models
|
||||
if any(k in name.lower() for k in ["embed", "minilm", "rerank"]):
|
||||
continue
|
||||
|
||||
@@ -100,7 +99,131 @@ def get_installed_ollama_models():
|
||||
return []
|
||||
|
||||
|
||||
def benchmark_model(model_name: str, url: str, force: bool = False, iterations: int = 3):
|
||||
def _run_telepathic_scenario(scenario, model_name, url, iterations):
|
||||
"""Run a telepathic (JSON element selection) scenario."""
|
||||
system_prompt = (
|
||||
"You identify which UI element to tap based ONLY on a JSON array of parsed Android elements. "
|
||||
'Output ONLY valid JSON: {"index": number, "reason": "brief reason"}'
|
||||
)
|
||||
|
||||
user_prompt = (
|
||||
f"Which element should I tap to: {scenario['task']}\n\n"
|
||||
f"Elements:\n{json.dumps(scenario['nodes'], indent=1)}\n\n"
|
||||
"Rules:\n"
|
||||
"- Pick the SMALLEST, most specific button or icon\n"
|
||||
"- NEVER pick large containers\n"
|
||||
'Return: {"index": number, "reason": "..."}'
|
||||
)
|
||||
|
||||
latencies = []
|
||||
scores = []
|
||||
successes = 0
|
||||
|
||||
for _ in range(iterations):
|
||||
start_time = time.time()
|
||||
try:
|
||||
resp_str = query_telepathic_llm(model_name, url, system_prompt, user_prompt)
|
||||
latency = int((time.time() - start_time) * 1000)
|
||||
latencies.append(latency)
|
||||
except Exception as e:
|
||||
print(f" ❌ API Request failed: {e}")
|
||||
scores.append(0)
|
||||
continue
|
||||
|
||||
raw_points = 0
|
||||
try:
|
||||
clean = resp_str.strip()
|
||||
if clean.startswith("```json"):
|
||||
clean = clean[7:]
|
||||
if clean.endswith("```"):
|
||||
clean = clean[:-3]
|
||||
data = json.loads(clean)
|
||||
|
||||
if "index" in data and "reason" in data:
|
||||
raw_points += 40
|
||||
if data["index"] == scenario["target_index"]:
|
||||
raw_points += 60
|
||||
successes += 1
|
||||
else:
|
||||
print(f" ❌ Wrong index ({data.get('index')}). Target was {scenario['target_index']}.")
|
||||
else:
|
||||
print(" ❌ JSON missing fields.")
|
||||
except Exception:
|
||||
print(" ❌ JSON Parsing failed.")
|
||||
|
||||
scores.append(raw_points)
|
||||
|
||||
return scores, latencies, successes
|
||||
|
||||
|
||||
def _run_brain_scenario(scenario, model_name, url, iterations):
|
||||
"""Run a brain action extraction scenario (format_json=False)."""
|
||||
system_prompt = (
|
||||
f"You are an autonomous Instagram agent. Your goal is: '{scenario['task']}'.\n"
|
||||
f"You are currently on screen: {scenario['screen_type']}.\n"
|
||||
f"Available actions: {scenario['available_actions']}\n"
|
||||
"INSTRUCTIONS: Reply with ONLY the action string. Nothing else."
|
||||
)
|
||||
|
||||
user_prompt = "Choose the next best action."
|
||||
|
||||
latencies = []
|
||||
scores = []
|
||||
successes = 0
|
||||
|
||||
for _ in range(iterations):
|
||||
start_time = time.time()
|
||||
try:
|
||||
# CRITICAL: Use format_json=False — this is the Brain code path
|
||||
ans = query_llm(
|
||||
url=url,
|
||||
model=model_name,
|
||||
prompt=user_prompt,
|
||||
system=system_prompt,
|
||||
format_json=False,
|
||||
timeout=30,
|
||||
temperature=0.0,
|
||||
max_tokens=50,
|
||||
)
|
||||
latency = int((time.time() - start_time) * 1000)
|
||||
latencies.append(latency)
|
||||
except Exception as e:
|
||||
print(f" ❌ API Request failed: {e}")
|
||||
scores.append(0)
|
||||
continue
|
||||
|
||||
raw_points = 0
|
||||
if ans and "response" in ans:
|
||||
response = ans["response"].strip().lower()
|
||||
|
||||
# Points for structural adherence (returned a clean string)
|
||||
if response and response in [a.lower() for a in scenario["available_actions"]]:
|
||||
raw_points += 40
|
||||
|
||||
# Points for correctness
|
||||
if scenario.get("accept_any_valid"):
|
||||
# Any valid action from the list is acceptable
|
||||
raw_points += 60
|
||||
successes += 1
|
||||
elif response == scenario["target_action"].lower():
|
||||
raw_points += 60
|
||||
successes += 1
|
||||
else:
|
||||
print(f" ⚠️ Valid but suboptimal: '{response}' (target: '{scenario['target_action']}')")
|
||||
raw_points += 20 # Partial credit for valid but wrong action
|
||||
else:
|
||||
print(f" ❌ Invalid response: '{response}' not in available actions")
|
||||
else:
|
||||
print(" ❌ Empty or null response from LLM")
|
||||
|
||||
scores.append(raw_points)
|
||||
|
||||
return scores, latencies, successes
|
||||
|
||||
|
||||
def benchmark_model(model_name: str, url: str, force: bool = False, iterations: int = MIN_ITERATIONS):
|
||||
iterations = max(iterations, MIN_ITERATIONS) # Enforce minimum
|
||||
|
||||
db = load_json(BENCHMARKS_FILE) or {"models": {}}
|
||||
scenarios_data = load_json(SCENARIOS_FILE)
|
||||
if not scenarios_data:
|
||||
@@ -113,95 +236,46 @@ def benchmark_model(model_name: str, url: str, force: bool = False, iterations:
|
||||
print(f"Typical execution skip for {model_name} (Rel: {pct}%). Use --force.")
|
||||
return
|
||||
|
||||
print(f"\n🚀 [Competitive Benchmarking] Model: {model_name}")
|
||||
print(f"\n🚀 [Competitive Benchmarking] Model: {model_name} ({iterations} iterations)")
|
||||
|
||||
total_raw = 0
|
||||
total_latency = 0
|
||||
results_detail = {}
|
||||
passed_all = True
|
||||
|
||||
system_prompt = (
|
||||
"You identify which UI element to tap based ONLY on a JSON array of parsed Android elements. "
|
||||
'Output ONLY valid JSON: {"index": number, "reason": "brief reason"}'
|
||||
)
|
||||
|
||||
scenarios = scenarios_data["scenarios"]
|
||||
for scenario in scenarios:
|
||||
print(f"--- Running: {scenario['name']} ---")
|
||||
scenario_type = scenario.get("type", "telepathic")
|
||||
print(f"--- [{scenario_type.upper()}] {scenario['name']} ---")
|
||||
|
||||
user_prompt = (
|
||||
f"Which element should I tap to: {scenario['task']}\n\n"
|
||||
f"Elements:\n{json.dumps(scenario['nodes'], indent=1)}\n\n"
|
||||
"Rules:\n"
|
||||
"- Pick the SMALLEST, most specific button or icon\n"
|
||||
"- NEVER pick large containers\n"
|
||||
"Return: {\"index\": number, \"reason\": \"...\"}"
|
||||
)
|
||||
|
||||
scenario_latencies = []
|
||||
scenario_scores = []
|
||||
successes = 0
|
||||
|
||||
for _ in range(iterations):
|
||||
start_time = time.time()
|
||||
try:
|
||||
resp_str = query_telepathic_llm(model_name, url, system_prompt, user_prompt)
|
||||
latency = int((time.time() - start_time) * 1000)
|
||||
scenario_latencies.append(latency)
|
||||
except Exception as e:
|
||||
print(f" ❌ API Request failed for scenario {scenario['id']}: {e}")
|
||||
passed_all = False
|
||||
continue
|
||||
|
||||
raw_points = 0
|
||||
try:
|
||||
clean = resp_str.strip()
|
||||
if clean.startswith("```json"):
|
||||
clean = clean[7:]
|
||||
if clean.endswith("```"):
|
||||
clean = clean[:-3]
|
||||
data = json.loads(clean)
|
||||
|
||||
# Points for structural adherence
|
||||
if "index" in data and "reason" in data:
|
||||
raw_points += 40
|
||||
|
||||
# Points for correctness
|
||||
if data["index"] == scenario["target_index"]:
|
||||
raw_points += 60
|
||||
successes += 1
|
||||
else:
|
||||
print(f" ❌ Wrong index ({data.get('index')}). Target was {scenario['target_index']}.")
|
||||
else:
|
||||
print(" ❌ JSON missing fields.")
|
||||
except Exception:
|
||||
print(" ❌ JSON Parsing failed.")
|
||||
|
||||
scenario_scores.append(raw_points)
|
||||
|
||||
avg_scenario_score = int(sum(scenario_scores) / len(scenario_scores)) if scenario_scores else 0
|
||||
avg_scenario_latency = int(sum(scenario_latencies) / len(scenario_latencies)) if scenario_latencies else 0
|
||||
if scenario_type == "telepathic":
|
||||
scores, latencies, successes = _run_telepathic_scenario(scenario, model_name, url, iterations)
|
||||
elif scenario_type == "brain_action":
|
||||
scores, latencies, successes = _run_brain_scenario(scenario, model_name, url, iterations)
|
||||
else:
|
||||
print(f" ⚠️ Unknown scenario type: {scenario_type}")
|
||||
continue
|
||||
|
||||
avg_score = int(sum(scores) / len(scores)) if scores else 0
|
||||
avg_latency = int(sum(latencies) / len(latencies)) if latencies else 0
|
||||
pass_rate = (successes / iterations) * 100
|
||||
|
||||
if pass_rate < 100.0:
|
||||
passed_all = False
|
||||
|
||||
print(
|
||||
f" Result: {pass_rate:.0f}% Pass Rate | Avg Score: {avg_scenario_score}/100 | Avg Latency: {avg_scenario_latency}ms"
|
||||
)
|
||||
print(f" Result: {pass_rate:.0f}% Pass | Avg Score: {avg_score}/100 | Avg Latency: {avg_latency}ms")
|
||||
|
||||
# Consistent format: always an object
|
||||
results_detail[scenario["id"]] = {
|
||||
"avg_score": avg_scenario_score,
|
||||
"avg_score": avg_score,
|
||||
"pass_rate": pass_rate,
|
||||
"latency": avg_scenario_latency,
|
||||
"latency": avg_latency,
|
||||
}
|
||||
total_raw += avg_scenario_score
|
||||
total_latency += avg_scenario_latency
|
||||
total_raw += avg_score
|
||||
total_latency += avg_latency
|
||||
|
||||
avg_latency = total_latency // len(scenarios) if scenarios else 0
|
||||
print(
|
||||
f"\n📊 {model_name} Result: {'PASS' if passed_all else 'FAIL'} | Avg Score: {total_raw} | Latency: {avg_latency}ms"
|
||||
)
|
||||
print(f"\n📊 {model_name}: {'PASS' if passed_all else 'FAIL'} | Total: {total_raw} | Latency: {avg_latency}ms")
|
||||
|
||||
if model_name not in db["models"]:
|
||||
db["models"][model_name] = {}
|
||||
@@ -209,16 +283,17 @@ def benchmark_model(model_name: str, url: str, force: bool = False, iterations:
|
||||
db["models"][model_name].update(
|
||||
{
|
||||
"raw_score": total_raw,
|
||||
"scenario_count": len(scenarios),
|
||||
"telepathic_score": int((total_raw / (len(scenarios) * 100)) * 100) if scenarios else 0,
|
||||
"latency_ms": avg_latency,
|
||||
"last_tested": datetime.utcnow().isoformat() + "Z",
|
||||
"details": results_detail,
|
||||
"passed_all": passed_all,
|
||||
"is_unsuitable": not passed_all,
|
||||
"iterations": iterations,
|
||||
}
|
||||
)
|
||||
|
||||
# Recalculate relative scores across all models
|
||||
db = normalize_scores(db)
|
||||
save_json(BENCHMARKS_FILE, db)
|
||||
|
||||
@@ -233,7 +308,7 @@ if __name__ == "__main__":
|
||||
parser.add_argument("--force", action="store_true", help="Force re-testing")
|
||||
parser.add_argument("--all-ollama", action="store_true", help="Automatically find and test all local Ollama models")
|
||||
parser.add_argument(
|
||||
"--iterations", type=int, default=3, help="Number of iterations per scenario to measure reliability"
|
||||
"--iterations", type=int, default=MIN_ITERATIONS, help=f"Iterations per scenario (min: {MIN_ITERATIONS})"
|
||||
)
|
||||
|
||||
args, unknown = parser.parse_known_args()
|
||||
|
||||
113
tests/unit/test_benchmark_integrity.py
Normal file
113
tests/unit/test_benchmark_integrity.py
Normal file
@@ -0,0 +1,113 @@
|
||||
"""
|
||||
Benchmark Integrity Tests
|
||||
==========================
|
||||
|
||||
These tests ensure the benchmark infrastructure produces RELIABLE,
|
||||
COMPARABLE results across model evaluations.
|
||||
|
||||
Covers:
|
||||
1. Scenario data consistency (no mixed formats)
|
||||
2. Brain-type scenarios exist and are tested via format_json=False
|
||||
3. Scoring normalization (per-scenario, not raw totals)
|
||||
4. Minimum iteration count enforcement
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
BENCHMARKS_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "benchmarks", "data")
|
||||
SCENARIOS_FILE = os.path.join(BENCHMARKS_DIR, "benchmark_scenarios.json")
|
||||
RESULTS_FILE = os.path.join(BENCHMARKS_DIR, "llm_benchmarks.json")
|
||||
|
||||
|
||||
class TestBenchmarkScenarioIntegrity:
|
||||
"""Contract: Benchmark scenarios must cover BOTH bot capabilities."""
|
||||
|
||||
def test_scenarios_file_exists(self):
|
||||
assert os.path.exists(SCENARIOS_FILE), "benchmark_scenarios.json is missing!"
|
||||
|
||||
def test_scenarios_have_required_fields(self):
|
||||
with open(SCENARIOS_FILE) as f:
|
||||
data = json.load(f)
|
||||
|
||||
for scenario in data["scenarios"]:
|
||||
assert "id" in scenario, f"Scenario missing 'id': {scenario}"
|
||||
assert "name" in scenario, f"Scenario missing 'name': {scenario}"
|
||||
assert "task" in scenario, f"Scenario missing 'task': {scenario}"
|
||||
assert "type" in scenario, (
|
||||
f"Scenario '{scenario['id']}' missing 'type' field. " f"Must be 'telepathic' or 'brain_action'."
|
||||
)
|
||||
assert scenario["type"] in ("telepathic", "brain_action"), (
|
||||
f"Scenario '{scenario['id']}' has invalid type '{scenario['type']}'. "
|
||||
f"Must be 'telepathic' or 'brain_action'."
|
||||
)
|
||||
|
||||
def test_brain_action_scenarios_exist(self):
|
||||
"""CRITICAL: Brain action extraction MUST be benchmarked."""
|
||||
with open(SCENARIOS_FILE) as f:
|
||||
data = json.load(f)
|
||||
|
||||
brain_scenarios = [s for s in data["scenarios"] if s.get("type") == "brain_action"]
|
||||
assert len(brain_scenarios) >= 3, (
|
||||
f"Only {len(brain_scenarios)} brain_action scenarios found. "
|
||||
f"Need at least 3 to reliably evaluate Brain action extraction."
|
||||
)
|
||||
|
||||
def test_brain_scenarios_have_available_actions(self):
|
||||
"""Brain scenarios must provide available_actions list."""
|
||||
with open(SCENARIOS_FILE) as f:
|
||||
data = json.load(f)
|
||||
|
||||
for scenario in data["scenarios"]:
|
||||
if scenario.get("type") != "brain_action":
|
||||
continue
|
||||
assert "available_actions" in scenario, f"Brain scenario '{scenario['id']}' missing 'available_actions'"
|
||||
assert "target_action" in scenario, f"Brain scenario '{scenario['id']}' missing 'target_action'"
|
||||
assert scenario["target_action"] in scenario["available_actions"], (
|
||||
f"Brain scenario '{scenario['id']}': target_action "
|
||||
f"'{scenario['target_action']}' not in available_actions"
|
||||
)
|
||||
|
||||
def test_telepathic_scenarios_have_nodes(self):
|
||||
"""Telepathic scenarios must provide nodes and target_index."""
|
||||
with open(SCENARIOS_FILE) as f:
|
||||
data = json.load(f)
|
||||
|
||||
for scenario in data["scenarios"]:
|
||||
if scenario.get("type") != "telepathic":
|
||||
continue
|
||||
assert "nodes" in scenario, f"Telepathic scenario '{scenario['id']}' missing 'nodes'"
|
||||
assert "target_index" in scenario, f"Telepathic scenario '{scenario['id']}' missing 'target_index'"
|
||||
|
||||
|
||||
class TestBenchmarkResultsIntegrity:
|
||||
"""Contract: Stored results must be consistent and comparable."""
|
||||
|
||||
@pytest.fixture
|
||||
def results(self):
|
||||
if not os.path.exists(RESULTS_FILE):
|
||||
pytest.skip("No benchmark results file yet")
|
||||
with open(RESULTS_FILE) as f:
|
||||
return json.load(f)
|
||||
|
||||
def test_details_format_is_consistent(self, results):
|
||||
"""All model details must use the same format (object, not raw int)."""
|
||||
for model_name, data in results.get("models", {}).items():
|
||||
details = data.get("details", {})
|
||||
for scenario_id, value in details.items():
|
||||
assert isinstance(value, dict), (
|
||||
f"Model '{model_name}' scenario '{scenario_id}' uses "
|
||||
f"legacy format (raw int: {value}). Must be "
|
||||
f"{{'avg_score': int, 'pass_rate': float, 'latency': int}}"
|
||||
)
|
||||
|
||||
def test_relative_performance_is_normalized(self, results):
|
||||
"""Relative performance must not exceed 100% (the leader)."""
|
||||
for model_name, data in results.get("models", {}).items():
|
||||
pct = data.get("relative_performance_pct", 0)
|
||||
assert pct <= 100.0, (
|
||||
f"Model '{model_name}' has relative_performance_pct={pct}% > 100%. "
|
||||
f"Scoring is not normalized by scenario count!"
|
||||
)
|
||||
Reference in New Issue
Block a user