diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 4f44831c..37aa7f1f 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -2,7 +2,7 @@ version: 2 updates: # Docker image dependency updates - package-ecosystem: "docker" - directory: "/" + directory: "/router" schedule: interval: "daily" time: "18:22" @@ -11,7 +11,6 @@ updates: separator: "/" commit-message: prefix: "chore(deps)" - prefix-major: "chore(deps)!" include: "scope" labels: - "dependencies" @@ -20,11 +19,6 @@ updates: - "sheepdestroyer" allow: - dependency-type: "direct" - ignore: - # Ignore major version updates for critical services - # Uncomment and adjust as needed - # - dependency-name: "postgres" - # update-types: ["version-update:semver-major"] # GitHub Actions dependency updates (optional - add if using GH Actions) - package-ecosystem: "github-actions" diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 00000000..93a280fb --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,28 @@ +name: Run Tests + +on: + push: + branches: [ master ] + pull_request: + branches: [ master ] + +jobs: + test: + runs-on: ubuntu-latest + steps: + - name: Checkout Repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Run Circuit Breaker Tests + run: python3 test_circuit_breaker.py + + - name: Run Breaker Verification + run: python3 verify_breaker.py + + - name: Run Integration Verification + run: python3 test_a2_verify.py diff --git a/.jules/bolt.md b/.jules/bolt.md new file mode 100644 index 00000000..f2935c10 --- /dev/null +++ b/.jules/bolt.md @@ -0,0 +1,3 @@ +## 2024-06-16 - Synchronous I/O in Async API Handlers +**Learning:** `save_persisted_stats()` was being called synchronously on every API request, cache hit, and tool usage log, triggering blocking disk I/O in the main event loop. +**Action:** Always throttle or batch background telemetry writes in async Python applications to prevent blocking the event loop under load. diff --git a/pod.yaml b/pod.yaml index 9f47601b..bad4946c 100644 --- a/pod.yaml +++ b/pod.yaml @@ -127,6 +127,8 @@ spec: volumeMounts: - mountPath: /config/router_dir name: router-config + - mountPath: /app/data + name: dataset-data - mountPath: /config/gemini_auth name: gemini-secrets - mountPath: /root/.gemini @@ -466,3 +468,7 @@ spec: - hostPath: path: /home/gpav/Vrac/LAB/AI/LLM-Routing name: env-file + - hostPath: + path: /home/gpav/Vrac/LAB/AI/LLM-Routing/data + type: DirectoryOrCreate + name: dataset-data diff --git a/router/Containerfile b/router/Containerfile index b21bf253..c26ad091 100644 --- a/router/Containerfile +++ b/router/Containerfile @@ -6,7 +6,8 @@ WORKDIR /app RUN pip install --no-cache-dir fastapi uvicorn httpx pyyaml python-multipart asyncpg langfuse # 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 best_free_model.json /app/ +COPY main.py agy_proxy.py circuit_breaker.py aa_scores.json free_models_roster.json /app/ +COPY static/ /app/static/ EXPOSE 5000 diff --git a/router/agy_proxy.py b/router/agy_proxy.py index 0a2c03c2..918bfcb6 100644 --- a/router/agy_proxy.py +++ b/router/agy_proxy.py @@ -253,7 +253,7 @@ async def try_agy_proxy(prompt: str, messages: list = None, # Determine which breaker to use for this tier # Tier 0 (idx 0): gemini-3.5-flash → google_breaker # Tier 1 (idx 1): claude-opus-4.6 → vendor_breaker - is_google_tier = (actual_tier_idx == 0) + is_google_tier = "gemini" in tier.get("model_name", "").lower() tier_breaker = google_breaker if is_google_tier else vendor_breaker if not tier_breaker.is_allowed(): @@ -323,8 +323,8 @@ async def try_agy_proxy(prompt: str, messages: list = None, # Success! Stream has started. tier_breaker.record_success() - # Define the async generator async def token_generator(stream_resp, httpx_client, initial_line, current_conv_id): + """Asynchronously yields tokens from the agy daemon stream and manages session state updates.""" # Yield the initial token if it was a token init_data = json.loads(initial_line) if init_data.get("type") == "token" and init_data.get("content"): diff --git a/router/circuit_breaker.py b/router/circuit_breaker.py index b40a1969..df17fcd2 100644 --- a/router/circuit_breaker.py +++ b/router/circuit_breaker.py @@ -34,6 +34,7 @@ class PerModelBreaker: """Tracks quota exhaustion for a specific model family with exponential cooldown.""" def __init__(self, name: str): + """Initialize the per-model circuit breaker with a name and default tier/cooldown states.""" self.name = name self.tier: int = 0 # 0 = open (allowed), 1-3 = cooldown active self.cooldown_until: float = 0.0 @@ -84,8 +85,6 @@ def record_failure(self): if self.tier == 0: new_tier = 1 - elif self.probe_granted: - new_tier = min(self.tier + 1, MAX_TIER) else: new_tier = min(self.tier + 1, MAX_TIER) @@ -130,6 +129,7 @@ class DualCircuitBreaker: """ def __init__(self): + """Initialize the dual circuit breaker with separate google and vendor sub-breakers.""" self.google = PerModelBreaker("google") self.vendor = PerModelBreaker("vendor") @@ -137,6 +137,7 @@ def __init__(self): # Default to checking BOTH — if either allows, return allowed. # This ensures old code without model awareness works correctly. def is_allowed(self) -> bool: + """Check if either the google or vendor breaker allows the request (backward-compat).""" return self.google.is_allowed() or self.vendor.is_allowed() def record_failure(self): @@ -151,9 +152,11 @@ def record_success(self): @property def tier(self) -> int: + """Return the maximum cooldown tier across both google and vendor sub-breakers.""" return max(self.google.tier, self.vendor.tier) def status(self) -> dict: + """Return the aggregated status dictionary of both sub-breakers for the dashboard.""" return { "google": self.google.status(), "vendor": self.vendor.status(), diff --git a/router/main.py b/router/main.py index 64efce64..9db0142c 100644 --- a/router/main.py +++ b/router/main.py @@ -5,12 +5,18 @@ import socket import asyncio import logging +import copy +import tempfile import yaml import httpx from contextlib import asynccontextmanager from fastapi import FastAPI, Request, HTTPException, Response from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse +from fastapi.staticfiles import StaticFiles +from pathlib import Path from circuit_breaker import get_breaker +from pydantic import BaseModel +from typing import Dict, Optional, Union # Configure logging — respect LOG_LEVEL env var (default: WARNING) _log_level_str = os.getenv("LOG_LEVEL", "WARNING").upper() @@ -126,6 +132,10 @@ async def push_aggregate_scores(): STATS_JSON_PATH = "/config/router_dir/router_stats.json" +# Module-level set to hold references to fire-and-forget background tasks, +# preventing premature garbage collection before the task completes (Ruff RUF006). +_background_tasks: set = set() + def load_persisted_stats(): """Loads persisted statistics from disk on startup to prevent resets on pod redeployment.""" global stats @@ -151,13 +161,60 @@ def load_persisted_stats(): except Exception as e: logger.error(f"Failed to load persisted stats: {e}") -def save_persisted_stats(): - """Persists current statistics in-memory structure to disk securely.""" +def _atomic_write_json_sync(path: str, data) -> None: + """Synchronously write JSON data to path using atomic temp-file + os.replace.""" + os.makedirs(os.path.dirname(path), exist_ok=True) + fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(path), suffix=".tmp") + try: + try: + f = os.fdopen(fd, "w", encoding="utf-8") + except Exception: + os.close(fd) + raise + + with f: + json.dump(data, f, indent=2) + os.replace(tmp_path, path) + except Exception: + try: + os.unlink(tmp_path) + except Exception: + pass + raise + + +async def _atomic_write_json_async(path: str, data) -> None: + """Asynchronously write JSON data to path via thread pool executor. + + Deep-copies the data to prevent concurrent modification from the main thread + while the executor thread is serializing it. + """ + loop = asyncio.get_running_loop() + data_copy = copy.deepcopy(data) + await loop.run_in_executor(None, _atomic_write_json_sync, path, data_copy) + + +_last_stats_save = 0.0 + +async def save_persisted_stats(force=False): + """Persists current statistics in-memory structure to disk securely (non-blocking). + + Offloads the synchronous file write to a thread pool executor so the + event loop is not blocked. The 2-second throttle is checked before + dispatching. + """ + global _last_stats_save + now = time.monotonic() + + # Throttle disk writes to max once per 2 seconds, unless forced + if not force and (now - _last_stats_save < 2.0): + return + + _last_stats_save = now # Set immediately to prevent concurrent writes during await try: - os.makedirs(os.path.dirname(STATS_JSON_PATH), exist_ok=True) - with open(STATS_JSON_PATH, "w") as f: - json.dump(stats, f, indent=2) + await _atomic_write_json_async(STATS_JSON_PATH, stats) except Exception as e: + _last_stats_save = 0.0 # Reset on failure to allow immediate retry logger.error(f"Failed to persist stats to disk: {e}") # Load initial stats from persistent storage @@ -234,6 +291,7 @@ async def sync_adaptive_router_roster(master_key: str): if max_score < 1.0: max_score = 55.0 # safety floor def norm(s: float) -> float: + """Helper to scale raw model index score against max score in roster to 0-100 range.""" return (s / max_score) * 100.0 for score, mid in free_models: # include all models — top 2 are also assigned to their correct tier n = norm(score) @@ -325,16 +383,34 @@ async def lifespan(app: FastAPI): await asyncio.sleep(1) else: logger.warning("⚠️ LiteLLM not ready within timeout — proceeding without roster sync") - yield - return - # Sync free-model roster into LiteLLM (separate try so sync failure doesn't loop) + + # Sync free-model roster into LiteLLM (non-fatal if it fails) + if litellm_master_key: + try: + await sync_adaptive_router_roster(litellm_master_key) + except Exception as e: + logger.error(f"Roster sync failed: {e}") + + # Start background task before yield so it runs during app lifetime + task = asyncio.create_task(push_aggregate_scores()) + try: - await sync_adaptive_router_roster(litellm_master_key) - except Exception as e: - logger.error(f"Roster sync failed: {e}") - yield - # Start aggregate score-push background task (runs for server lifetime) - asyncio.create_task(push_aggregate_scores()) + yield + finally: + # Cancel background score task + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + # Flush any buffered stats/timeline on clean shutdown (always runs) + await save_persisted_stats(force=True) + try: + timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") + await _atomic_write_json_async(timeline_path, stats["timeline"]) + except Exception as e: + logger.warning(f"Failed to persist timeline on shutdown: {e}") app = FastAPI(title="LLM Triage Router", lifespan=lifespan) @@ -375,7 +451,7 @@ async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_tra if time.time() - cached_time < CACHE_TTL_SECONDS: logger.info(f"⚡ Triage Cache Hit for prompt: '{normalized_prompt[:50]}...' -> routed to '{cached_decision}'") stats["cache_hits"] = stats.get("cache_hits", 0) + 1 - save_persisted_stats() + await save_persisted_stats() return cached_decision, 0.0, True, cached_decision # was_cache_hit=True start_time = time.time() @@ -388,7 +464,7 @@ async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_tra if time.time() - cached_time < CACHE_TTL_SECONDS: logger.info(f"⚡ Triage Cache Hit (post-queue) for prompt: '{normalized_prompt[:50]}...' -> routed to '{cached_decision}'") stats["cache_hits"] = stats.get("cache_hits", 0) + 1 - save_persisted_stats() + await save_persisted_stats() return cached_decision, 0.0, True, cached_decision try: @@ -478,6 +554,7 @@ async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_tra return "agent-advanced-core", latency, False, "advanced (exception)" def get_live_gemini_oauth_token() -> str | None: + """Retrieve the current valid Gemini OAuth access token from local storage if not expired.""" try: creds_path = "/config/gemini_auth/oauth_creds.json" if os.path.exists(creds_path): @@ -597,7 +674,12 @@ def detect_active_tool(body: dict) -> str: return "none" def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int, model: str, latency_ms: float, route: str = "litellm_fallback"): - """Accumulates token counts in memory for active tools and tracks request timelines.""" + """Accumulates token counts in memory for active tools and tracks request timelines. + + File writes are offloaded to a thread pool executor to avoid blocking the + event loop. The 2-second throttle is checked synchronously before + dispatching. + """ if tool_name == "none": tool_name = "other" @@ -625,13 +707,55 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int stats["timeline"].append(event) if len(stats["timeline"]) > 15: stats["timeline"].pop(0) - save_persisted_stats() + + # Fire-and-forget stats write via save_persisted_stats (non-blocking). + # Store the task reference in _background_tasks to prevent GC before completion (RUF006). + now = time.monotonic() try: - timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") - with open(timeline_path, "w") as f: - json.dump(stats["timeline"], f) - except Exception: - pass # disk persistence failure is non-fatal + loop = asyncio.get_running_loop() + _task = loop.create_task(save_persisted_stats()) + _background_tasks.add(_task) + _task.add_done_callback(_background_tasks.discard) + except RuntimeError: + # No running event loop (e.g. during early startup) — fall back to sync write + try: + global _last_stats_save + if now - _last_stats_save >= 2.0: + _atomic_write_json_sync(STATS_JSON_PATH, stats) + _last_stats_save = now + except Exception as e: + logger.error(f"Failed to persist stats to disk: {e}") + + # Throttle timeline file writes independently of the stats file (max once per 2 s) + timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") + if now - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: + try: + loop = asyncio.get_running_loop() + fut = loop.run_in_executor( + None, + _atomic_write_json_sync, + timeline_path, + copy.deepcopy(list(stats["timeline"])) + ) + record_tool_usage._last_save = now + + def done_callback(f): + """Log any uncaught exceptions returned from the background timeline executor thread.""" + try: + f.result() + except Exception as e: + logger.warning(f"Failed to persist timeline in background: {e}") + + fut.add_done_callback(done_callback) + except RuntimeError: + # No running event loop — fall back to sync write + try: + _atomic_write_json_sync(timeline_path, stats["timeline"]) + record_tool_usage._last_save = now + except Exception as e: + logger.warning(f"Failed to persist timeline: {e}") + except Exception as e: + logger.warning(f"Failed to persist timeline: {e}") def get_goose_sessions() -> list: """Queries the live mounted SQLite goose database to fetch the latest agentic sessions.""" @@ -717,6 +841,7 @@ async def get_llamacpp_metrics() -> dict: _AA_SCORES_LOADED = False def _load_aa_scores(): + """Load the Artificial Analysis agentic scores cache from local config.""" global _AA_SCORES_CACHE, _AA_SCORES_LOADED if _AA_SCORES_LOADED: return @@ -945,6 +1070,7 @@ async def proxy_models(): @app.post("/v1/chat/completions") async def chat_completions(request: Request): + """Handle incoming OpenAI-compatible chat completions requests and route them dynamically based on triage logic.""" global stats start_time = time.time() @@ -1037,7 +1163,7 @@ async def chat_completions(request: Request): stats["reasoning_requests"] = stats.get("reasoning_requests", 0) + 1 elif target_model == "agent-advanced-core": stats["advanced_requests"] = stats.get("advanced_requests", 0) + 1 - save_persisted_stats() + await save_persisted_stats() # Update the parent Langfuse observation with classification results if parent_obs: @@ -1119,74 +1245,152 @@ async def chat_completions(request: Request): except Exception: pass + is_stream_requested = body.get("stream", False) agy_response = await try_agy_proxy( prompt=last_prompt, messages=messages, session_id=session_id, total_timeout=300.0, + stream=is_stream_requested, target_tier=target_model ) if agy_response: - latency_ms = (time.time() - start_time) * 1000.0 model_name = agy_response.get("model", "gemini-3.5-flash (via agy)") - usage = agy_response.get("usage", {}) - prompt_tokens = usage.get("prompt_tokens", 0) - completion_tokens = usage.get("completion_tokens", 0) - record_tool_usage( - active_tool, prompt_tokens, completion_tokens, - model_name, latency_ms, route="google_oauth_direct" - ) - logger.info(f"✅ agy proxy succeeded: {model_name}, {latency_ms:.0f}ms") - - # Finalize agy span - if agy_span_obj: - try: - agy_span_obj.end( - output={"model": model_name, "tokens": completion_tokens}, - metadata={"latency_ms": latency_ms, "tier": target_model}, - ) - except Exception: - pass - - if body.get("stream", False): - content = agy_response.get("choices", [{}])[0].get("message", {}).get("content", "") - async def agy_stream_generator(): + + if "stream" in agy_response: + # Real native stream generator + async def native_agy_stream_generator(stream_gen, model_name): + """Asynchronous generator yielding native OpenAI-compatible streaming chunks from the real agy daemon.""" import uuid created_time = int(time.time()) chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}" - chunk_size = 40 - for i in range(0, len(content), chunk_size): - chunk_text = content[i:i+chunk_size] - chunk_data = { + token_count = 0 + try: + async for token in stream_gen: + if not token: + continue + token_count += 1 + chunk_data = { + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created_time, + "model": model_name, + "choices": [{ + "index": 0, + "delta": {"content": token}, + "finish_reason": None + }] + } + yield f"data: {json.dumps(chunk_data)}\n\n".encode("utf-8") + + # End of stream chunk + finish_data = { "id": chunk_id, "object": "chat.completion.chunk", "created": created_time, "model": model_name, "choices": [{ "index": 0, - "delta": {"content": chunk_text}, - "finish_reason": None + "delta": {}, + "finish_reason": "stop" }] } - yield f"data: {json.dumps(chunk_data)}\n\n".encode("utf-8") - await asyncio.sleep(0.005) - - finish_data = { - "id": chunk_id, - "object": "chat.completion.chunk", - "created": created_time, - "model": model_name, - "choices": [{ - "index": 0, - "delta": {}, - "finish_reason": "stop" - }] - } - yield f"data: {json.dumps(finish_data)}\n\n".encode("utf-8") - yield b"data: [DONE]\n\n" - return StreamingResponse(agy_stream_generator(), media_type="text/event-stream") + yield f"data: {json.dumps(finish_data)}\n\n".encode("utf-8") + yield b"data: [DONE]\n\n" + + # Success telemetry + latency_ms = (time.time() - start_time) * 1000.0 + # Approximate prompt tokens based on messages characters + prompt_chars = sum(len(m.get("content", "")) for m in messages) + approx_prompt_tokens = max(1, prompt_chars // 4) + + record_tool_usage( + active_tool, approx_prompt_tokens, token_count, + model_name, latency_ms, route="google_oauth_direct" + ) + logger.info(f"✅ native agy stream succeeded: {model_name}, {latency_ms:.0f}ms") + if agy_span_obj: + try: + agy_span_obj.end( + output={"model": model_name, "tokens": token_count}, + metadata={"latency_ms": latency_ms, "tier": target_model}, + ) + except Exception: + pass + except Exception as stream_err: + logger.error(f"Error during native agy stream generation: {stream_err}") + if agy_span_obj: + try: + agy_span_obj.end( + output={"error": str(stream_err)[:200]}, + metadata={"status": "failed"}, + ) + except Exception: + pass + raise + return StreamingResponse(native_agy_stream_generator(agy_response["stream"], model_name), media_type="text/event-stream") else: - return agy_response + latency_ms = (time.time() - start_time) * 1000.0 + usage = agy_response.get("usage", {}) + prompt_tokens = usage.get("prompt_tokens", 0) + completion_tokens = usage.get("completion_tokens", 0) + record_tool_usage( + active_tool, prompt_tokens, completion_tokens, + model_name, latency_ms, route="google_oauth_direct" + ) + logger.info(f"✅ agy proxy succeeded: {model_name}, {latency_ms:.0f}ms") + + # Finalize agy span + if agy_span_obj: + try: + agy_span_obj.end( + output={"model": model_name, "tokens": completion_tokens}, + metadata={"latency_ms": latency_ms, "tier": target_model}, + ) + except Exception: + pass + + if is_stream_requested: + # Robust fallback: simulate stream if we requested stream but got buffered response + content = agy_response.get("choices", [{}])[0].get("message", {}).get("content", "") + async def agy_stream_generator(): + """Asynchronous generator yielding simulated OpenAI-compatible streaming chunks from a static agy response.""" + import uuid + created_time = int(time.time()) + chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}" + chunk_size = 40 + for i in range(0, len(content), chunk_size): + chunk_text = content[i:i+chunk_size] + chunk_data = { + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created_time, + "model": model_name, + "choices": [{ + "index": 0, + "delta": {"content": chunk_text}, + "finish_reason": None + }] + } + yield f"data: {json.dumps(chunk_data)}\n\n".encode("utf-8") + await asyncio.sleep(0.005) + + finish_data = { + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created_time, + "model": model_name, + "choices": [{ + "index": 0, + "delta": {}, + "finish_reason": "stop" + }] + } + yield f"data: {json.dumps(finish_data)}\n\n".encode("utf-8") + yield b"data: [DONE]\n\n" + return StreamingResponse(agy_stream_generator(), media_type="text/event-stream") + else: + return agy_response except ImportError: if agy_span_obj: try: @@ -1308,6 +1512,7 @@ async def agy_stream_generator(): r = await client.send(req, stream=True) if r.status_code == 200: async def stream_generator(): + """Asynchronous generator that yields streaming chunks from LiteLLM completions response and logs usage stats on completion.""" completion_chars = 0 request_tokens = len(json.dumps(body_to_send)) // 4 try: @@ -1445,6 +1650,7 @@ async def metrics(): @app.get("/dashboard", response_class=HTMLResponse) async def get_dashboard(): + """Render the router main dashboard HTML showing system metrics, health checks, and recent token usage.""" # 1. Run live health checks valkey_status = await check_tcp_port("127.0.0.1", 6379) litellm_status = await check_http_endpoint("http://127.0.0.1:4000/") @@ -1700,6 +1906,7 @@ async def get_dashboard(): # Source badge helper: generates a colored inline source tag def src_badge(label, color): + """Generate inline HTML span styled as a colored status/category badge.""" return f"{label}" # 10. Pre-compute llama.cpp HTML cards @@ -2160,6 +2367,9 @@ def src_badge(label, color):
Antigravity Gateway
System Control Center
+
+ 📊 Dataset Visualizer +
{oauth_banner_html} @@ -2415,6 +2625,80 @@ def src_badge(label, color): """ return html_content +# --- Static files (visualizer, data files) --- +STATIC_DIR = Path(__file__).resolve().parent / "static" +DATA_DIR = Path(__file__).resolve().parent / "data" +DATA_DIR.mkdir(exist_ok=True) +app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") +app.mount("/data", StaticFiles(directory=str(DATA_DIR)), name="data") + +@app.get("/visualizer", response_class=HTMLResponse) +async def get_visualizer(): + """Serve the dataset visualizer for human review.""" + vis_path = STATIC_DIR / "visualizer.html" + if vis_path.exists(): + return HTMLResponse(vis_path.read_text()) + return HTMLResponse("

Visualizer not found

", status_code=404) + +class AnnotationItem(BaseModel): + """Pydantic model representing a single human dataset review annotation.""" + tier: Union[int, str, None] = None + note: str = "" + ts: Optional[str] = None + +VALID_TIERS = {"agent-simple-core", "agent-medium-core", "agent-complex-core", "agent-reasoning-core", "agent-advanced-core"} + +@app.post("/dashboard/save-annotations") +async def save_annotations(payload: Dict[str, AnnotationItem]): + """Save human review annotations to disk.""" + if len(payload) > 1000: + raise HTTPException( + status_code=400, + detail="Payload size limit exceeded: maximum of 1000 annotations allowed per request." + ) + + for k, item in payload.items(): + if not k.isdigit(): + raise HTTPException( + status_code=400, + detail=f"Invalid payload key '{k}': keys must be numeric strings (dataset indexes)." + ) + + t = item.tier + if t is not None: + if isinstance(t, int): + if t < 0 or t > 4: + raise HTTPException( + status_code=400, + detail=f"Invalid tier index {t} for index {k}: must be between 0 and 4." + ) + elif isinstance(t, str): + if t not in VALID_TIERS and t != "?": + raise HTTPException( + status_code=400, + detail=f"Invalid tier string '{t}' for index {k}." + ) + else: + raise HTTPException( + status_code=400, + detail=f"Invalid tier type for index {k}: must be int, str, or null." + ) + + if len(item.note) > 1000: + raise HTTPException( + status_code=400, + detail=f"Note length limit exceeded at index {k}: maximum of 1000 characters allowed." + ) + + try: + body = {k: (v.model_dump() if hasattr(v, "model_dump") else v.dict()) for k, v in payload.items()} + ann_path = DATA_DIR / "annotations.json" + await _atomic_write_json_async(str(ann_path), body) + return JSONResponse({"status": "ok", "saved": len(body)}) + except Exception as e: + logger.error(f"Failed to save annotations: {e}") + raise HTTPException(status_code=500, detail="Failed to save annotations") + if __name__ == "__main__": import uvicorn logger.info(f"Starting LLM Triage Router on {host}:{port}...") diff --git a/router/static/visualizer.html b/router/static/visualizer.html new file mode 100644 index 00000000..3a837ae8 --- /dev/null +++ b/router/static/visualizer.html @@ -0,0 +1,389 @@ + + + + + +Classifier Dataset Visualizer + + + +
+

📊 Dataset Visualizer

+
+
Total: 0
+
Agree: 0
+
Conflict: 0
+
Reviewed: 0
+ +
+
+
+ + + + + Showing 0 + +
+
+
+
+
+

Select a prompt from the list

+

LLM evaluation vs classifier prediction side by side

+
+
+
+ + + + diff --git a/scripts/benchmark_classifier.py b/scripts/benchmark_classifier.py new file mode 100644 index 00000000..bbeace10 --- /dev/null +++ b/scripts/benchmark_classifier.py @@ -0,0 +1,140 @@ +"""Benchmark gemma4-26a4b-routing classifier against labeled dataset.""" +import json, urllib.request, time, sys +from collections import defaultdict, Counter +from pathlib import Path + +# Load dataset +dataset_path = Path(__file__).resolve().parent.parent / "data" / "classified_dataset.json" +with open(dataset_path) as f: + dataset = json.load(f) + +# Classifier prompt (same as router/config.yaml) +PROMPT_TEMPLATE = """Classify the coding task complexity. Output ONLY the tier name. + +agent-simple-core: trivial one-liners, syntax fixes, single-line edits +agent-medium-core: single-function changes, light refactoring, simple tests +agent-complex-core: multi-file changes, algorithmic work, data pipelines +agent-reasoning-core: deep analysis, architecture decisions, debugging complex systems +agent-advanced-core: system-level architecture, cross-cutting concerns, novel design + +Task: """ + +TIERS = [ + "agent-simple-core", "agent-medium-core", "agent-complex-core", + "agent-reasoning-core", "agent-advanced-core" +] + +def classify(prompt): + """Call gemma4-26a4b-routing via llama-server.""" + payload = { + "model": "gemma4-26a4b-routing", + "messages": [{"role": "user", "content": PROMPT_TEMPLATE + prompt}], + "temperature": 0.0, + "max_tokens": 15, + } + req = urllib.request.Request( + "http://127.0.0.1:8080/v1/chat/completions", + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json", "Authorization": "Bearer local-token"} + ) + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read()) + choices = data.get("choices", []) + if not choices: + return "ERROR" + return choices[0].get("message", {}).get("content", "").strip() + +total = len(dataset.get("prompts", [])) +print(f"Benchmark: gemma4-26a4b-routing vs {total} labeled prompts\n") + +# Run classification +results = [] +correct = 0 +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", [])): + 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", "") + + try: + predicted = 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 + + per_tier[expected]["total"] += 1 + if predicted == expected: + correct += 1 + per_tier[expected]["correct"] += 1 + + confusion[expected][predicted] += 1 + + # Progress + if (i + 1) % 20 == 0: + acc = correct / (i + 1) * 100 + print(f" {i+1}/{total} — accuracy {acc:.1f}%") + + time.sleep(0.05) # minimal rate-limit (model handles concurrency via llama-server slots) + +# Report +overall = (correct / total * 100) if total > 0 else 0.0 + +print(f"\n{'='*60}") +print(f"Overall accuracy: {correct}/{total} ({overall:.1f}%)") +print(f"{'='*60}") + +print(f"\nPer-tier accuracy:") +for tier in TIERS: + t = per_tier[tier] + pct = t["correct"] / t["total"] * 100 if t["total"] > 0 else 0 + bar = "█" * int(pct / 5) + "░" * (20 - int(pct / 5)) + print(f" {tier:30s} {t['correct']:3d}/{t['total']:3d} {bar} {pct:.1f}%") + +print(f"\nConfusion matrix (expected → predicted):") +header = " " * 30 + "".join(f"{t:25s}" for t in TIERS) +print(header) +for exp_tier in TIERS: + row = f"{exp_tier:30s}" + for pred_tier in TIERS: + count = confusion[exp_tier].get(pred_tier, 0) + count_str = f"{count:3d}" + if exp_tier == pred_tier: + cell = f" \033[32m{count_str}\033[0m" + row += f"{cell:34s}" + elif count > 0: + cell = f" \033[31m{count_str}\033[0m" + row += f"{cell:34s}" + else: + cell = f" {count_str}" + row += f"{cell:25s}" + print(row) + +# Save detailed results +out_path = Path(__file__).resolve().parent.parent / "data" / "benchmark_results.json" +with open(out_path, 'w') as f: + json.dump({ + "classifier": "gemma4-26a4b-routing", + "dataset_total": total, + "overall_accuracy": round(overall, 1), + "per_tier": {t: { + "correct": per_tier[t]["correct"], + "total": per_tier[t]["total"], + "accuracy": round(per_tier[t]["correct"] / per_tier[t]["total"] * 100, 1) if per_tier[t]["total"] > 0 else 0 + } for t in TIERS}, + "confusion": {t: dict(confusion[t]) for t in TIERS}, + "details": results, + }, f, indent=2, ensure_ascii=False) + +print(f"\nDetailed results saved to {out_path}") diff --git a/scripts/classify_direct.py b/scripts/classify_direct.py new file mode 100644 index 00000000..ddf8a99a --- /dev/null +++ b/scripts/classify_direct.py @@ -0,0 +1,108 @@ +"""Direct classification of Hermes prompts using gemma4-26a4b-routing.""" +import json, urllib.request, time +from pathlib import Path + +PROMPT_TEMPLATE = """Classify the coding task complexity. Output ONLY the tier name. + +agent-simple-core: trivial one-liners, syntax fixes, single-line edits +agent-medium-core: single-function changes, light refactoring, simple tests +agent-complex-core: multi-file changes, algorithmic work, data pipelines +agent-reasoning-core: deep analysis, architecture decisions, debugging complex systems +agent-advanced-core: system-level architecture, cross-cutting concerns, novel design + +Task: """ +LLAMA_SERVER_URL = "http://127.0.0.1:8080/v1/chat/completions" +TIERS = { + "agent-simple-core", "agent-medium-core", "agent-complex-core", + "agent-reasoning-core", "agent-advanced-core" +} + +def classify(prompt): + """Query the llama-server to classify the prompt task complexity.""" + payload = { + "model": "gemma4-26a4b-routing", + "messages": [{"role": "user", "content": PROMPT_TEMPLATE + prompt}], + "temperature": 0.0, + "max_tokens": 15, + } + req = urllib.request.Request( + LLAMA_SERVER_URL, + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json", "Authorization": "Bearer local-token"} + ) + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read()) + return data["choices"][0]["message"].get("content", "").strip() + +# Load prompts +data_dir = Path(__file__).resolve().parent.parent / "data" +with open(data_dir / "raw_prompts_hermes.json") as f: + prompts = json.load(f) + +print(f"Classifying {len(prompts)} prompts with gemma4-26a4b-routing...") + +results = [] +errors = 0 +for i, p in enumerate(prompts): + prompt = p['prompt'] + # Truncate very long prompts to classifier context window (~3500 chars safe margin) + if len(prompt) > 3500: + prompt = prompt[:3500] + + try: + raw_tier = classify(prompt) + if raw_tier in TIERS: + tier = raw_tier + extra = {} + else: + tier = "ERROR" + extra = {"raw_output": raw_tier} + errors += 1 + results.append({"id": i, "tier": tier, "prompt_snippet": prompt[:60], **extra}) + except Exception as e: + tier = f"ERROR" + errors += 1 + results.append({"id": i, "tier": tier, "error": str(e)[:100]}) + print(f" [{i}] ERROR: {str(e)[:80]}") + + if (i + 1) % 30 == 0: + print(f" {i+1}/{len(prompts)} — {errors} errors") + time.sleep(0.05) + +# Count tiers +from collections import Counter +counts = Counter(r["tier"] for r in results) + +print(f"\nDone. {len(results)} classified, {errors} errors") +print(f"Counts:") +for tier in ['agent-simple-core', 'agent-medium-core', 'agent-complex-core', 'agent-reasoning-core', 'agent-advanced-core']: + c = counts.get(tier, 0) + print(f" {tier}: {c}") +error_count = sum(1 for r in results if r["tier"] == "ERROR") +if error_count: + print(f" ERROR: {error_count}") + +# Build dataset +dataset_prompts = [] +for p, r in zip(prompts, results): + dataset_prompts.append({ + "prompt": p['prompt'], + "tier": r['tier'], + "classifier": "uuid", + "session_id": p.get('session_id', ''), + }) + +dataset = { + "prompts": dataset_prompts, + "counts": dict(counts), + "total": len(dataset_prompts), + "gaps": [t for t in ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] + if counts.get(t, 0) < 20] +} + +out_path = data_dir / "classified_dataset.json" +with open(out_path, 'w') as f: + json.dump(dataset, f, indent=2, ensure_ascii=False) + +print(f"\nSaved to {out_path}") +print(f"Gaps: {dataset['gaps']}") \ No newline at end of file diff --git a/scripts/extract_complex.py b/scripts/extract_complex.py new file mode 100644 index 00000000..75af8375 --- /dev/null +++ b/scripts/extract_complex.py @@ -0,0 +1,138 @@ +"""Final gap-fill: deep extraction targeting complex + advanced tiers only.""" +import os, base64, json, urllib.request, time +from pathlib import Path + +env = {} +env_path = Path(__file__).resolve().parent.parent / ".env" +with open(env_path) as f: + for line in f: + line = line.strip() + if line and not line.startswith('#') and '=' in line: + k, v = line.split('=', 1) + env[k.strip()] = v.strip().strip('"').strip("'") + +pk = env['LANGFUSE_PUBLIC_KEY'] +sk = env['LANGFUSE_SECRET_KEY'] +auth = base64.b64encode(f"{pk}:{sk}".encode()).decode() +base_url = "http://localhost:3001" + +existing = set() +dataset_path = Path(__file__).resolve().parent.parent / "data" / "classified_dataset.json" +if dataset_path.exists(): + try: + with open(dataset_path) as f: + existing_data = json.load(f) + for p in existing_data.get('prompts', []): + existing.add(p['prompt'].strip().lower()) + except Exception as e: + print(f"Warning: Failed to load existing dataset: {e}") + +print(f"Already classified: {len(existing)} prompts") + +def fetch_observations(page=1, limit=50): + """Fetch a paginated list of observations from the Langfuse public API.""" + url = f"{base_url}/api/public/observations?limit={limit}&page={page}&orderBy=timestamp.desc&level=DEFAULT&name=litellm-acompletion" + req = urllib.request.Request(url) + req.add_header("Authorization", f"Basic {auth}") + with urllib.request.urlopen(req, timeout=15) as resp: + return json.loads(resp.read()) + +def extract_user_prompt(obs): + """Extract and parse the FIRST real user prompt from the observation input payload. + + Uses forward iteration (not reversed) to return the first user message, + matching the semantics of extract_prompts.py. Skips pseudo-user system notes. + """ + inp = obs.get('input') + if not inp: return None + if isinstance(inp, str): + try: inp = json.loads(inp) + except: return None + if not isinstance(inp, dict): return None + messages = inp.get('messages', []) + if not messages: return None + for msg in messages: # forward iteration: first user message + if isinstance(msg, dict) and msg.get('role') == 'user': + content = msg.get('content', '') + if not isinstance(content, str) or len(content.strip()) <= 3: + continue + stripped = content.strip() + # Skip Hermes system notes injected as user messages + if stripped.startswith('[System:') or stripped.startswith('[Note:'): + continue + return stripped + return None + +# Keywords suggesting complex/advanced work +COMPLEX_KEYWORDS = [ + 'refactor', 'architect', 'design', 'implement', 'migrate', 'debug', + 'diagnose', 'review', 'analyze', 'optimize', 'restructure', 'pipeline', + 'system', 'module', 'framework', 'pattern', 'strategy', 'algorithm', + 'multi-tenant', 'distributed', 'sharding', 'consensus', 'isolation', + 'security', 'vulnerability', 'deadlock', 'concurrent', 'scale', + 'infrastructure', 'deploy', 'orchestrate', 'integrate', 'protocol', +] + +def looks_complex(prompt): + """Heuristic: longer prompts with complex keywords.""" + lower = prompt.lower() + if len(prompt) < 200: + return False + score = sum(1 for kw in COMPLEX_KEYWORDS if kw in lower) + return score >= 2 + +print("Deep extraction: targeting complex/advanced prompts...") +prompts = [] +seen = set() +page = 1 +target = 50 +max_pages = 200 # go deep into history + +while len(prompts) < target and page <= max_pages: + try: + data = fetch_observations(page=page, limit=50) + except Exception as e: + print(f" Page {page} failed: {e}") + break + + obs_list = data.get('data', []) + if not obs_list: + print(f" Page {page}: empty, stopping") + break + + added = 0 + for obs in obs_list: + if len(prompts) >= target: + break + prompt = extract_user_prompt(obs) + if not prompt: continue + norm = prompt.strip().lower() + if norm in seen: continue + if norm in existing: continue + if not looks_complex(prompt): continue + + seen.add(norm) + prompts.append({ + "prompt": prompt, + "observation_id": obs.get('id', ''), + "trace_id": obs.get('traceId', ''), + "timestamp": obs.get('startTime', ''), + }) + added += 1 + + print(f" Page {page}: +{added} new → {len(prompts)} total") + page += 1 + time.sleep(0.1) + +out_path = Path(__file__).resolve().parent.parent / "data" / "raw_prompts_complex.json" +with open(out_path, 'w') as f: + json.dump(prompts, f, indent=2, ensure_ascii=False) + +print(f"\nSaved {len(prompts)} complex prompts to {out_path}") +lengths = [len(p['prompt']) for p in prompts] +if lengths: + print(f"Length range: {min(lengths)}-{max(lengths)} chars, avg: {sum(lengths)/len(lengths):.0f}") + for p in prompts[:5]: + print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])} chars) {p['prompt'][:120]}...") +else: + print("No prompts collected.") diff --git a/scripts/extract_gapfill.py b/scripts/extract_gapfill.py new file mode 100644 index 00000000..cdb5f525 --- /dev/null +++ b/scripts/extract_gapfill.py @@ -0,0 +1,150 @@ +"""Gap-fill extraction: pull longer/older prompts targeting complex+ tiers.""" +import os, base64, json, urllib.request, time, re +from pathlib import Path + +env = {} +env_path = Path(__file__).resolve().parent.parent / ".env" +with open(env_path) as f: + for line in f: + line = line.strip() + if line and not line.startswith('#') and '=' in line: + k, v = line.split('=', 1) + env[k.strip()] = v.strip().strip('"').strip("'") + +pk = env['LANGFUSE_PUBLIC_KEY'] +sk = env['LANGFUSE_SECRET_KEY'] +auth = base64.b64encode(f"{pk}:{sk}".encode()).decode() +base_url = "http://localhost:3001" + +# Load already-classified prompts to skip +existing = set() +dataset_path = Path(__file__).resolve().parent.parent / "data" / "classified_dataset.json" +if dataset_path.exists(): + with open(dataset_path) as f: + existing_data = json.load(f) + for p in existing_data.get('prompts', []): + existing.add(p['prompt'].strip().lower()) + +print(f"Already classified: {len(existing)} prompts") + +def fetch_observations(page=1, limit=50): + """Fetch DEFAULT-level litellm-acompletion observations.""" + url = f"{base_url}/api/public/observations?limit={limit}&page={page}&orderBy=timestamp.desc&level=DEFAULT&name=litellm-acompletion" + req = urllib.request.Request(url) + req.add_header("Authorization", f"Basic {auth}") + with urllib.request.urlopen(req, timeout=15) as resp: + return json.loads(resp.read()) + +def extract_user_prompt(obs): + """Extract and parse the FIRST real user prompt from the observation input payload. + + Uses forward iteration (not reversed) to return the first user message, + matching the semantics of extract_prompts.py. Skips pseudo-user system notes. + """ + inp = obs.get('input') + if not inp: + return None + if isinstance(inp, str): + try: inp = json.loads(inp) + except: return None + if not isinstance(inp, dict): + return None + messages = inp.get('messages', []) + if not messages: + return None + for msg in messages: # forward iteration: first user message + if isinstance(msg, dict) and msg.get('role') == 'user': + content = msg.get('content', '') + if not isinstance(content, str) or len(content.strip()) <= 3: + continue + stripped = content.strip() + # Skip Hermes system notes injected as user messages + if stripped.startswith('[System:') or stripped.startswith('[Note:'): + continue + return stripped + return None + +def is_trivial(prompt): + """Check if the prompt matches a list of trivial test patterns to filter out.""" + lower = prompt.strip().lower() + if len(lower) < 20: + return True + trivial = ["say hello", "hi", "test", "ping", "hello", "hey", "what's up", + "how are you", "good morning", "what model are you", "who are you", + "tell me a joke", "what is 2+2", "what is the capital"] + for pat in trivial: + if len(lower) < 50: + escaped = re.escape(pat) + if re.search(r'\b' + escaped + r'\b', lower): + return True + return False + +print("Extracting gap-fill prompts (longer, older, targeting complex+)...") +prompts = [] +seen = set() +page = 1 +target = 80 # generous — workers will filter further +max_pages = 100 + +while len(prompts) < target and page <= max_pages: + try: + data = fetch_observations(page=page, limit=50) + except Exception as e: + print(f" Page {page} failed: {e}") + break + + obs_list = data.get('data', []) + if not obs_list: + print(f" Page {page}: empty, stopping") + break + + added = 0 + for obs in obs_list: + if len(prompts) >= target: + break + + prompt = extract_user_prompt(obs) + if not prompt: + continue + if is_trivial(prompt): + continue + + norm = prompt.strip().lower() + if norm in seen: + continue + if norm in existing: + continue + + # Bias toward complex: prefer longer prompts (>200 chars) + # but don't exclude shorter ones entirely + if len(prompt) < 100 and len(prompts) > 40: + continue # after 40 collected, only take substantial prompts + + seen.add(norm) + prompts.append({ + "prompt": prompt, + "observation_id": obs.get('id', ''), + "trace_id": obs.get('traceId', ''), + "timestamp": obs.get('startTime', ''), + "model": obs.get('model', ''), + }) + added += 1 + + print(f" Page {page}: +{added} new → {len(prompts)} total") + page += 1 + time.sleep(0.1) + +out_dir = Path(__file__).resolve().parent.parent / "data" +out_path = out_dir / "raw_prompts_gapfill.json" +with open(out_path, 'w') as f: + json.dump(prompts, f, indent=2, ensure_ascii=False) + +print(f"\nSaved {len(prompts)} gap-fill prompts to {out_path}") +lengths = [len(p['prompt']) for p in prompts] +if lengths: + print(f"Length range: {min(lengths)}-{max(lengths)} chars, avg: {sum(lengths)/len(lengths):.0f}") + print(f"Sample:") + for p in prompts[:5]: + print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])} chars) {p['prompt'][:100]}...") +else: + print("No prompts collected.") diff --git a/scripts/extract_prompts.py b/scripts/extract_prompts.py index 8c3e454d..028baa8d 100644 --- a/scripts/extract_prompts.py +++ b/scripts/extract_prompts.py @@ -1,5 +1,12 @@ -"""Extract 300 meaningful coding prompts from Langfuse observations.""" -import os, base64, json, urllib.request, urllib.error, sys, time +"""Re-extract prompts using FIRST user message (not last) — fixes truncation.""" +import os +import base64 +import json +import urllib.request +import urllib.error +import sys +import time +import re from pathlib import Path env = {} @@ -25,13 +32,15 @@ ] def is_trivial(prompt): - """Filter out test pings and trivial prompts.""" + """Filter out test pings and trivial prompts using word boundaries to avoid partial matches.""" lower = prompt.strip().lower() if len(lower) < 20: return True for pat in TRIVIAL_PATTERNS: - if pat in lower and len(lower) < 50: - return True + if len(lower) < 50: + escaped = re.escape(pat) + if re.search(r'\b' + escaped + r'\b', lower): + return True return False def fetch_observations(page=1, limit=50): @@ -42,34 +51,46 @@ def fetch_observations(page=1, limit=50): with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read()) -def extract_user_prompt(obs): - """Extract the last user message from an observation's input.""" +def extract_first_user_prompt(obs): + """Extract the FIRST real user message (skip system notes).""" inp = obs.get('input') if not inp: return None if isinstance(inp, str): try: inp = json.loads(inp) - except: + except Exception: return None if not isinstance(inp, dict): return None messages = inp.get('messages', []) if not messages: return None - for msg in reversed(messages): - if isinstance(msg, dict) and msg.get('role') == 'user': - content = msg.get('content', '') - if isinstance(content, str) and len(content.strip()) > 3: - return content.strip() + + for msg in messages: + if not isinstance(msg, dict): + continue + if msg.get('role') != 'user': + continue + content = msg.get('content', '') + if not isinstance(content, str) or len(content.strip()) <= 3: + continue + # Skip Hermes system notes injected as user messages + stripped = content.strip() + if stripped.startswith('[System:') or stripped.startswith('[Note:'): + continue + if stripped.startswith('[IMPORTANT:'): + # Skill invocations — keep these, they're real prompts + pass + return stripped return None -print("Extracting meaningful coding prompts from Langfuse observations...") +print("Re-extracting prompts using FIRST user message...") prompts = [] seen = set() page = 1 target = 300 -max_pages = 50 # 50 pages × 50 = 2500 observations +max_pages = 100 while len(prompts) < target and page <= max_pages: try: @@ -79,26 +100,21 @@ def extract_user_prompt(obs): break obs_list = data.get('data', []) - total_available = data.get('meta', {}).get('totalItems', 0) - if not obs_list: print(f" Page {page}: empty, stopping") break - added_this_page = 0 + added = 0 for obs in obs_list: if len(prompts) >= target: break - prompt = extract_user_prompt(obs) + prompt = extract_first_user_prompt(obs) if not prompt: continue - - # Skip trivial test pings if is_trivial(prompt): continue - # Deduplicate norm = prompt.strip().lower() if norm in seen: continue @@ -111,26 +127,23 @@ def extract_user_prompt(obs): "timestamp": obs.get('startTime', ''), "model": obs.get('model', ''), }) - added_this_page += 1 + added += 1 - print(f" Page {page}: {len(obs_list)} obs, +{added_this_page} new → {len(prompts)} total (of {total_available} available)") + print(f" Page {page}: +{added} new → {len(prompts)} total") page += 1 - time.sleep(0.1) # gentle rate limit + time.sleep(0.1) -# Save out_dir = Path(__file__).resolve().parent.parent / "data" -out_dir.mkdir(exist_ok=True) -out_path = out_dir / "raw_prompts.json" - +out_path = out_dir / "raw_prompts_v2.json" with open(out_path, 'w') as f: json.dump(prompts, f, indent=2, ensure_ascii=False) print(f"\nSaved {len(prompts)} prompts to {out_path}") -# Stats lengths = [len(p['prompt']) for p in prompts] -print(f"Length range: {min(lengths)}-{max(lengths)} chars") -print(f"Avg length: {sum(lengths)/len(lengths):.0f} chars") -print(f"\nSample (first 10):") -for p in prompts[:10]: - print(f" [{p['timestamp'][:19]}] {p['prompt'][:120]}...") +if lengths: + print(f"Length: min={min(lengths)}, max={max(lengths)}, median={sorted(lengths)[len(lengths)//2]}, avg={sum(lengths)/len(lengths):.0f}") + print(f"Short (<100 chars): {sum(1 for l in lengths if l < 100)}") + print(f"\nSample (first 10):") + for p in prompts[:10]: + print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])}c) {p['prompt'][:120]}...") diff --git a/scripts/reclassify_all.py b/scripts/reclassify_all.py new file mode 100644 index 00000000..91ea6b61 --- /dev/null +++ b/scripts/reclassify_all.py @@ -0,0 +1,111 @@ +"""Re-run gemma4 classifier (with grammar) on all dataset prompts via router.""" +import json, urllib.request, time, sys, os, tempfile +from pathlib import Path +from collections import Counter + +TIERS = ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] + +LLAMA_SERVER_URL = "http://127.0.0.1:8080/v1/chat/completions" +print(f"Using llama-server on {LLAMA_SERVER_URL}") + +PROMPT_TEMPLATE = """Analyze the request complexity. Respond with exactly one of: +- simple boilerplate: agent-simple-core +- moderate complexity: agent-medium-core +- deep algorithms: agent-complex-core +- heavy multi-step reasoning: agent-reasoning-core +- system-level / novel design: agent-advanced-core + +Request: """ + +def classify(prompt): + """Query the llama-server to classify the prompt complexity with grammar enforcement.""" + payload = { + 'model': 'gemma4-26a4b-routing', + 'messages': [{'role': 'user', 'content': PROMPT_TEMPLATE + prompt}], + 'max_tokens': 15, 'temperature': 0, + 'grammar': 'root ::= "agent-simple-core" | "agent-medium-core" | "agent-complex-core" | "agent-reasoning-core" | "agent-advanced-core"' + } + req = urllib.request.Request(LLAMA_SERVER_URL, data=json.dumps(payload).encode(), headers={'Content-Type':'application/json','Authorization':'Bearer local-token'}) + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read()) + choices = data.get('choices', []) + if not choices: + return f"ERROR: empty response" + return choices[0].get('message', {}).get('content', '').strip() + +# Load existing dataset (kanban/llm evals) +data_dir = Path(__file__).resolve().parent.parent / 'data' +with open(data_dir / 'classified_dataset.json') as f: + dataset = json.load(f) + +# Load raw prompts for full text +with open(data_dir / 'raw_prompts_hermes.json') as f: + all_prompts = json.load(f) + +# Build prompt lookup +prompt_map = {} +for p in all_prompts: + prompt_map[p['prompt']] = p + +print(f"Classifying {len(dataset['prompts'])} prompts with gemma4-26a4b (grammar-enforced)...") + +results = [] +for i, item in enumerate(dataset['prompts']): + prompt = item['prompt'] + + # Original LLM/kanban eval + llm_tier = item.get('llm_tier') or item.get('tier', '?') + + # Classifier eval + try: + clf_tier = classify(prompt) + except Exception as e: + clf_tier = f"ERROR: {str(e)[:50]}" + + results.append({ + 'prompt': prompt, + 'llm_tier': llm_tier, + 'clf_tier': clf_tier, + 'session_id': item.get('session_id', ''), + }) + + if (i + 1) % 30 == 0: + agree = sum(1 for r in results if r['llm_tier'] == r['clf_tier']) + print(f" {i+1}/{len(dataset['prompts'])} — {agree}/{i+1} agree ({agree/(i+1)*100:.0f}%)") + sys.stdout.flush() + +# Stats +clf_counts = Counter(r['clf_tier'] for r in results) +llm_counts = Counter(r['llm_tier'] for r in results) +agree = sum(1 for r in results if r['llm_tier'] == r['clf_tier']) + +total_results = len(results) +print(f"\n{'='*60}") +if total_results > 0: + print(f"Agreement: {agree}/{total_results} ({agree/total_results*100:.1f}%)") +else: + print("Agreement: 0/0 (0.0%)") +print("\nTier distribution:") +print(f"{'Tier':30s} {'LLM':>6s} {'CLF':>6s} {'Δ':>6s}") +for t in TIERS: + lc = llm_counts.get(t, 0) + cc = clf_counts.get(t, 0) + print(f" {t:30s} {lc:>6d} {cc:>6d} {cc-lc:>+6d}") + +# Save combined dataset atomically +combined = { + 'total': total_results, + 'agreement': round(agree / total_results * 100, 1) if total_results > 0 else 0.0, + 'llm_counts': dict(llm_counts), + 'clf_counts': dict(clf_counts), + 'prompts': results, +} + +dest_path = data_dir / 'classified_dataset.json' +with tempfile.NamedTemporaryFile('w', dir=str(data_dir), delete=False, encoding='utf-8') as tmp_f: + json.dump(combined, tmp_f, indent=2, ensure_ascii=False) + tmp_name = tmp_f.name + +os.replace(tmp_name, str(dest_path)) + +print("\nSaved to classified_dataset.json (now with llm_tier + clf_tier)") diff --git a/scripts/retry_errors.py b/scripts/retry_errors.py new file mode 100644 index 00000000..26724ccc --- /dev/null +++ b/scripts/retry_errors.py @@ -0,0 +1,111 @@ +"""Retry the 94 failed prompts with 800-char truncation (safe for 4096-ctx model).""" +import json, urllib.request, time, subprocess +from pathlib import Path +from collections import Counter + +PROMPT_TEMPLATE = """Classify the coding task complexity. Output ONLY the tier name. + +agent-simple-core: trivial one-liners, syntax fixes, single-line edits +agent-medium-core: single-function changes, light refactoring, simple tests +agent-complex-core: multi-file changes, algorithmic work, data pipelines +agent-reasoning-core: deep analysis, architecture decisions, debugging complex systems +agent-advanced-core: system-level architecture, cross-cutting concerns, novel design + +Task: """ + +MAX_CHARS = 600 # proven safe across all prompt types in this dataset + +def get_model_port(): + """Discover the gemma4-26a4b-routing model's direct port (bypass router prompt-cache bug).""" + req = urllib.request.Request('http://127.0.0.1:8080/v1/models') + with urllib.request.urlopen(req, timeout=5) as resp: + data = json.loads(resp.read()) + for m in data.get('data', []): + if 'gemma4-26a4b' in m.get('id', ''): + status_obj = m.get('status') or {} + args = status_obj.get('args', []) if isinstance(status_obj, dict) else [] + for i, v in enumerate(args): + if v == '--port' and i + 1 < len(args): + return args[i + 1] + raise RuntimeError("gemma4-26a4b-routing model port not found") + +MODEL_PORT = get_model_port() +MODEL_URL = f"http://127.0.0.1:{MODEL_PORT}/v1/chat/completions" +print(f"Using model directly on port {MODEL_PORT}") + +def classify(prompt): + """Query the direct model port to classify the prompt complexity, handling truncations.""" + if len(prompt) > MAX_CHARS: + prompt = prompt[:MAX_CHARS] + payload = { + "model": "gemma4-26a4b-routing", + "messages": [{"role": "user", "content": PROMPT_TEMPLATE + prompt}], + "temperature": 0.0, + "max_tokens": 15, + } + req = urllib.request.Request( + MODEL_URL, + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json"} + ) + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read()) + choices = data.get("choices", []) + content = "" + if choices: + content = choices[0].get("message", {}).get("content", "").strip() + # Normalize: strip "tier:" prefix, extract just the tier name + for tier in TIERS: + if tier in content: + return tier + return "ERROR" + +TIERS = ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] + +data_dir = Path(__file__).resolve().parent.parent / "data" +with open(data_dir / "classified_dataset.json") as f: + dataset = json.load(f) +with open(data_dir / "raw_prompts_hermes.json") as f: + all_prompts = json.load(f) + +# Schema-aware: support both old schema ("tier") and new reclassify_all.py schema ("clf_tier") +error_indices = [ + i for i, p in enumerate(dataset.get('prompts', [])) + if p.get('tier') == 'ERROR' or p.get('clf_tier') == 'ERROR' +] +print(f"Retrying {len(error_indices)} failed prompts (max {MAX_CHARS} chars)...") + +fixed = 0 +errors = 0 + +for batch_start in range(0, len(error_indices), 5): + batch = error_indices[batch_start:batch_start + 5] + for idx in batch: + prompts_list = dataset.get('prompts', []) + prompt = prompts_list[idx].get('prompt') if idx < len(prompts_list) else "" + try: + tier = classify(prompt) + if idx < len(prompts_list): + prompts_list[idx]['tier'] = tier + if tier != 'ERROR': # only count as fixed if classification succeeded + fixed += 1 + except Exception as e: + errors += 1 + print(f" [{idx}] still failing: {str(e)[:80]}") + time.sleep(3) # single-slot server needs headroom + if batch_start + 5 < len(error_indices): + print(f" batch {batch_start//5 + 1}/{(len(error_indices)+4)//5}: {fixed} fixed, {errors} errors") + time.sleep(5) + +from collections import Counter +new_counts = Counter(p.get('llm_tier') or p.get('tier', 'ERROR') for p in dataset.get('prompts', [])) +dataset['counts'] = {k: v for k, v in new_counts.items()} +dataset['gaps'] = [t for t in ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] + if new_counts.get(t, 0) < 20] + +with open(data_dir / "classified_dataset.json", 'w') as f: + json.dump(dataset, f, indent=2, ensure_ascii=False) + +print(f"\nDone. Fixed: {fixed}, Errors: {errors}") +for tier in sorted(new_counts.keys()): + print(f" {tier:30s} {new_counts[tier]:3d}") \ No newline at end of file diff --git a/test_a2_verify.py b/test_a2_verify.py index 13c872c0..42cde1e1 100644 --- a/test_a2_verify.py +++ b/test_a2_verify.py @@ -1,16 +1,18 @@ #!/usr/bin/env python3 """Verify circuit breaker integration into agy_proxy.py""" import sys -sys.path.insert(0, '/home/gpav/Vrac/LAB/AI/LLM-Routing/router') +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent / 'router')) from circuit_breaker import get_breaker from agy_proxy import try_agy_proxy import asyncio, time b = get_breaker() -b.tier = 3 -b.cooldown_until = time.time() + 18000 -b.probe_granted = False +for sub in (b.google, b.vendor): + sub.tier = 3 + sub.cooldown_until = time.time() + 18000 + sub.probe_granted = False result = asyncio.run(try_agy_proxy('test prompt')) assert result is None, f'Breaker should return None when blocked, got: {result}' diff --git a/test_circuit_breaker.py b/test_circuit_breaker.py index 932c2960..becb686a 100644 --- a/test_circuit_breaker.py +++ b/test_circuit_breaker.py @@ -1,135 +1,176 @@ #!/usr/bin/env python3 """ -Integration test for the agy circuit breaker. +Integration test for the agy dual circuit breaker. -Simulates 4 consecutive quota failures and verifies: +Simulates consecutive quota failures and verifies: + - Independent google and vendor breakers - Tier 1 cooldown (5 min) after 1st failure - Tier 2 cooldown (30 min) after 2nd failure - Tier 3 cooldown (5 hours) after 3rd failure - Probe behavior: one allowed attempt after cooldown - Reset to Tier 0 on success - Stay at Tier 3 on repeated failure + - Backward compatibility of master breaker methods """ import sys import time -sys.path.insert(0, '/home/gpav/Vrac/LAB/AI/LLM-Routing/router') +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent / 'router')) from circuit_breaker import get_breaker, TIER_COOLDOWNS, MAX_TIER +def reset_breakers(): + b = get_breaker() + for sub in (b.google, b.vendor): + sub.tier = 0 + sub.cooldown_until = 0.0 + sub.probe_granted = False + sub.total_trips = 0 + sub.last_trip_time = 0.0 + + def test_initial_state(): """Breaker starts at Tier 0 (open).""" + reset_breakers() b = get_breaker() - b.tier = 0 - b.cooldown_until = 0 - b.probe_granted = False - b.total_trips = 0 - assert b.is_allowed() == True + assert b.is_allowed() assert b.tier == 0 - print("✓ Initial state: Tier 0, agy allowed") + assert b.google.is_allowed() + assert b.vendor.is_allowed() + print("✓ Initial state: Tier 0, allowed") def test_first_failure_trips_to_tier1(): """1st failure → Tier 1, 5 min cooldown.""" + reset_breakers() b = get_breaker() - b.tier = 0 - b.cooldown_until = 0 - b.probe_granted = False - b.record_failure() - assert b.tier == 1, f"Expected tier 1, got {b.tier}" - assert b.cooldown_until > time.time(), "Cooldown should be set" - assert b.is_allowed() == False, "Should block during cooldown" - print("✓ 1st failure → Tier 1 (5 min cooldown)") + + b.google.record_failure() + assert b.google.tier == 1 + assert b.google.cooldown_until > time.time() + assert not b.google.is_allowed() + + # Master breaker is still allowed because vendor is allowed (backward compatible fallback) + assert b.is_allowed() + print("✓ 1st failure → Tier 1 (5 min cooldown) on google breaker") def test_probe_granted_after_cooldown(): """After cooldown expires, exactly one probe is allowed.""" + reset_breakers() b = get_breaker() - b.tier = 1 - b.cooldown_until = time.time() - 10 # expired 10s ago - b.probe_granted = False - assert b.is_allowed() == True, "Probe should be granted" - assert b.probe_granted == True, "Probe flag should be set" - assert b.is_allowed() == False, "Second call should be denied" + + b.google.tier = 1 + b.google.cooldown_until = time.time() - 10 # expired 10s ago + b.google.probe_granted = False + + assert b.google.is_allowed(), "Probe should be granted" + assert b.google.probe_granted == True, "Probe flag should be set" + assert not b.google.is_allowed(), "Second call should be denied" print("✓ Probe granted after cooldown expiry, consumed on next check") def test_probe_failure_advances_tier(): """Probe failure → advance to next tier.""" + reset_breakers() b = get_breaker() - b.tier = 1 - b.cooldown_until = time.time() - 10 - b.probe_granted = True # probe was granted - b.record_failure() # probe fails - assert b.tier == 2, f"Expected tier 2, got {b.tier}" - assert b.probe_granted == False + + b.google.tier = 1 + b.google.cooldown_until = time.time() - 10 + b.google.probe_granted = True # probe was granted + b.google.record_failure() # probe fails + + assert b.google.tier == 2, f"Expected tier 2, got {b.google.tier}" + assert not b.google.probe_granted print("✓ Failed probe at Tier 1 → advanced to Tier 2 (30 min)") def test_tier3_stays_at_tier3(): """At Tier 3, failure → stays at Tier 3 (renews cooldown).""" + reset_breakers() b = get_breaker() - b.tier = MAX_TIER - b.cooldown_until = time.time() - 10 - b.probe_granted = True - old_until = b.cooldown_until - b.record_failure() - assert b.tier == MAX_TIER, "Should stay at Tier 3" - assert b.cooldown_until > old_until, "Cooldown should be renewed" - assert b.probe_granted == False + + b.google.tier = MAX_TIER + b.google.cooldown_until = time.time() - 10 + b.google.probe_granted = True + old_until = b.google.cooldown_until + b.google.record_failure() + + assert b.google.tier == MAX_TIER, "Should stay at Tier 3" + assert b.google.cooldown_until > old_until, "Cooldown should be renewed" + assert not b.google.probe_granted print("✓ Tier 3 failure → stays at Tier 3 (renews 5-hour cooldown)") def test_success_resets(): """Success at any tier → reset to Tier 0.""" + reset_breakers() b = get_breaker() - b.tier = 2 - b.cooldown_until = time.time() + 1000 - b.probe_granted = False - b.record_success() - assert b.tier == 0 - assert b.is_allowed() == True + + b.google.tier = 2 + b.google.cooldown_until = time.time() + 1000 + b.google.probe_granted = False + b.google.record_success() + + assert b.google.tier == 0 + assert b.google.is_allowed() print("✓ Success resets breaker to Tier 0 from any tier") +def test_backward_compatibility(): + """Master breaker record_failure and record_success affect both breakers.""" + reset_breakers() + b = get_breaker() + + b.record_failure() + assert b.google.tier == 1 + assert b.vendor.tier == 1 + assert not b.is_allowed() # both blocked + + b.record_success() + assert b.google.tier == 0 + assert b.vendor.tier == 0 + assert b.is_allowed() + print("✓ Master record_failure and record_success maintain compatibility") + + def test_full_cycle(): """Complete cycle: success → 3 failures → probe success → reset.""" + reset_breakers() b = get_breaker() - b.tier = 0 - b.cooldown_until = 0 - b.probe_granted = False - b.total_trips = 0 + sub = b.google # Operate normally - assert b.is_allowed() - b.record_success() - assert b.tier == 0 + assert sub.is_allowed() + sub.record_success() + assert sub.tier == 0 # 1st failure - b.record_failure() - assert b.tier == 1 - assert not b.is_allowed() + sub.record_failure() + assert sub.tier == 1 + assert not sub.is_allowed() # Simulate cooldown expiry - b.cooldown_until = time.time() - 10 - assert b.is_allowed() # probe granted - b.record_failure() # probe fails - assert b.tier == 2 + sub.cooldown_until = time.time() - 10 + assert sub.is_allowed() # probe granted + sub.record_failure() # probe fails + assert sub.tier == 2 # Simulate cooldown expiry - b.cooldown_until = time.time() - 10 - assert b.is_allowed() # probe granted - b.record_failure() # probe fails again - assert b.tier == 3 + sub.cooldown_until = time.time() - 10 + assert sub.is_allowed() # probe granted + sub.record_failure() # probe fails again + assert sub.tier == 3 assert TIER_COOLDOWNS[3] == 18000, "Tier 3 must be 5 hours" # Simulate cooldown expiry + probe success - b.cooldown_until = time.time() - 10 - assert b.is_allowed() # probe granted - b.record_success() # probe succeeds - assert b.tier == 0 - assert b.total_trips == 3 + sub.cooldown_until = time.time() - 10 + assert sub.is_allowed() # probe granted + sub.record_success() # probe succeeds + assert sub.tier == 0 + assert sub.total_trips == 3 print("✓ Full cycle: 3 failures → Tier 3 → probe success → reset") @@ -141,6 +182,7 @@ def test_full_cycle(): test_probe_failure_advances_tier() test_tier3_stays_at_tier3() test_success_resets() + test_backward_compatibility() test_full_cycle() print("\n" + "=" * 60) diff --git a/verify_breaker.py b/verify_breaker.py index 4dbbd194..9198b583 100644 --- a/verify_breaker.py +++ b/verify_breaker.py @@ -1,17 +1,26 @@ #!/usr/bin/env python3 """Verification test for the agy circuit breaker.""" +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent)) + from router.circuit_breaker import get_breaker b = get_breaker() -assert b.is_allowed() == True, 'Tier 0 should be open' -b.record_failure() -assert b.tier == 1, 'Should be at Tier 1' -assert b.is_allowed() == False, 'Tier 1 should block (cooldown active)' -# Force cooldown expiry -b.cooldown_until = 0 -assert b.is_allowed() == True, 'Probe should be granted' -assert b.probe_granted == True -b.record_failure() # probe fails -assert b.tier == 2, 'Should advance to Tier 2' +assert b.is_allowed(), 'Tier 0 should be open' + +for sub in (b.google, b.vendor): + assert sub.is_allowed() + sub.record_failure() + assert sub.tier == 1, 'Should be at Tier 1' + assert not sub.is_allowed(), 'Tier 1 should block (cooldown active)' + # Force cooldown expiry + sub.cooldown_until = 0 + assert sub.is_allowed(), 'Probe should be granted' + assert sub.probe_granted == True + sub.record_failure() # probe fails + assert sub.tier == 2, 'Should advance to Tier 2' + +assert b.tier == 2 print('All assertions passed')