⚡ Optimize benchmark_classifier.py with ThreadPoolExecutor

README.md

@@ -213,7 +213,7 @@ All configurations, automation scripts, and databases are self-contained within
 
 ```
 /home/gpav/Vrac/LAB/AI/LLM-Routing/
-├── .env # Environment file for OpenRouter API Key (ignored by git)
+├── .env # Environment file for API keys, passwords, and generated secrets (ignored by git)
 ├── .gitignore # Git ignore policy protecting secrets & database files
 ├── README.md # In-depth system and operational guide
 ├── pod.yaml # Podman Kubernetes template defining the 10-container stack

router/Dockerfile

@@ -3,7 +3,7 @@ FROM python:3.14-slim
 WORKDIR /app
 
 # Install deps in a single layer (no pip cache, no extra files)
-RUN pip install --no-cache-dir fastapi "pydantic>=2.0,<3.0" uvicorn httpx pyyaml python-multipart asyncpg langfuse redis
+RUN pip install --no-cache-dir fastapi "pydantic>=2.0,<3.0" uvicorn httpx pyyaml python-multipart asyncpg langfuse redis aiofiles
 
 # Copy all source in one layer — removes dead config COPY (volume-mounted at runtime)
 COPY main.py agy_proxy.py circuit_breaker.py aa_scores.json free_models_roster.json /app/

router/main.py

@@ -1452,8 +1452,8 @@ async def proxy_models():
  {"id": "llm-routing-agy", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 1048576},
  {"id": "llm-routing-ollama", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 524288},
  ]
-data["data"] = routing_models + data["data"]
-
+for entry in reversed(routing_models):
+ data["data"].insert(0, entry)
  return JSONResponse(content=data, status_code=200)
  except Exception as parse_err:
  logger.warning(f"Failed to parse /v1/models JSON despite status 200: {parse_err}")
@@ -3382,23 +3382,10 @@ class AnnotationItem(BaseModel):
 # the atomic file-replace mechanism, which is acceptable for this dashboard feature.
 annotations_lock = asyncio.Lock()
 
-_annotations_cache = {}
 
 def _read_annotations_sync(path) -> dict:
- import copy
-
- # Do not swallow OSError if file doesn't exist to preserve original behavior.
- # The caller (save_annotations) handles the exception when reading existing annotations.
- current_mtime = os.path.getmtime(path)
-
- cache_entry = _annotations_cache.get(path)
-
- if cache_entry is None or current_mtime != cache_entry["mtime"]:
- with open(path, "r", encoding="utf-8") as f:
- data = json.load(f)
- _annotations_cache[path] = {"mtime": current_mtime, "data": data}
-
- return copy.deepcopy(_annotations_cache[path]["data"])
+ with open(path, "r", encoding="utf-8") as f:
+ return json.load(f)
 
 @app.post("/dashboard/save-annotations")
 async def save_annotations(payload: Dict[str, AnnotationItem]):

scripts/benchmark_classifier.py

@@ -1,5 +1,7 @@
 """Benchmark gemma4-26a4b-routing classifier against labeled dataset."""
 import os
+import concurrent.futures
+import threading
 import json, urllib.request, time, sys
 from collections import defaultdict, Counter
 from pathlib import Path
@@ -54,7 +56,8 @@ def classify(prompt):
 per_tier = {t: {"correct": 0, "total": 0} for t in TIERS}
 confusion = defaultdict(Counter) # confusion[expected][predicted]
 
-for i, item in enumerate(dataset.get("prompts", [])):
+
+def process_item(item):
  prompt = item["prompt"]
  # Support both old schema ("tier") and new schema ("llm_tier" / "clf_tier")
  expected = item.get("tier") or item.get("llm_tier") or item.get("clf_tier", "")
@@ -64,31 +67,57 @@ def classify(prompt):
  except Exception as e:
  predicted = f"ERROR: {str(e)[:50]}"
 
- results.append({
- "prompt": prompt[:100],
- "expected": expected,
- "predicted": predicted,
- })
-
- # Only score against known tiers — skip ERROR/unknown labels gracefully
- if expected not in per_tier:
- confusion[expected][predicted] += 1
- continue
+ return expected, predicted
 
- per_tier[expected]["total"] += 1
- if predicted == expected:
- correct += 1
- per_tier[expected]["correct"] += 1
+results_list = [None] * total
 
- confusion[expected][predicted] += 1
+with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
+ # Submit all tasks immediately, but executor map will process them and yield results in order
+ # To maintain rate limit properly without the quadratic delay or blocking progress, we can use a threading.Lock
+ rate_limit_lock = threading.Lock()
 
- # Progress
- if (i + 1) % 20 == 0:
- scored_so_far = sum(t["total"] for t in per_tier.values())
- acc = (correct / scored_so_far * 100) if scored_so_far > 0 else 0.0
- print(f" {i+1}/{total} — accuracy {acc:.1f}%")
+ def process_item_with_rate_limit(index_and_item):
+ i, item = index_and_item
+ # Enforce rate limit globally across all threads
+ with rate_limit_lock:
+ time.sleep(0.05)
+
+ expected, predicted = process_item(item)
+ return i, item, expected, predicted
+
+ # We map over items, but map blocks if we process in order. We can use as_completed and store by index to preserve order.
+ futures = [executor.submit(process_item_with_rate_limit, (i, item)) for i, item in enumerate(dataset.get("prompts", []))]
 
- time.sleep(0.05) # minimal rate-limit (model handles concurrency via llama-server slots)
+ completed_count = 0
+ for future in concurrent.futures.as_completed(futures):
+ i, item, expected, predicted = future.result()
+
+ results_list[i] = {
+ "prompt": item["prompt"][:100],
+ "expected": expected,
+ "predicted": predicted,
+ }
+
+ # Only score against known tiers — skip ERROR/unknown labels gracefully
+ if expected not in per_tier:
+ confusion[expected][predicted] += 1
+ else:
+ per_tier[expected]["total"] += 1
+ if predicted == expected:
+ correct += 1
+ per_tier[expected]["correct"] += 1
+ confusion[expected][predicted] += 1
+
+ completed_count += 1
+
+ # Progress
+ if completed_count % 20 == 0:
+ scored_so_far = sum(t["total"] for t in per_tier.values())
+ acc = (correct / scored_so_far * 100) if scored_so_far > 0 else 0.0
+ print(f" {completed_count}/{total} — accuracy {acc:.1f}%")
+
+# Filter out Nones if any
+results = [r for r in results_list if r is not None]
 
 # Report
 scored_total = sum(t["total"] for t in per_tier.values())