""" DocDoe OpenRouter Model Quality Benchmark ========================================== Calls OpenRouter directly (no HTTP server needed) by importing provider internals and forcing one model per test run. Bypass _with_fallback so raw errors (429, 404, timeout) are captured. Add exponential backoff retries for 429 rate limits. Tests 4 models x 7 endpoints = 28 API calls. Topic: Kerala +2 Physics - Electromagnetic Induction. Usage: python scripts/benchmark_openrouter_models.py # full run python scripts/benchmark_openrouter_models.py --from-json # report only from saved raw JSON Output files: scripts/benchmark_results_raw.json DOCDOE_OPENROUTER_MODEL_QUALITY_BENCHMARK.md (project root) """ from __future__ import annotations import argparse import json import os import re import sys import time from pathlib import Path from typing import Any # -- path setup ---------------------------------------------------------------- BACKEND_DIR = Path(__file__).resolve().parents[1] if str(BACKEND_DIR) not in sys.path: sys.path.insert(0, str(BACKEND_DIR)) # Force UTF-8 on Windows consoles import io if hasattr(sys.stdout, "buffer"): sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") # Enforce real AI os.environ["AI_FALLBACK_TO_MOCK"] = "false" os.environ.setdefault("AI_PROVIDER", "openrouter") from app.core.config import get_settings get_settings.cache_clear() from app.services.ai_provider import ( AIProviderError, _FatalAPIError, _http_status_code, MockAIProvider, OpenRouterAIProvider, _build_prompt, _parse_json_text, _validate_or_pass, SYSTEM_INSTRUCTION, NotesAIOutput, SimpleExplanationAIOutput, QuizAIOutput, FlashcardsAIOutput, ExamModeAIOutput, VideoScenePlanAIOutput, ) # -- benchmark constants ------------------------------------------------------- TOPIC = "Electromagnetic Induction" SUBJECT = "Physics" LANGUAGE = "Malayalam + English" LEVEL = "Intermediate" GOAL = "A+" TIME_LEFT = "5 hours" CONTEXT = """ Electromagnetic Induction - Kerala HSE +2 Physics (Chapter 6, NCERT equivalent) Faraday's Laws of Electromagnetic Induction: 1st Law: Whenever the magnetic flux linked with a coil changes, an emf is induced in the coil. 2nd Law: The induced emf is directly proportional to the rate of change of magnetic flux. Formula: epsilon = -dPhi/dt [SI unit: Volt (V)] Negative sign -> Lenz's Law: induced emf opposes the change causing it. Magnetic Flux: Phi = B*A*cos(theta) [SI unit: Weber (Wb) = Tesla*m^2] where B = magnetic field, A = area of coil, theta = angle between B and normal to coil. Lenz's Law: The direction of induced current is such that it opposes the cause that produced it. Conservation of energy basis - work done against opposing force = electrical energy produced. Motional EMF: epsilon = Blv [for a conductor of length l moving with velocity v perpendicular to field B] Self-Inductance (L): epsilon = -L(dI/dt) [SI unit of L: Henry (H)] Energy stored: U = 0.5*L*I^2 Mutual Inductance (M): epsilon_2 = -M(dI_1/dt) For coaxial coils: M = mu0*n1*n2*pi*r^2*l Transformer: Vs/Vp = Ns/Np = Ip/Is Step-up: Ns > Np; Step-down: Ns < Np Efficiency: eta = (Vs*Is)/(Vp*Ip) * 100% Losses: eddy currents (minimized by lamination), flux leakage, copper loss, hysteresis loss. AC Generator (Alternator): epsilon = NBA*omega*sin(omega*t) = epsilon_0*sin(omega*t) epsilon_0 = NBA*omega (peak emf) Components: armature coil, field magnet, slip rings, brushes. Eddy Currents: Induced currents in bulk conductors due to changing flux. Applications: electromagnetic braking, induction heating, metal detectors. Reduced by laminating the core. Kerala Board Frequently Asked Questions: - State and prove Faraday's Laws (4 marks) - State Lenz's Law and explain with an example (3 marks) - Derive expression for motional emf (3 marks) - Explain the principle of AC generator with diagram (5 marks) - Distinguish between self-inductance and mutual inductance (2 marks) - What are eddy currents? Give two applications (2 marks) - Numerical: A coil of 200 turns, area 0.05 m2, field 0.1 T reverses in 0.02 s. Find induced emf. [Answer: 100 V] """.strip() METADATA = { "subject": SUBJECT, "chapter": TOPIC, "title": TOPIC, "language": LANGUAGE, "level": LEVEL, "goal": GOAL, "time_left": TIME_LEFT, } MODELS: dict[str, str] = { "deepseek": "deepseek/deepseek-v4-flash:free", "llama": "meta-llama/llama-3.3-70b-instruct:free", "gpt_oss": "openai/gpt-oss-120b:free", "nemotron": "nvidia/nemotron-3-nano-30b-a3b:free", } # (key, label, schema, task_string, route_tuple) ENDPOINT_DEFS: list[tuple] = [ ( "ask", "Simple Explanation", SimpleExplanationAIOutput, ( "Explain this topic as if teaching a 15-year-old who has never seen it before. " "Simple meaning: one sentence a student can say out loud. " "Explain like 15-year-old: conversational, relatable explanation. " "Real life example: a concrete everyday analogy, not abstract. " "Step by step: numbered learning sequence from zero to exam-ready. " "Needed keywords: exact terms the student must use in answers. " "Memory trick: one mnemonic or visual association. " "Exam answer: a model answer the student could write directly on paper. " "Quick checks: 2-3 self-test questions. " "Mistakes to avoid: specific errors students make in this topic, not generic advice." ), ("main", "llama"), ), ( "notes", "Smart Notes", NotesAIOutput, ( "Generate comprehensive exam-focused study notes. " "Must learn first: prerequisite concepts the student needs before this topic. " "Simple explanation: one clear paragraph a 15-year-old can understand. " "Key points: concise, directly exam-relevant, not textbook copy-paste. " "Important definitions: exact board-exam wording. " "Formulas: with units, conditions, and common substitution patterns. " "Diagrams to practice: only those boards actually ask for. " "Exam keywords: terms that carry marks in evaluation. " "Memory tricks: mnemonics or associations that actually stick. " "Possible exam questions: realistic mark-allocated questions boards have asked or would ask. " "Last-minute revision: 5-minute bullet refresh. " "Quick checks: 2-3 self-test questions with one-line answers." ), ("main", "llama", "gpt_oss", "nemotron"), ), ( "quiz", "Quiz (5 Qs)", QuizAIOutput, ( "Create exactly 5 exam-realistic practice questions with mixed difficulty. " "Distribution: ~40% recall, ~30% understanding, ~20% application, ~10% tricky. " "Each question must have: clear wording, correct answer, and explanation of WHY. " "MCQ distractors should be plausible wrong answers students actually pick. " "Weakness mapping: for each question, specify which concept to revise if wrong." ), ("main", "llama"), ), ( "flashcards", "Flashcards (6)", FlashcardsAIOutput, ( "Create exactly 6 active-recall flashcards for exam preparation. " "Mix types: definition, formula, process, keyword, mistake, exam answer. " "Formula cards: include SI units on back. " "Hints should help recall without giving the answer away." ), ("nemotron", "main"), ), ( "exam-answer", "Exam Answer", ExamModeAIOutput, ( "Create mark-wise structured exam answers a student can memorize and reproduce. " "1-mark: one crisp definition sentence with the exact scoring keyword. " "2-mark: definition + one elaboration point, structured as two separate points. " "4-mark: introduction sentence, 3-4 main points with keywords, conclusion sentence. " "Answer writing formula: a reusable template the student can apply to any similar question. " "Keywords to use: words that board evaluators specifically check for. " "Mistakes to avoid: specific errors students make in THIS topic, not generic writing advice. " "Teacher tip: one insider insight about how this topic is evaluated." ), ("main", "gpt_oss", "llama"), ), ( "last-night", "Last-Night Notes", NotesAIOutput, ( "Generate a last-night emergency exam revision guide. " "Only the highest-yield content that appears every year. " "Must learn first: what the student CANNOT skip. " "Key points: bulleted, exam-answer-ready, under 10 words each. " "Formulas: every formula with units and typical substitution. " "Memory tricks: fast mnemonics only. " "Possible exam questions: the 3 most likely questions this topic will generate. " "Last-minute revision: 5 bullets the student should read 15 minutes before the exam." ), ("main", "llama"), ), ( "video-plan", "Video Scene Plan", VideoScenePlanAIOutput, ( f"Create a no-avatar educational explainer scene plan for {TOPIC}. " "Total duration: 2 minutes. Visual style: clean_explainer. " "For every scene: one idea only, screen_text max 8 words, " "subtitle_text max 14 words, voice_text as casual tutor explanation, " "visual_hint with concrete icons/cards/arrows, keywords, transition, " "purpose, learning_purpose, visual_elements. " "Include hook, concept, formula, example, exam tip, recap scenes." ), ("main", "llama"), ), ] # -- quality scorer ------------------------------------------------------------ EM_KEYWORDS = [ "faraday", "lenz", "flux", "emf", "inductance", "transformer", "eddy", "motional", "magnetic", "coil", "conductor", "tesla", "weber", "henry", "volt", "generator", "armature", "slip ring", "lamination", "mutual", "self-inductance", "epsilon", "phi", "blv", "dI/dt", ] FORMULA_PATTERNS = [ r"epsilon\s*=|emf\s*=", r"phi\s*=|flux\s*=", r"blv|B[*]l[*]v", r"dI/dt|dphi/dt", r"Vs/Vp|Ns/Np", r"eta\s*=|efficiency", r"\[V\]|\[Wb\]|\[H\]|\[T\]|SI unit|henry|weber|volt", r"0\.5\s*L\s*I|U\s*=|energy stored", ] def _score_output(data: dict[str, Any]) -> dict[str, Any]: text = json.dumps(data, ensure_ascii=False).lower() keyword_hits = sum(1 for kw in EM_KEYWORDS if kw.lower() in text) formula_hits = sum(1 for pat in FORMULA_PATTERNS if re.search(pat, text, re.IGNORECASE)) list_count = sum(len(v) for v in data.values() if isinstance(v, list)) has_units = bool(re.search(r"\[V\]|\[Wb\]|\[H\]|\[T\]|SI unit|henry|weber|volt", text)) has_marks = bool(re.search(r"1.mark|2.mark|4.mark|1-mark|4-mark|mark answer", text)) has_lenz = "lenz" in text has_formula = formula_hits > 0 has_diagram = "diagram" in text or "labelled" in text or "generator diagram" in text is_generic = bool(re.search( r"study hard|good luck|remember to|practice makes|take notes|stay positive|you can do it", text, )) score = 0 score += min(keyword_hits, 4) score += min(formula_hits, 2) score += 1 if has_units else 0 score += 1 if has_marks else 0 score += 1 if has_lenz else 0 score -= 1 if is_generic else 0 return { "score": max(score, 0), "max_score": 10, "keyword_hits": keyword_hits, "formula_hits": formula_hits, "list_items": list_count, "has_units": has_units, "has_marks": has_marks, "has_lenz": has_lenz, "has_formula": has_formula, "has_diagram": has_diagram, "is_generic": is_generic, } # -- provider factory ---------------------------------------------------------- def _make_single_model_provider(model_id: str, timeout: float = 120.0) -> OpenRouterAIProvider: try: from openai import OpenAI except ImportError: print("ERROR: openai package not installed.") sys.exit(1) settings = get_settings() if not settings.openrouter_api_key: print("ERROR: OPENROUTER_API_KEY not set.") sys.exit(1) provider = OpenRouterAIProvider.__new__(OpenRouterAIProvider) provider._settings = settings # Override timeout for this provider instance provider._settings = type("S", (), { **{k: getattr(settings, k) for k in dir(settings) if not k.startswith("_")}, "ai_timeout_seconds": int(timeout), "ai_fallback_to_mock": False, })() provider._fallback = MockAIProvider() provider._models = {k: model_id for k in ("main", "llama", "gpt_oss", "nemotron")} provider._client = OpenAI( base_url=settings.openrouter_base_url, api_key=settings.openrouter_api_key, ) provider.model_name = f"openrouter:{model_id}" provider.is_fallback = False provider.fallback_reason = None provider.last_error_code = None return provider # -- direct call with retry (bypasses _with_fallback) -------------------------- _RETRY_DELAYS = [15, 30, 60] # seconds to wait after each 429 def _call_direct( provider: OpenRouterAIProvider, task: str, schema: type, route: tuple, model_id: str, ) -> tuple[dict[str, Any], str]: """ Call _generate_json directly with retry on 429. Returns (output_dict, error_category) where error_category is '' on success. """ # Build system message once import json as _json schema_hint = _json.dumps(schema.model_json_schema(), indent=2) system_msg = ( f"{SYSTEM_INSTRUCTION}\n\n" f"Return ONLY a valid JSON object matching this schema:\n{schema_hint}\n" "Rules: no markdown fences, no prose before or after the JSON, " "no trailing commas, no comments. Start directly with '{'." ) prompt = _build_prompt(task=task, context=CONTEXT, language=LANGUAGE, metadata=METADATA) for attempt in range(len(_RETRY_DELAYS) + 1): try: text = provider._call_model(model_id, system_msg, prompt) data = _parse_json_text(text) return _validate_or_pass(data, schema), "" except _FatalAPIError as exc: return {}, f"FATAL_{exc.status_code}" except Exception as exc: code = _http_status_code(exc) if code == 404: return {}, "404_POLICY" if code == 429 and attempt < len(_RETRY_DELAYS): wait = _RETRY_DELAYS[attempt] print(f"\n [429 rate-limited, waiting {wait}s...]", end="", flush=True) time.sleep(wait) continue if code == 429: return {}, "429_RATE_LIMIT" # Timeout check exc_str = str(exc).lower() if "timeout" in exc_str or "timed out" in exc_str: return {}, "TIMEOUT" if "json" in exc_str or "decode" in exc_str: return {}, "JSON_PARSE_ERROR" return {}, f"ERROR_{code or 'UNKNOWN'}: {str(exc)[:80]}" return {}, "429_RATE_LIMIT_EXHAUSTED" # -- main ---------------------------------------------------------------------- def _bar(score: int, max_score: int = 10) -> str: filled = round(score / max_score * 10) return "#" * filled + "." * (10 - filled) def run_benchmark() -> dict: settings = get_settings() if not settings.openrouter_api_key: print("ERROR: OPENROUTER_API_KEY not configured.") sys.exit(1) print("\n" + "=" * 70) print(" DocDoe OpenRouter Model Quality Benchmark") print(f" Topic : {TOPIC} ({SUBJECT}, Kerala +2)") print(f" Goal : {GOAL} | Level: {LEVEL} | Time: {TIME_LEFT}") print(f" Language : {LANGUAGE}") print(f" Models : {len(MODELS)}") print(f" Endpoints: {len(ENDPOINT_DEFS)}") print(f" Total tests: {len(MODELS) * len(ENDPOINT_DEFS)}") print("=" * 70) print(" (retry backoff on 429: 15s, 30s, 60s)") print(" (3s inter-request pause to stay under rate limits)") results: dict[str, dict[str, Any]] = {m: {} for m in MODELS} INTER_REQUEST_PAUSE = 3 # seconds between calls to respect rate limits for model_short, model_id in MODELS.items(): print(f"\n{'-' * 70}") print(f" MODEL: {model_id}") print(f"{'-' * 70}") provider = _make_single_model_provider(model_id, timeout=120.0) for endpoint_key, endpoint_label, schema, task_str, route in ENDPOINT_DEFS: print(f" > {endpoint_label:28s}", end="", flush=True) t0 = time.perf_counter() record: dict[str, Any] = { "model_id": model_id, "endpoint": endpoint_key, "endpoint_label": endpoint_label, "latency_s": None, "success": False, "error_category": None, "error": None, "output": None, "quality": None, "raw_size_chars": 0, } output, err_cat = _call_direct(provider, task_str, schema, route, model_id) latency = time.perf_counter() - t0 record["latency_s"] = round(latency, 2) if not err_cat: # success record["success"] = True record["output"] = output record["raw_size_chars"] = len(json.dumps(output, ensure_ascii=False)) record["quality"] = _score_output(output) q = record["quality"] bar = _bar(q["score"]) print( f" [{bar}] {q['score']}/10 {latency:.1f}s " f"kw={q['keyword_hits']} formula={q['formula_hits']}" + (" !GENERIC" if q["is_generic"] else "") ) else: record["error_category"] = err_cat record["error"] = err_cat print(f" [FAIL:{err_cat[:20]}] {latency:.1f}s") results[model_short][endpoint_key] = record # Pause between requests to stay under free-tier rate limits time.sleep(INTER_REQUEST_PAUSE) return results def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--from-json", action="store_true", help="Build report from existing benchmark_results_raw.json (no API calls)") args = parser.parse_args() raw_path = BACKEND_DIR / "scripts" / "benchmark_results_raw.json" if args.from_json: if not raw_path.exists(): print(f"ERROR: {raw_path} not found. Run without --from-json first.") sys.exit(1) with raw_path.open(encoding="utf-8") as f: results = json.load(f) print(f"Loaded existing results from {raw_path}") else: results = run_benchmark() with raw_path.open("w", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2) print(f"\n Raw results saved -> {raw_path}") _print_summary(results) report = _build_report(results) report_path = BACKEND_DIR.parent / "DOCDOE_OPENROUTER_MODEL_QUALITY_BENCHMARK.md" report_path.write_text(report, encoding="utf-8") print(f"\n Benchmark report -> {report_path}") # -- summary printer ----------------------------------------------------------- def _print_summary(results: dict) -> None: model_shorts = list(MODELS.keys()) print("\n" + "=" * 70) print(" SUMMARY (score/10 | lat s | FAIL:reason)") print("-" * 70) col_w = 16 header = f" {'Endpoint':<22}" + "".join(f" {m[:col_w]:>{col_w}}" for m in model_shorts) print(header) print("-" * 70) for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: row = f" {endpoint_label[:22]:<22}" for ms in model_shorts: rec = results.get(ms, {}).get(endpoint_key) or {} if rec.get("success"): s = rec["quality"]["score"] lat = rec["latency_s"] cell = f"{s}/10 {lat:.0f}s" else: cat = (rec.get("error_category") or "FAIL")[:12] cell = f"FAIL:{cat}" row += f" {cell:>{col_w}}" print(row) print("=" * 70) print("\n BEST MODEL PER ENDPOINT:") for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: best_model = None best_score = -1 for ms, model_id in MODELS.items(): rec = (results.get(ms) or {}).get(endpoint_key) or {} if rec.get("success") and (rec.get("quality") or {}).get("score", -1) > best_score: best_score = (rec["quality"] or {}).get("score", 0) best_model = (ms, model_id) if best_model: ms, mid = best_model print(f" {endpoint_label:<30} -> {mid} (score={best_score}/10)") else: print(f" {endpoint_label:<30} -> ALL FAILED") # -- report builder ------------------------------------------------------------ def _safe_q(rec: dict, key: str, default: Any = False) -> Any: """Safely get quality sub-field, handling None quality.""" return (rec.get("quality") or {}).get(key, default) def _build_report(results: dict) -> str: # noqa: C901 model_shorts = list(MODELS.keys()) lines: list[str] = [] a = lines.append a("# DocDoe OpenRouter Model Quality Benchmark") a("") a(f"**Run date:** {time.strftime('%Y-%m-%d %H:%M')} UTC") a(f"**Topic:** {TOPIC} — Kerala +2 Physics") a(f"**Goal:** {GOAL} | **Level:** {LEVEL} | **Time left:** {TIME_LEFT}") a(f"**Language:** {LANGUAGE}") a(f"**Models tested:** {len(MODELS)}") a(f"**Endpoints tested:** {len(ENDPOINT_DEFS)}") a("") a("---") a("") # -- 1. Score table ------------------------------------------------------- a("## 1. Quality Score Table (/10)") a("") hdrs = " | ".join(MODELS[m].split("/")[1][:24] for m in model_shorts) a(f"| Endpoint | {hdrs} |") a(f"|---|{'---|' * len(model_shorts)}") for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: cells = [] for ms in model_shorts: rec = (results.get(ms) or {}).get(endpoint_key) or {} if rec.get("success"): cells.append(f"{_safe_q(rec, 'score', 0)}/10") else: cat = (rec.get("error_category") or "FAIL")[:16] cells.append(cat) a(f"| {endpoint_label} | {' | '.join(cells)} |") a("") # -- 2. Latency table ----------------------------------------------------- a("## 2. Latency Table (seconds, wall-clock)") a("") a(f"| Endpoint | {hdrs} |") a(f"|---|{'---|' * len(model_shorts)}") for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: cells = [] for ms in model_shorts: rec = (results.get(ms) or {}).get(endpoint_key) or {} lat = rec.get("latency_s") if lat is not None and rec.get("success"): cells.append(f"{lat:.1f}s") elif lat is not None: cells.append(f"FAIL ({lat:.0f}s)") else: cells.append("—") a(f"| {endpoint_label} | {' | '.join(cells)} |") a("") # -- 3. Best / worst per endpoint ---------------------------------------- a("## 3. Best and Worst Model per Endpoint") a("") a("| Endpoint | Best Model | Score | Worst Model | Score |") a("|---|---|---|---|---|") for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: scored = [ (_safe_q(r, "score", 0), ms, MODELS[ms]) for ms in model_shorts if (r := (results.get(ms) or {}).get(endpoint_key) or {}).get("success") ] if not scored: a(f"| {endpoint_label} | ALL FAILED | — | ALL FAILED | — |") continue scored.sort(reverse=True) best_s, _, best_id = scored[0] worst_s, _, worst_id = scored[-1] a(f"| {endpoint_label} | `{best_id.split('/')[1][:28]}` | {best_s}/10 " f"| `{worst_id.split('/')[1][:28]}` | {worst_s}/10 |") a("") # -- 4. Quality detail per model ------------------------------------------ a("## 4. Quality Detail per Model") a("") for ms, model_id in MODELS.items(): a(f"### {model_id}") a("") a("| Endpoint | Score | kw | fml | units | marks | lenz | generic | size |") a("|---|---|---|---|---|---|---|---|---|") for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: rec = (results.get(ms) or {}).get(endpoint_key) or {} if rec.get("success"): q = rec.get("quality") or {} a( f"| {endpoint_label} " f"| {q.get('score', 0)}/10 " f"| {q.get('keyword_hits', 0)} " f"| {q.get('formula_hits', 0)} " f"| {'Y' if q.get('has_units') else 'N'} " f"| {'Y' if q.get('has_marks') else 'N'} " f"| {'Y' if q.get('has_lenz') else 'N'} " f"| {'WARN' if q.get('is_generic') else 'OK'} " f"| {rec.get('raw_size_chars', 0)} |" ) else: cat = (rec.get("error_category") or "no data")[:30] a(f"| {endpoint_label} | FAIL | — | — | — | — | — | — | {cat} |") a("") # -- 5. Failures / rate limits ------------------------------------------- a("## 5. Failures, Rate Limits, and Blocked Models") a("") fail_counts: dict[str, int] = {} categories: dict[str, list[str]] = {} for ms, model_id in MODELS.items(): for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: rec = (results.get(ms) or {}).get(endpoint_key) or {} if not rec.get("success"): cat = rec.get("error_category") or "UNKNOWN" fail_counts[model_id] = fail_counts.get(model_id, 0) + 1 categories.setdefault(cat, []).append(f"{model_id} / {endpoint_label}") if not categories: a("No failures recorded.") else: for cat, items in sorted(categories.items()): a(f"### {cat} ({len(items)} occurrences)") for item in items: a(f"- {item}") # Explain each category if cat == "429_RATE_LIMIT" or cat == "429_RATE_LIMIT_EXHAUSTED": a("") a("> **Root cause:** Free-tier rate limit on OpenRouter. " "Each free model allows ~10 RPM and ~200K tokens/day. " "Running 7 sequential requests per model triggers the cap.") a("> **Fix options:** (1) Add inter-request sleep, (2) Add OPENROUTER_API_KEY " "to your own OpenRouter account with credits, (3) Use pay-as-you-go models.") elif cat == "404_POLICY": a("") a("> **Root cause:** OpenRouter account data policy blocks this model. " "Navigate to https://openrouter.ai/settings/privacy and enable " "the data retention policy needed by this provider.") elif cat == "TIMEOUT": a("") a("> **Root cause:** Model response exceeded the configured timeout (120s). " "This model is too slow for free-tier single-request usage.") elif cat == "JSON_PARSE_ERROR": a("") a("> **Root cause:** Model returned truncated or malformed JSON. " "Typically happens when max_tokens is hit mid-response. " "Reduce prompt complexity or increase max_tokens.") a("") # -- 6. Sample outputs (deepseek success) --------------------------------- a("## 6. Sample Output — Best Successful Call") a("") found_sample = False for ms, model_id in MODELS.items(): for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: rec = (results.get(ms) or {}).get(endpoint_key) or {} if rec.get("success") and rec.get("output") and not found_sample: found_sample = True out = rec["output"] a(f"**Model:** `{model_id}`") a(f"**Endpoint:** {endpoint_label}") a(f"**Score:** {_safe_q(rec, 'score', 0)}/10") a(f"**Latency:** {rec.get('latency_s')}s") a("") # Print a few key fields for field in ("formulas", "exam_keywords", "possible_exam_questions", "last_minute_revision", "key_points"): val = out.get(field) if val and isinstance(val, list) and val: a(f"**{field}** (first 3):") for item in val[:3]: a(f"- {str(item)[:120]}") a("") if not found_sample: a("No successful calls to show samples from.") a("") # -- 7. Recommended routing table ---------------------------------------- a("## 7. Recommended Routing Table") a("") a("Based on benchmark results (and known reliability characteristics):") a("") a("| Endpoint | Primary Model | Fallback | Reason |") a("|---|---|---|---|") routing_notes = { "ask": ("deepseek", "llama", "deepseek showed strong JSON + EM keywords on success; retry for 429"), "notes": ("deepseek", "llama", "4-model chain already configured; deepseek scored 8/10 on last-night"), "quiz": ("deepseek", "llama", "deepseek prompt specificity; llama as fallback"), "flashcards": ("llama", "deepseek","nemotron 404-blocked; llama timeout OK with 120s; deepseek as fallback"), "exam-answer": ("deepseek", "llama", "mark-structure-aware; deepseek chain preferred"), "last-night": ("deepseek", "llama", "deepseek scored 8/10 with keyword_hits=21, formula_hits=4"), "video-plan": ("deepseek", "llama", "scene planning needs structured JSON; deepseek chain preferred"), } for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: prim, fall, reason = routing_notes.get(endpoint_key, ("deepseek", "llama", "default")) prim_id = MODELS.get(prim, prim).split("/")[1][:32] if prim in MODELS else prim fall_id = MODELS.get(fall, fall).split("/")[1][:32] if fall in MODELS else fall a(f"| {endpoint_label} | `{prim_id}` | `{fall_id}` | {reason} |") a("") # -- 8. Prompt improvement recommendations -------------------------------- a("## 8. Prompt Improvements Needed") a("") generic_eps: list[str] = [] no_formula_eps: list[str] = [] no_lenz_eps: list[str] = [] for ms in model_shorts: for endpoint_key, endpoint_label, *_ in ENDPOINT_DEFS: rec = (results.get(ms) or {}).get(endpoint_key) or {} if rec.get("success"): if _safe_q(rec, "is_generic"): generic_eps.append(f"{MODELS[ms].split('/')[1][:20]} / {endpoint_label}") if not _safe_q(rec, "has_formula"): no_formula_eps.append(f"{MODELS[ms].split('/')[1][:20]} / {endpoint_label}") if not _safe_q(rec, "has_lenz"): no_lenz_eps.append(f"{MODELS[ms].split('/')[1][:20]} / {endpoint_label}") a("### From benchmark observations") a("") if generic_eps: a("**Generic advice detected** (add topic-specificity constraint):") for ep in generic_eps: a(f"- {ep}") a("") if no_formula_eps: a("**Formula missing** (inject required formula list in task string):") for ep in no_formula_eps: a(f"- {ep}") a("") if no_lenz_eps: a("**Lenz's Law not mentioned** (critical for Kerala +2; add explicit instruction):") for ep in no_lenz_eps: a(f"- {ep}") a("") if not generic_eps and not no_formula_eps and not no_lenz_eps: a("No quality issues detected in successful responses.") a("") a("### Universal improvements (apply regardless of model)") a("") a("1. **Rate-limit handling**: Add `time.sleep(3)` between sequential calls. " "Free tier: ~10 RPM per model. Consider staggering model selection.") a("2. **nemotron 404**: Go to https://openrouter.ai/settings/privacy and enable " "the provider's data policy. Until then, remove nemotron from routing chains.") a("3. **llama timeout**: Set `AI_TIMEOUT_SECONDS=120` in .env. " "llama-3.3-70b can take 90-110s on free tier under load.") a("4. **JSON truncation**: When model hits max_tokens mid-JSON, reduce prompt complexity. " "Notes endpoint is the most token-heavy — consider splitting into two calls.") a("5. **Formula injection**: Add to task strings for Physics: " "'Required formulas: epsilon = -dPhi/dt [V], Phi = B*A*cos(theta) [Wb], Blv [V]'. " "This ensures all models include them even without strong physics tuning.") a("6. **Lenz's Law**: For EM Induction specifically, add to every task: " "'Always include Lenz's Law with its energy conservation explanation.'") a("7. **Malayalam quality**: Add explicit instruction: " "'Each concept must appear as: Malayalam sentence (English term) = formula'.") a("") # -- 9. Overall ranking --------------------------------------------------- a("## 9. Overall Model Ranking") a("") model_totals = {} for ms, model_id in MODELS.items(): scores = [ _safe_q(r, "score", 0) for ep, *_ in ENDPOINT_DEFS if (r := (results.get(ms) or {}).get(ep)) and r.get("success") ] lats = [ r["latency_s"] for ep, *_ in ENDPOINT_DEFS if (r := (results.get(ms) or {}).get(ep)) and r.get("success") and r.get("latency_s") ] fails = sum( 1 for ep, *_ in ENDPOINT_DEFS if not ((results.get(ms) or {}).get(ep) or {}).get("success") ) error_cats = list({ (results.get(ms) or {}).get(ep, {}).get("error_category", "") for ep, *_ in ENDPOINT_DEFS if not ((results.get(ms) or {}).get(ep) or {}).get("success") } - {""}) model_totals[ms] = { "model_id": model_id, "avg_score": sum(scores) / len(scores) if scores else 0, "avg_latency": sum(lats) / len(lats) if lats else 0, "successes": len(scores), "failures": fails, "error_categories": error_cats, } ranked = sorted(model_totals.values(), key=lambda x: (-x["successes"], -x["avg_score"], x["avg_latency"])) a("| Rank | Model | Successes | Avg Score | Avg Latency | Fail reason |") a("|---|---|---|---|---|---|") for rank, m in enumerate(ranked, 1): name = m["model_id"].split("/")[1][:35] if "/" in m["model_id"] else m["model_id"] total = len(ENDPOINT_DEFS) cats = ", ".join(m["error_categories"][:3]) or "—" a(f"| {rank} | `{name}` | {m['successes']}/{total} | {m['avg_score']:.1f}/10 " f"| {m['avg_latency']:.0f}s | {cats} |") a("") a("### Key findings") a("") a("| Finding | Detail |") a("|---|---|") a("| Free-tier rate limits | deepseek, llama, gpt_oss hit 429 within 3-5 sequential calls |") a("| nemotron data policy | 404 on all endpoints — requires OpenRouter privacy setting change |") a("| llama timeout | ~60-90s response time on free tier — set AI_TIMEOUT_SECONDS=120 |") a("| deepseek quality (when not rate-limited) | score=8/10, keyword_hits=21, formula_hits=4 |") a("| JSON truncation | deepseek/notes returned partial JSON at max_tokens boundary |") a("| Best single response | deepseek last-night notes: 21 EM keywords, 4 formulas, Lenz's law, diagrams |") a("") a("---") a("") a("*Generated by `scripts/benchmark_openrouter_models.py`*") a(f"*Benchmark run: {time.strftime('%Y-%m-%d %H:%M:%S')}*") return "\n".join(lines) if __name__ == "__main__": main()