From dd8285e1cea3444b1c3c2fbf340ccd3c96ea4855 Mon Sep 17 00:00:00 2001 From: Marc Mintel Date: Wed, 29 Apr 2026 00:14:29 +0200 Subject: [PATCH] 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. --- benchmarks/run_competitive_benchmark.py | 255 +++++++++++++++--------- tests/unit/test_benchmark_integrity.py | 113 +++++++++++ 2 files changed, 278 insertions(+), 90 deletions(-) create mode 100644 tests/unit/test_benchmark_integrity.py diff --git a/benchmarks/run_competitive_benchmark.py b/benchmarks/run_competitive_benchmark.py index 47ccbe1..0469711 100644 --- a/benchmarks/run_competitive_benchmark.py +++ b/benchmarks/run_competitive_benchmark.py @@ -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() diff --git a/tests/unit/test_benchmark_integrity.py b/tests/unit/test_benchmark_integrity.py new file mode 100644 index 0000000..e51be81 --- /dev/null +++ b/tests/unit/test_benchmark_integrity.py @@ -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!" + )