⚡ 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())