diff --git a/.github/dependabot.yml b/.github/dependabot.yml index a2108716..2010274f 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -19,7 +19,7 @@ updates: directory: "/" schedule: interval: "daily" - time: "03:23" + time: "02:55" open-pull-requests-limit: 5 labels: - "dependencies" diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 0bd6e32a..3de6b85c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -4,7 +4,6 @@ on: push: branches: [ master ] pull_request: - branches: [ master ] jobs: test: @@ -20,6 +19,9 @@ jobs: with: python-version: '3.11' + - name: Install dependencies + run: pip install httpx==0.28.1 + - name: Run Circuit Breaker Tests run: python3 test_circuit_breaker.py diff --git a/README.md b/README.md index b2838ecf..d7f0807e 100644 --- a/README.md +++ b/README.md @@ -141,16 +141,33 @@ sequenceDiagram end else Route = ollama (llm-routing-ollama, auto-ollama, auto-agy-ollama chain) - Note over Router: Proxy to LiteLLM as ollama-deepseek-v4-pro + Note over Router: Proxy to LiteLLM as ollama-deepseek-v4-pro / -flash Router->>Proxy: POST /v1/chat/completions (model=ollama-deepseek-v4-pro) Proxy->>Provider: Call api.ollama.com (ollama_chat provider) alt Ollama Succeeds Provider-->>Proxy: deepseek-v4-pro response Proxy-->>Router: Return response - Router-->>Client: Return chat completion + Router-->>Client: Return response else Ollama fails (rate-limited / unavailable) - Note over Proxy: Fallback: ollama-deepseek-v4-pro → agent-advanced-core - Proxy->>Provider: Call OpenRouter (agent-advanced-core tier) + Provider-->>Proxy: Error / HTTP 429 + Proxy-->>Router: Error Response + Note over Router: Activate 5-min router-side Ollama cooldown + alt Model = auto-ollama / auto-agy-ollama (triage-gated) + Note over Router: Catch error → Fall back to free tier + Router->>Proxy: POST /v1/chat/completions (model=original_target_model) + Proxy->>Provider: Call OpenRouter (free tier cascade) + Provider-->>Proxy: Response + Proxy-->>Router: Response + Router-->>Client: Return response + else Model = llm-routing-ollama (direct / fallback chain) + Note over Router: Return 429 (cooldown active) + Router-->>Proxy: HTTP 429 (Ollama cooled down) + Note over Proxy: Skip llm-routing-ollama → cascade to openrouter-auto + Proxy->>Provider: Call OpenRouter (openrouter-auto) + Provider-->>Proxy: Response + Proxy-->>Router: Response + Router-->>Client: Return response + end end else Route = LiteLLM (all other models) @@ -180,11 +197,11 @@ The gateway supports multiple routing modes controlled by the `model` field: | Model | Behavior | |-------|----------| | `llm-routing-auto-free` | **Full classifier pipeline** → routes to best free tier. Recommended default. | -| `llm-routing-auto-agy` | **Classifier + agy (gated)**: runs classifier, tries agy only if classified as advanced. | -| `llm-routing-auto-ollama` | **Classifier + Ollama (gated)**: runs classifier, tries Ollama only if advanced. | -| `llm-routing-auto-agy-ollama` | **Classifier → agy → ollama (gated)**: runs classifier, chains agy then Ollama only if advanced. | +| `llm-routing-auto-agy` | **Classifier + agy (gated)**: runs classifier, tries agy only if classified as advanced/reasoning. | +| `llm-routing-auto-ollama` | **Classifier + Ollama (gated)**: runs classifier, reasoning & advanced → `ollama-deepseek-v4-pro`, complex → `ollama-deepseek-v4-flash`, below (medium/simple) → bypasses Ollama to LiteLLM free tiers. | +| `llm-routing-auto-agy-ollama` | **Classifier → agy → ollama (gated)**: runs classifier, chains agy then Ollama only if advanced/reasoning/complex. | | `llm-routing-agy` | **Direct agy**: skips classifier, agy proxy (Gemini/Claude) → LiteLLM fallback. | -| `llm-routing-ollama` | **Direct Ollama**: skips classifier, Ollama deepseek-v4-pro → LiteLLM fallback. | +| `llm-routing-ollama` | **Gated Ollama**: runs classifier, reasoning & advanced → `ollama-deepseek-v4-pro`, complex & below → `ollama-deepseek-v4-flash`. | | `agent-simple-core` / `agent-medium-core` / `agent-complex-core` / `agent-reasoning-core` / `agent-advanced-core` | **Direct routing**: bypasses classifier, goes straight to LiteLLM with that tier name. | | Anything else | Returns HTTP 400 with the list of available models | @@ -234,19 +251,22 @@ Exposes the entry endpoint (`http://localhost:5000/v1`) and evaluates prompt com **Backend targets dispatched by the router** (all resolve through LiteLLM on port 4000): -| Model | Classifier | Premium backend | Fallback | -|:---|---:|:---|:---| -| `llm-routing-auto-free` | ✅ | — | LiteLLM with classified tier | 256K | -| `llm-routing-auto-agy` | ✅ | agy (gated: reasoning → gemini-3.5-flash, advanced → gemini-3.5-flash → claude-opus-4.6) | LiteLLM with classified tier | 256K | -| `llm-routing-auto-ollama` | ✅ | Ollama (gated: reasoning → deepseek-v4-flash, advanced → deepseek-v4-pro) | LiteLLM agent-advanced-core | 256K | -| `llm-routing-auto-agy-ollama` | ✅ | agy → Ollama (gated: reasoning/advanced only) | LiteLLM with classified tier | 256K | -| `llm-routing-agy` | ❌ | agy (Gemini/Claude) — unconditional | LiteLLM agent-advanced-core | 256K | -| `llm-routing-ollama` | ❌ | Ollama deepseek-v4-pro — unconditional | LiteLLM agent-advanced-core | 256K | -| `agent-advanced-core` | ❌ | — | LiteLLM openrouter-auto | -| `agent-reasoning-core` | ❌ | — | LiteLLM fallback chain | -| `agent-complex-core` | ❌ | — | LiteLLM fallback chain | -| `agent-medium-core` | ❌ | — | LiteLLM fallback chain | -| `agent-simple-core` | ❌ | — | LiteLLM fallback chain | +| Model | Classifier | Premium backend | Fallback | Context Length | +|:---|---:|:---|:---|:---| +| `llm-routing-auto-free` | ✅ | — | LiteLLM with classified tier | 262K | +| `llm-routing-auto-agy` | ✅ | agy (gated: reasoning → gemini-3.5-flash, advanced → gemini-3.5-flash → claude-opus-4.6) | LiteLLM with classified tier | 262K | +| `llm-routing-auto-ollama` | ✅ | Ollama (gated: reasoning & advanced → ollama-deepseek-v4-pro, complex → ollama-deepseek-v4-flash, below → bypass) | LiteLLM with classified tier | 512K | +| `llm-routing-auto-agy-ollama` | ✅ | agy → Ollama (gated: reasoning/advanced/complex) | LiteLLM with classified tier | 512K | +| `llm-routing-agy` | ❌ | agy (Gemini/Claude) — unconditional | LiteLLM agent-advanced-core | 1M | +| `llm-routing-ollama` | ✅ | Ollama (gated: reasoning & advanced → ollama-deepseek-v4-pro, complex & below → ollama-deepseek-v4-flash) | LiteLLM agent-advanced-core / agent-reasoning-core | 512K | +| `agent-advanced-core` | ❌ | — | LiteLLM openrouter-auto | 262K | +| `agent-reasoning-core` | ❌ | — | LiteLLM fallback chain | 262K | +| `agent-complex-core` | ❌ | — | LiteLLM fallback chain | 262K | +| `agent-medium-core` | ❌ | — | LiteLLM fallback chain | 262K | +| `agent-simple-core` | ❌ | — | LiteLLM fallback chain | 256K | + +> [!TIP] +> Model capabilities, token limits, and costs are visible in LiteLLM's Model Hub Table at `http://localhost:4000/ui/?page=model-hub-table` (or port 4000 on the gateway host). ### B. LiteLLM Proxy Gateway (`litellm/config.yaml`) - **Version Pinning**: The LiteLLM gateway runs `ghcr.io/berriai/litellm:v1.88.0` (latest stable as of June 2026). The tag is explicitly pinned in `pod.yaml` — never use `:latest`. Check available tags with `skopeo list-tags docker://ghcr.io/berriai/litellm` before upgrading. ClickHouse runs `docker.io/clickhouse/clickhouse-server:26.5.1` (upgraded from 24.8, June 2026). Valkey Cache runs `docker.io/valkey/valkey:9.1.0-alpine` (upgraded from 8, June 2026). @@ -259,14 +279,62 @@ Orchestrates routing fallback chains, Redis caching, and telemetry callbacks: - `embedding_model: "local-nomic-embed"` — uses the local nomic-embed model (no API costs) - `collection_name: "litellm_semantic_cache"` — stores embeddings for similarity-based cache lookups - **Cascading Fallback Chains** (configured in `litellm_settings.fallbacks`): - Each tier escalates through increasingly capable models. All chains end at `openrouter-auto` (LiteLLM's internal fallback to OpenRouter `/auto`). `local-qwen-3.6` (35B) was disabled 2026-06-08 to free 23GB RAM/GTT. - - **`agent-simple-core`**: medium-core → complex-core → reasoning-core → advanced-core → `openrouter-auto` - - **`agent-medium-core`**: complex-core → reasoning-core → advanced-core → `openrouter-auto` - - **`agent-complex-core`**: reasoning-core → advanced-core → `openrouter-auto` - - **`agent-reasoning-core`**: advanced-core → `openrouter-auto` - - **`agent-advanced-core`**: `openrouter-auto` - - **`ollama-deepseek-v4-pro`**: advanced-core → `openrouter-auto` - All tiers ultimately land on the local Ryzen APU MoE (`qwen-35b-q4ks` via llama-server on :8080) as the final safety net. + Each tier escalates through increasingly capable free models, then paid local/remote Ollama models, and finally falls back to `openrouter-auto` (LiteLLM's internal fallback to OpenRouter `/auto`). `local-qwen-3.6` (35B) was disabled 2026-06-08 to free 23GB RAM/GTT. + + ```mermaid + graph TD + %% Define styles + classDef simple fill:#4F46E5,stroke:#312E81,stroke-width:2px,color:#fff; + classDef medium fill:#7C3AED,stroke:#4C1D95,stroke-width:2px,color:#fff; + classDef complex fill:#DB2777,stroke:#831843,stroke-width:2px,color:#fff; + classDef reasoning fill:#EA580C,stroke:#7C2D12,stroke-width:2px,color:#fff; + classDef advanced fill:#E11D48,stroke:#881337,stroke-width:2px,color:#fff; + classDef premium fill:#059669,stroke:#064E3B,stroke-width:2px,color:#fff; + classDef auto fill:#4B5563,stroke:#1F2937,stroke-width:2px,color:#fff; + + subgraph Simple["agent-simple-core Fallback Tree"] + S[agent-simple-core]:::simple --> SM[agent-medium-core]:::medium + SM --> SC[agent-complex-core]:::complex + SC --> SR[agent-reasoning-core]:::reasoning + SR --> SA[agent-advanced-core]:::advanced + SA --> SO1[llm-routing-ollama]:::premium + SO1 --> SAU[openrouter-auto]:::auto + end + + subgraph Medium["agent-medium-core Fallback Tree"] + M[agent-medium-core]:::medium --> MC[agent-complex-core]:::complex + MC --> MR[agent-reasoning-core]:::reasoning + MR --> MA[agent-advanced-core]:::advanced + MA --> MO1[llm-routing-ollama]:::premium + MO1 --> MAU[openrouter-auto]:::auto + end + + subgraph Complex["agent-complex-core Fallback Tree"] + C[agent-complex-core]:::complex --> CR[agent-reasoning-core]:::reasoning + CR --> CA[agent-advanced-core]:::advanced + CA --> CO1[llm-routing-ollama]:::premium + CO1 --> CAU[openrouter-auto]:::auto + end + + subgraph Reasoning["agent-reasoning-core Fallback Tree"] + R[agent-reasoning-core]:::reasoning --> RA[agent-advanced-core]:::advanced + RA --> RO1[llm-routing-ollama]:::premium + RO1 --> RAU[openrouter-auto]:::auto + end + + subgraph Advanced["agent-advanced-core Fallback Tree"] + A[agent-advanced-core]:::advanced --> AO1[llm-routing-ollama]:::premium + AO1 --> AAU[openrouter-auto]:::auto + end + ``` + + - **`agent-simple-core`**: medium-core → complex-core → reasoning-core → advanced-core → `llm-routing-ollama` → `openrouter-auto` + - **`agent-medium-core`**: complex-core → reasoning-core → advanced-core → `llm-routing-ollama` → `openrouter-auto` + - **`agent-complex-core`**: reasoning-core → advanced-core → `llm-routing-ollama` → `openrouter-auto` + - **`agent-reasoning-core`**: advanced-core → `llm-routing-ollama` → `openrouter-auto` + - **`agent-advanced-core`**: `llm-routing-ollama` → `openrouter-auto` + - **`llm-routing-ollama`** (classifier-gated proxy): `reasoning & advanced` → `ollama-deepseek-v4-pro`, `complex & below` → `ollama-deepseek-v4-flash`. Note: Ollama cooldowns are managed by the triage router internally (5-minute window on failure); during cooldown the router returns 429 immediately so LiteLLM skips to `openrouter-auto`. + All tiers ultimately land on OpenRouter auto/free model pools or the local Speculative MoE when enabled. *Note: Premium routing is controlled by the model name, not by the tier. `llm-routing-agy` and `llm-routing-auto-agy` trigger the agy proxy (Google/Claude via Cloud Code Assist) — but auto models only trigger agy if the classifier returns `agent-advanced-core`. `llm-routing-ollama` and `llm-routing-auto-ollama` route through Ollama.com (deepseek-v4-pro via LiteLLM's ollama_chat provider) — same gating for auto models. `llm-routing-auto-agy-ollama` chains both: agy first, then Ollama if agy is exhausted, both gated on advanced classification. The `agent-advanced-core` tier itself is a plain LiteLLM tier with no premium trigger. See §2 for the full routing table.* ### C. Valkey Caching (`redis_settings` in LiteLLM) @@ -434,18 +502,22 @@ This endpoint outputs plain-text metrics (`Content-Type: text/plain; version=0.0 | Metric | Type | Description | | :--- | :---: | :--- | | `triage_requests_total` | gauge | Total number of requests processed | -| `simple_requests_total` | gauge | Number of simple/routine requests | +| `simple_requests_total` | gauge | Number of simple requests | +| `medium_requests_total` | gauge | Number of medium requests | | `complex_requests_total` | gauge | Number of complex requests | +| `reasoning_requests_total` | gauge | Number of reasoning requests | +| `advanced_requests_total` | gauge | Number of advanced requests | | `cache_hits_total` | gauge | Triage cache hit count | | `avg_triage_latency_ms` | gauge | Average triage classification latency | | `avg_proxy_latency_ms` | gauge | Average proxy/inference latency | | `prompt_tokens_total` | counter | Total prompt tokens processed | | `completion_tokens_total` | counter | Total completion tokens generated | -| `circuit_breaker_tier` | gauge | Current circuit breaker cooldown tier (0=open, 1-3=cooldown) | +| `circuit_breaker_google_tier` | gauge | Google breaker cooldown tier (0=open, 3=max) | +| `circuit_breaker_vendor_tier` | gauge | Vendor breaker cooldown tier (0=open, 3=max) | | `circuit_breaker_agy_allowed` | gauge | Whether agy proxy requests are currently allowed (0/1) | -| `circuit_breaker_cooldown_remaining_seconds` | gauge | Seconds until cooldown expires | -| `circuit_breaker_total_trips` | counter | Total number of circuit breaker trips | -| `circuit_breaker_probe_granted` | gauge | Whether a probe request has been granted (0/1) | +| `circuit_breaker_total_trips` | counter | Total trips across both breakers | +| `ollama_cooldown_active` | gauge | Whether Ollama is in router-side cooldown (1=active, 0=inactive) | +| `ollama_cooldown_remaining_seconds` | gauge | Seconds remaining in Ollama cooldown | Verify the endpoint: ```bash @@ -676,29 +748,35 @@ LiteLLM calls `https://api.ollama.com/api/chat` with Bearer authentication using | Model | Ollama tag | Purpose | |-------|-----------|---------| -| `ollama-deepseek-v4-pro` | `deepseek-v4-pro` | Primary paid tier — 1.6T parameter model | +| `ollama-deepseek-v4-pro` | `deepseek-v4-pro` | Primary paid tier — 1.6T parameter reasoning & advanced model | +| `ollama-deepseek-v4-flash` | `deepseek-v4-flash` | Lightweight paid tier — fast complex & below model | Additional Ollama.com models can be added to `litellm/config.yaml` using the same `ollama_chat/` prefix pattern. -### Fallback Chain +### Fallback and Cooldown Behavior -``` -ollama-deepseek-v4-pro → agent-advanced-core → openrouter-auto -``` +To prevent cascading fallback loops where a rate-limited Ollama backend repeatedly receives requests from different tiers, the **Triage Router manages Ollama cooldowns internally** rather than relying on LiteLLM's deployment cooldown mechanism (which is unreliable for single-deployment model groups in Community Edition). + +The cooldown mechanism works as follows: -If Ollama.com is rate-limited or unavailable, LiteLLM automatically falls back through -the agent tier cascade to OpenRouter's free model pool. +1. **Failure Detection**: When calls to `ollama-deepseek-v4-pro` or `ollama-deepseek-v4-flash` fail (due to rate limiting, 429/502/503 errors, or connection issues), the failure is caught by the Triage Router. +2. **Router-Side Cooldown Activation**: The Triage Router activates an internal **5-minute cooldown** (configurable via `OLLAMA_COOLDOWN_SECONDS` env var). During this window, subsequent requests targeting the Ollama backend are immediately short-circuited (either rejected or redirected) to prevent downstream Ollama backend calls. +3. **Direct / Fallback Requests (`llm-routing-ollama`)**: + - During cooldown, the Triage Router returns an HTTP `429` immediately. + - LiteLLM receives this 429, skips `llm-routing-ollama` in the fallback chain, and cascades directly to `openrouter-auto`. +4. **Auto-Routing Requests (`llm-routing-auto-ollama` or `llm-routing-auto-agy-ollama`)**: + - During cooldown, the Triage Router silently falls back to the original classified free tier model (e.g., `agent-advanced-core`), querying LiteLLM for a free model. ### Routing Modes | Model | Behavior | |-------|----------| -| `llm-routing-ollama` | Direct — skips classifier, goes straight to Ollama deepseek-v4-pro | -| `llm-routing-auto-ollama` | Auto — classifier runs first, then Ollama deepseek-v4-pro is tried regardless of tier | -| `llm-routing-auto-agy-ollama` | Chained — classifier runs, tries agy first, then chains to Ollama if agy exhausted | +| `llm-routing-ollama` | **Gated direct**: runs classifier, routes reasoning & advanced → `ollama-deepseek-v4-pro`, complex & below → `ollama-deepseek-v4-flash` | +| `llm-routing-auto-ollama` | **Gated auto**: runs classifier, reasoning & advanced → `ollama-deepseek-v4-pro`, complex → `ollama-deepseek-v4-flash`, below → bypasses Ollama to LiteLLM free tiers | +| `llm-routing-auto-agy-ollama` | **Gated chained**: runs classifier, tries agy first (advanced/reasoning only) then chains to Ollama if agy is exhausted; for complex tasks, it goes straight to Ollama (flash) | -All three modes ultimately fall back to LiteLLM's agent tier cascade if the premium backends fail. +For auto-routing modes, the Triage Router handles failures by silently falling back to the classified free tier cascade. For direct requests to `llm-routing-ollama`, the router returns `429` immediately during cooldown, allowing LiteLLM to skip this model group and cascade to `openrouter-auto`. The cooldown status is visible via the `/metrics` endpoint (`ollama_cooldown_active` and `ollama_cooldown_remaining_seconds` gauges). ## 9d. Hermes Cron Jobs & Health Check Alerting diff --git a/hello.py b/hello.py deleted file mode 100644 index 7df869a1..00000000 --- a/hello.py +++ /dev/null @@ -1 +0,0 @@ -print("Hello, World!") diff --git a/litellm/config.yaml b/litellm/config.yaml index c27b47c7..4c2e8ce7 100644 --- a/litellm/config.yaml +++ b/litellm/config.yaml @@ -1,5 +1,5 @@ general_settings: - master_key: sk-lit...33bf + master_key: os.environ/LITELLM_MASTER_KEY litellm_settings: # ------------------------------------------------------------------------- # FALLBACK CHAINS (cascading, in order of escalation) @@ -20,41 +20,76 @@ litellm_settings: - langfuse detailed_debug: false drop_params: true + public_model_groups: + - openrouter-auto + - llm-routing-ollama + - ollama-deepseek-v4-pro + - ollama-deepseek-v4-flash + - agent-advanced-core + - agent-reasoning-core + - agent-complex-core + - agent-medium-core + - agent-simple-core fallbacks: - agent-simple-core: - agent-medium-core - agent-complex-core - agent-reasoning-core - agent-advanced-core + - llm-routing-ollama - openrouter-auto # - local-qwen-3.6 # DISABLED: 35B model unloaded (23GB GTT saved) - agent-medium-core: - agent-complex-core - agent-reasoning-core - agent-advanced-core + - llm-routing-ollama - openrouter-auto # - local-qwen-3.6 # DISABLED - agent-complex-core: - agent-reasoning-core - agent-advanced-core + - llm-routing-ollama - openrouter-auto # - local-qwen-3.6 # DISABLED - agent-reasoning-core: - agent-advanced-core + - llm-routing-ollama - openrouter-auto # - local-qwen-3.6 # DISABLED - agent-advanced-core: + - llm-routing-ollama - openrouter-auto # - local-qwen-3.6 # DISABLED - - ollama-deepseek-v4-pro: - - agent-advanced-core - - ollama-deepseek-v4-flash: - - agent-reasoning-core + model_list: - litellm_params: model: openrouter/openrouter/auto request_timeout: 120 model_name: openrouter-auto + model_info: + supports_vision: true + supports_reasoning: true + supports_function_calling: true + mode: chat + max_tokens: 2000000 + max_input_tokens: 2000000 + is_public_model_group: true +- litellm_params: + model: openai/llm-routing-ollama + api_base: http://127.0.0.1:5000/v1 + api_key: os.environ/LITELLM_MASTER_KEY + request_timeout: 120 + model_name: llm-routing-ollama + model_info: + supports_vision: true + supports_reasoning: true + supports_function_calling: true + mode: chat + max_tokens: 524288 + max_input_tokens: 524288 + is_public_model_group: true + # DISABLED 2026-06-08 — 20GB on disk, 23GB GTT (system RAM as GPU buffer). # Uncomment to re-enable once a lighter model replaces it. #- litellm_params: @@ -73,7 +108,10 @@ model_list: # ================================================================================ # OLLAMA PAID TIER — ollama.com via LiteLLM's native ollama_chat provider. # LiteLLM calls https://api.ollama.com/api/chat with Bearer auth (OLLAMA_API_KEY). -# Fallback: ollama-deepseek-v4-pro → agent-advanced-core → openrouter-auto. +# No LiteLLM-level fallbacks: failures propagate back to the triage router +# (router/main.py) which manages Ollama cooldowns internally. When Ollama fails, +# the router activates a 5-minute cooldown and skips Ollama on subsequent +# requests, returning 429 so LiteLLM falls through to openrouter-auto. # ================================================================================ - model_name: ollama-deepseek-v4-pro litellm_params: @@ -81,10 +119,20 @@ model_list: api_base: https://api.ollama.com api_key: os.environ/OLLAMA_API_KEY request_timeout: 120 + model_info: + supports_vision: true + supports_reasoning: true + supports_function_calling: true + mode: chat + max_tokens: 524288 + max_input_tokens: 524288 + input_cost_per_token: 0.00000174 + output_cost_per_token: 0.00000348 + is_public_model_group: true # ================================================================================ # OLLAMA FLASH TIER — lighter/faster model for reasoning-tier requests. -# Fallback: ollama-deepseek-v4-flash → agent-reasoning-core → openrouter-auto. +# Same cooldown architecture as the pro tier above. # ================================================================================ - model_name: ollama-deepseek-v4-flash litellm_params: @@ -92,6 +140,16 @@ model_list: api_base: https://api.ollama.com api_key: os.environ/OLLAMA_API_KEY request_timeout: 120 + model_info: + supports_vision: true + supports_reasoning: true + supports_function_calling: true + mode: chat + max_tokens: 524288 + max_input_tokens: 524288 + input_cost_per_token: 0.00000014 + output_cost_per_token: 0.00000028 + is_public_model_group: true # ================================================================================ # AGENT TIER DEPLOYMENTS — safety-net entries (one per tier) @@ -119,43 +177,78 @@ model_list: - model_name: agent-advanced-core litellm_params: - model: openrouter/minimax/minimax-m2.5:free - request_timeout: 120 + model: openrouter/google/gemma-4-31b-it:free + request_timeout: 20 + model_info: + supports_vision: false + supports_reasoning: true + supports_function_calling: true + mode: chat + max_tokens: 262144 + max_input_tokens: 262144 + is_public_model_group: true - model_name: agent-reasoning-core litellm_params: - model: openrouter/moonshotai/kimi-k2.6:free - request_timeout: 120 + model: openrouter/google/gemma-4-26b-a4b-it:free + request_timeout: 20 + model_info: + supports_vision: false + supports_reasoning: true + supports_function_calling: true + mode: chat + max_tokens: 262144 + max_input_tokens: 262144 + is_public_model_group: true - model_name: agent-complex-core litellm_params: model: openrouter/nvidia/nemotron-3-ultra-550b-a55b:free - request_timeout: 120 + request_timeout: 20 + model_info: + supports_vision: false + supports_reasoning: true + supports_function_calling: true + mode: chat + max_tokens: 262144 + max_input_tokens: 262144 + is_public_model_group: true - model_name: agent-medium-core litellm_params: model: openrouter/google/gemma-4-26b-a4b-it:free - request_timeout: 120 + request_timeout: 20 + model_info: + supports_vision: false + supports_reasoning: true + supports_function_calling: true + mode: chat + max_tokens: 262144 + max_input_tokens: 262144 + is_public_model_group: true - model_name: agent-simple-core litellm_params: model: openrouter/nvidia/nemotron-3-nano-30b-a3b:free - request_timeout: 120 + request_timeout: 20 + model_info: + supports_vision: false + supports_reasoning: true + supports_function_calling: true + mode: chat + max_tokens: 256000 + max_input_tokens: 256000 + is_public_model_group: true redis_settings: redis_host: 127.0.0.1 redis_port: 6379 router_settings: - allowed_fails: 2 + allowed_fails: 0 cooldown_time: 300 enable_pre_call_checks: false num_retries: 1 routing_strategy: latency-based-routing - # Per-error-type cooldown thresholds (overrides allowed_fails for specific errors). - # Upstream rate limits ("temporarily rate-limited upstream") can last minutes — - # a 30s cooldown just wastes retries. 300s gives providers time to clear the limit. - # RateLimitError: cooldown on FIRST 429 (upstream rate limits persist for minutes — - # allowing even 1 retry wastes a request against a provider that's still throttling) - allowed_fails_policy: - RateLimitErrorAllowedFails: 0 - TimeoutErrorAllowedFails: 3 - BadRequestErrorAllowedFails: 1 + enable_health_check_routing: false + # NOTE: allowed_fails_policy is an enterprise-only feature in LiteLLM. + # Ollama cooldowns are handled by the triage router itself (router/main.py), + # not by LiteLLM's deployment cooldown mechanism. vector_store_settings: collection_name: litellm_semantic_cache connection_string: postgresql://postgres:***@127.0.0.1:5432/postgres diff --git a/pod.yaml b/pod.yaml index 8960f681..0394ccb1 100644 --- a/pod.yaml +++ b/pod.yaml @@ -90,6 +90,10 @@ spec: env: - name: CONFIG_PATH value: /config/router_dir/config.yaml + - name: LITELLM_CONFIG_PATH + value: /config/litellm_dir/config.yaml + - name: DATABASE_URL + value: postgresql://postgres:***@127.0.0.1:5432/postgres - name: DBUS_SESSION_BUS_ADDRESS value: unix:path=/run/user/1000/bus - name: LITELLM_MASTER_KEY @@ -127,6 +131,8 @@ spec: volumeMounts: - mountPath: /config/router_dir name: router-config + - mountPath: /config/litellm_dir + name: litellm-config - mountPath: /app/data name: dataset-data - mountPath: /config/gemini_auth diff --git a/router/Dockerfile b/router/Dockerfile index c26ad091..e663b49e 100644 --- a/router/Dockerfile +++ b/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 uvicorn httpx pyyaml python-multipart asyncpg langfuse +RUN pip install --no-cache-dir fastapi uvicorn httpx pyyaml python-multipart asyncpg langfuse redis # 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/ diff --git a/router/agy_proxy.py b/router/agy_proxy.py index 918bfcb6..c625b56d 100644 --- a/router/agy_proxy.py +++ b/router/agy_proxy.py @@ -25,7 +25,19 @@ import os import shlex import time -from typing import Optional +import httpx +from typing import Optional, Protocol, runtime_checkable + +@runtime_checkable +class CooldownPersistence(Protocol): + """Interface for persisting/syncing Valkey cooldown state.""" + async def sync(self) -> None: + """Pull latest cooldown state from Valkey.""" + ... + + async def save(self) -> None: + """Push updated cooldown state to Valkey.""" + ... from circuit_breaker import get_google_breaker, get_vendor_breaker @@ -73,14 +85,12 @@ def _get_last_conversation_id() -> Optional[str]: return None -async def _run_agy_print(prompt: str, model_override: str = "", +async def _run_agy_print(client: httpx.AsyncClient, prompt: str, model_override: str = "", conversation_id: Optional[str] = None, timeout: float = AGY_TIMEOUT_SECS) -> tuple[int, str, str, Optional[str]]: """ Forward the agy execution request to the host-side agy daemon. """ - import httpx - url = "http://127.0.0.1:5005/run" payload = { "prompt": prompt, @@ -94,18 +104,17 @@ async def _run_agy_print(prompt: str, model_override: str = "", logger.info(f"agy proxy forwarding to host: [{model_tag}]{conv_tag} {prompt[:60]}...") try: - async with httpx.AsyncClient(timeout=timeout + 5.0) as client: - r = await client.post(url, json=payload) - if r.status_code == 200: - result = r.json() - return ( - result.get("returncode", 0), - result.get("stdout", ""), - result.get("stderr", ""), - result.get("conversation_id", None) - ) - else: - return -1, "", f"Daemon returned HTTP status {r.status_code}", None + r = await client.post(url, json=payload, timeout=timeout + 5.0) + if r.status_code == 200: + result = r.json() + return ( + result.get("returncode", 0), + result.get("stdout", ""), + result.get("stderr", ""), + result.get("conversation_id", None) + ) + else: + return -1, "", f"Daemon returned HTTP status {r.status_code}", None except Exception as e: logger.error(f"Failed to communicate with Host agy Daemon: {e}") return -1, "", f"Daemon connection error: {e}", None @@ -176,7 +185,9 @@ async def try_agy_proxy(prompt: str, messages: list = None, session_id: str = None, total_timeout: float = AGY_TOTAL_TIMEOUT_SECS, stream: bool = False, - target_tier: str = "agent-advanced-core") -> Optional[dict]: + target_tier: str = "agent-advanced-core", + client: Optional[httpx.AsyncClient] = None, + cooldown_persistence: Optional[CooldownPersistence] = None) -> Optional[dict]: """ Attempt agy proxy with session-aware tier fallback. @@ -188,6 +199,8 @@ async def try_agy_proxy(prompt: str, messages: list = None, stream: If True, returns a dict with {"stream": async_generator, "model": model_name} target_tier: Classified tier — "agent-reasoning-core" uses gemini-3.5-flash (low thinking), "agent-advanced-core" uses full 2-tier chain (gemini-3.5-flash → claude-opus-4.6) + client: Shared HTTP client instance from caller + cooldown_persistence: Valkey synchronization callback interface Returns: OpenAI-compatible response dict, streaming dict, or None if all tiers failed. @@ -201,224 +214,262 @@ async def try_agy_proxy(prompt: str, messages: list = None, ] else: agy_tiers = AGY_FALLBACK_TIERS # full chain: gemini-3.5-flash → claude-opus-4.6 - # Per-model circuit breakers — Google and vendor (Claude/GPT) have independent - # rate-limit windows (separate 5-hour quota refresh cycles). - google_breaker = get_google_breaker() - vendor_breaker = get_vendor_breaker() - - # Check if ANY model path is available - if not google_breaker.is_allowed() and not vendor_breaker.is_allowed(): - logger.info( - f"agy proxy: both circuit breakers open (google tier={google_breaker.tier}, " - f"vendor tier={vendor_breaker.tier}) — skipping agy, falling through to LiteLLM" - ) - return None - # Build context-aware prompt from message history - proxy_prompt = prompt - if messages: - context_parts = [] - for msg in messages[-10:]: - role = msg.get("role", "user") - content = msg.get("content", "") - if role == "user": - context_parts.append(f"User: {content}") - elif role == "assistant": - context_parts.append(f"Assistant: {content}") - proxy_prompt = "\n".join(context_parts[-6:]) - - # Check if we have an existing session with a conversation ID - existing_conv_id = None - start_tier_index = 0 - if session_id and session_id in _session_store: - session = _session_store[session_id] - existing_conv_id = session.get("conversation_id") - start_tier_index = session.get("current_tier_index", 0) - logger.info(f"agy proxy: resuming session {session_id[:8]}..., " - f"conversation={existing_conv_id[:8]}...") - - start_time = time.time() - last_conv_id = existing_conv_id - - import httpx - - for tier_idx, tier in enumerate(agy_tiers[start_tier_index:]): - actual_tier_idx = start_tier_index + tier_idx - elapsed = time.time() - start_time - remaining = total_timeout - elapsed - if remaining <= 0: - logger.warning(f"agy proxy: total timeout exhausted at tier {tier['model_name']}") - break - - # 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 = "gemini" in tier.get("model_name", "").lower() - tier_breaker = google_breaker if is_google_tier else vendor_breaker - - if not tier_breaker.is_allowed(): - logger.info( - f"agy proxy: tier {tier['model_name']} blocked by circuit breaker " - f"(tier {tier_breaker.tier}, {tier_breaker.cooldown_until - time.time():.0f}s remaining) — skipping" - ) - continue + should_close_client = False + if client is None: + client = httpx.AsyncClient(timeout=total_timeout + 5.0) + should_close_client = True - tier_timeout = min(AGY_TIMEOUT_SECS, remaining) - - if stream: - url = "http://127.0.0.1:5005/run" - payload = { - "prompt": proxy_prompt, - "model_override": tier["env_override"], - "conversation_id": last_conv_id if actual_tier_idx > 0 or existing_conv_id else None, - "timeout": tier_timeout, - "stream": True - } - - model_tag = tier["env_override"] if tier["env_override"] else "default (gemini-3.5-flash)" - logger.info(f"agy proxy connecting stream to daemon: [{model_tag}]...") - - client = httpx.AsyncClient(timeout=tier_timeout + 5.0) - req = client.build_request("POST", url, json=payload) + stream_returned = False + try: + if cooldown_persistence is not None: try: - r = await client.send(req, stream=True) + await cooldown_persistence.sync() except Exception as e: - logger.error(f"Failed to connect stream to daemon: {e}") - await client.aclose() - continue - - # Read first line to see if it's successful or quota error - first_line = None - try: - lines_iter = r.aiter_lines() - first_line = await anext(lines_iter) - except (StopAsyncIteration, Exception): - pass - - if not first_line: - await r.aclose() - await client.aclose() - logger.warning(f"agy proxy: tier {tier['model_name']} returned empty stream. Trying next tier...") + logger.warning(f"Failed to sync state from Valkey: {e}") + + # Per-model circuit breakers — Google and vendor (Claude/GPT) have independent + # rate-limit windows (separate 5-hour quota refresh cycles). + google_breaker = get_google_breaker() + vendor_breaker = get_vendor_breaker() + + # Check if ANY model path is available + if not google_breaker.is_allowed() and not vendor_breaker.is_allowed(): + logger.info( + f"agy proxy: both circuit breakers open (google tier={google_breaker.tier}, " + f"vendor tier={vendor_breaker.tier}) — skipping agy, falling through to LiteLLM" + ) + return None + + # Build context-aware prompt from message history + proxy_prompt = prompt + if messages: + context_parts = [] + for msg in messages[-10:]: + if not isinstance(msg, dict): + continue + role = msg.get("role", "user") + content = msg.get("content") or "" + if isinstance(content, list): + content = "".join(block.get("text") or "" for block in content if isinstance(block, dict) and block.get("type") == "text") + if role == "user": + context_parts.append(f"User: {content}") + elif role == "assistant": + context_parts.append(f"Assistant: {content}") + proxy_prompt = "\n".join(context_parts[-6:]) + + # Check if we have an existing session with a conversation ID + existing_conv_id = None + start_tier_index = 0 + if session_id and session_id in _session_store: + session = _session_store[session_id] + existing_conv_id = session.get("conversation_id") + start_tier_index = session.get("current_tier_index", 0) + logger.info(f"agy proxy: resuming session {session_id[:8]}..., " + f"conversation={existing_conv_id[:8]}...") + + start_time = time.time() + last_conv_id = existing_conv_id + + for tier_idx, tier in enumerate(agy_tiers[start_tier_index:]): + actual_tier_idx = start_tier_index + tier_idx + elapsed = time.time() - start_time + remaining = total_timeout - elapsed + if remaining <= 0: + logger.warning(f"agy proxy: total timeout exhausted at tier {tier['model_name']}") + break + + # 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 = "gemini" in tier.get("model_name", "").lower() + tier_breaker = google_breaker if is_google_tier else vendor_breaker + + if not tier_breaker.is_allowed(): + logger.info( + f"agy proxy: tier {tier['model_name']} blocked by circuit breaker " + f"(tier {tier_breaker.tier}, {tier_breaker.cooldown_until - time.time():.0f}s remaining) — skipping" + ) continue + + tier_timeout = min(AGY_TIMEOUT_SECS, remaining) + + if stream: + url = "http://127.0.0.1:5005/run" + payload = { + "prompt": proxy_prompt, + "model_override": tier["env_override"], + "conversation_id": last_conv_id if actual_tier_idx > 0 or existing_conv_id else None, + "timeout": tier_timeout, + "stream": True + } - try: - first_data = json.loads(first_line) - except Exception: - await r.aclose() - await client.aclose() - logger.error(f"agy proxy: invalid JSON from daemon: {first_line}") - continue + model_tag = tier["env_override"] if tier["env_override"] else "default (gemini-3.5-flash)" + logger.info(f"agy proxy connecting stream to daemon: [{model_tag}]...") - # Check if first message is a status failure - if first_data.get("type") == "status": - rc = first_data.get("returncode", 0) - stderr_content = first_data.get("stderr", "") - if _is_quota_exhausted(rc, "", stderr_content) or rc != 0: - if _is_quota_exhausted(rc, "", stderr_content): - tier_breaker.record_failure() + req = client.build_request("POST", url, json=payload, timeout=tier_timeout + 5.0) + try: + r = await client.send(req, stream=True) + except Exception as e: + logger.error(f"Failed to connect stream to daemon: {e}") + continue + + # Read first line to see if it's successful or quota error + first_line = None + try: + lines_iter = r.aiter_lines() + first_line = await anext(lines_iter) + except (StopAsyncIteration, Exception): + pass + + if not first_line: await r.aclose() - await client.aclose() - logger.warning(f"agy proxy: tier {tier['model_name']} failed immediately (rc={rc}). Trying next tier...") + logger.warning(f"agy proxy: tier {tier['model_name']} returned empty stream. Trying next tier...") continue - # Success! Stream has started. - tier_breaker.record_success() - 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"): - yield init_data["content"] - elif init_data.get("type") == "conversation_id" and init_data.get("id"): - current_conv_id = init_data["id"] - if session_id: - _session_store[session_id] = { - "conversation_id": current_conv_id, - "current_tier_index": actual_tier_idx, - } - try: - async for line in lines_iter: - if not line.strip(): - continue - data = json.loads(line) - if data.get("type") == "token" and data.get("content"): - yield data["content"] - elif data.get("type") == "conversation_id" and data.get("id"): - current_conv_id = data["id"] - if session_id: - _session_store[session_id] = { - "conversation_id": current_conv_id, - "current_tier_index": actual_tier_idx, - } - finally: - await stream_resp.aclose() - await httpx_client.aclose() + first_data = json.loads(first_line) + except Exception: + await r.aclose() + logger.error(f"agy proxy: invalid JSON from daemon: {first_line}") + continue - return { - "stream": token_generator(r, client, first_line, last_conv_id), - "model": tier["model_name"] - } - - else: - # Non-streaming path - returncode, stdout, stderr, result_conv_id = await _run_agy_print( - proxy_prompt, - model_override=tier["env_override"], - conversation_id=last_conv_id if actual_tier_idx > 0 or existing_conv_id else None, - timeout=tier_timeout, - ) - - # Update the conversation ID from the result - if result_conv_id: - last_conv_id = result_conv_id - - # Check for quota exhaustion - if _is_quota_exhausted(returncode, stdout, stderr): - tier_breaker.record_failure() - logger.warning( - f"agy proxy: tier {tier['model_name']} quota exhausted. " - f"Falling to tier {actual_tier_idx + 2}..." - ) - continue - - # Check for other errors - if returncode != 0: - logger.warning( - f"agy proxy: tier {tier['model_name']} failed " - f"(rc={returncode}, stderr={stderr[:200]}). " - f"Falling to next tier..." - ) - continue - - # Success! - if stdout: - elapsed_total = time.time() - start_time + # Check if first message is a status failure + if first_data.get("type") == "status": + rc = first_data.get("returncode", 0) + stderr_content = first_data.get("stderr", "") + if _is_quota_exhausted(rc, "", stderr_content) or rc != 0: + if _is_quota_exhausted(rc, "", stderr_content): + tier_breaker.record_failure() + if cooldown_persistence is not None: + try: + await cooldown_persistence.save() + except Exception as e: + logger.warning(f"Failed to save cooldowns to Valkey: {e}") + await r.aclose() + logger.warning(f"agy proxy: tier {tier['model_name']} failed immediately (rc={rc}). Trying next tier...") + continue + + # Success! Stream has started. + tier_breaker.record_success() + if cooldown_persistence is not None: + try: + await cooldown_persistence.save() + except Exception as e: + logger.warning(f"Failed to save cooldowns to Valkey: {e}") - # Save session state for continuation - if session_id: - _session_store[session_id] = { - "conversation_id": last_conv_id, - "current_tier_index": actual_tier_idx, - } - logger.info(f"agy proxy: saved session {session_id[:8]}..." - f" → conversation={last_conv_id[:8]}..., tier={tier['model_name']}") + async def token_generator(stream_resp, httpx_client, initial_line, current_conv_id, close_client): + """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"): + yield init_data["content"] + elif init_data.get("type") == "conversation_id" and init_data.get("id"): + current_conv_id = init_data["id"] + if session_id: + _session_store[session_id] = { + "conversation_id": current_conv_id, + "current_tier_index": actual_tier_idx, + } + + try: + async for line in lines_iter: + if not line.strip(): + continue + data = json.loads(line) + if data.get("type") == "token" and data.get("content"): + yield data["content"] + elif data.get("type") == "conversation_id" and data.get("id"): + current_conv_id = data["id"] + if session_id: + _session_store[session_id] = { + "conversation_id": current_conv_id, + "current_tier_index": actual_tier_idx, + } + finally: + await stream_resp.aclose() + if close_client: + await httpx_client.aclose() + + stream_returned = True + return { + "stream": token_generator(r, client, first_line, last_conv_id, should_close_client), + "model": tier["model_name"] + } - logger.info( - f"agy proxy: ✅ tier {tier['model_name']} succeeded " - f"({len(stdout)} chars, {elapsed_total:.1f}s)" - ) - tier_breaker.record_success() - return _wrap_response(stdout, tier["model_name"], proxy_prompt) else: - logger.warning( - f"agy proxy: tier {tier['model_name']} returned empty response" + # Non-streaming path + returncode, stdout, stderr, result_conv_id = await _run_agy_print( + client, + proxy_prompt, + model_override=tier["env_override"], + conversation_id=last_conv_id if actual_tier_idx > 0 or existing_conv_id else None, + timeout=tier_timeout, ) - continue - # All tiers exhausted — clean up session - if session_id and session_id in _session_store: - del _session_store[session_id] - - logger.warning("agy proxy: all tiers exhausted — falling back to LiteLLM") - return None \ No newline at end of file + # Update the conversation ID from the result + if result_conv_id: + last_conv_id = result_conv_id + + # Check for quota exhaustion + if _is_quota_exhausted(returncode, stdout, stderr): + tier_breaker.record_failure() + if cooldown_persistence is not None: + try: + await cooldown_persistence.save() + except Exception as e: + logger.warning(f"Failed to save cooldowns to Valkey: {e}") + logger.warning( + f"agy proxy: tier {tier['model_name']} quota exhausted. " + f"Falling to tier {actual_tier_idx + 2}..." + ) + continue + + # Check for other errors + if returncode != 0: + logger.warning( + f"agy proxy: tier {tier['model_name']} failed " + f"(rc={returncode}, stderr={stderr[:200]}). " + f"Falling to next tier..." + ) + continue + + # Success! + if stdout: + elapsed_total = time.time() - start_time + + # Save session state for continuation + if session_id: + _session_store[session_id] = { + "conversation_id": last_conv_id, + "current_tier_index": actual_tier_idx, + } + logger.info(f"agy proxy: saved session {session_id[:8]}..." + f" → conversation={last_conv_id[:8]}..., tier={tier['model_name']}") + + logger.info( + f"agy proxy: ✅ tier {tier['model_name']} succeeded " + f"({len(stdout)} chars, {elapsed_total:.1f}s)" + ) + tier_breaker.record_success() + if cooldown_persistence is not None: + try: + await cooldown_persistence.save() + except Exception as e: + logger.warning(f"Failed to save cooldowns to Valkey: {e}") + return _wrap_response(stdout, tier["model_name"], proxy_prompt) + else: + logger.warning( + f"agy proxy: tier {tier['model_name']} returned empty response" + ) + continue + + # All tiers exhausted — clean up session + if session_id and session_id in _session_store: + del _session_store[session_id] + + logger.warning("agy proxy: all tiers exhausted — falling back to LiteLLM") + return None + finally: + if should_close_client and not stream_returned: + await client.aclose() \ No newline at end of file diff --git a/router/circuit_breaker.py b/router/circuit_breaker.py index df17fcd2..1b7c42a0 100644 --- a/router/circuit_breaker.py +++ b/router/circuit_breaker.py @@ -121,6 +121,42 @@ def status(self) -> dict: "probe_granted": self.probe_granted, } + async def sync_from_valkey(self, redis_client) -> None: + """Synchronize circuit breaker state from Valkey.""" + if not redis_client: + return + try: + state = await redis_client.hgetall(f"circuit_breaker:{self.name}") + if state: + self.tier = int(state.get("tier", "0")) + self.cooldown_until = float(state.get("cooldown_until", "0.0")) + self.probe_granted = state.get("probe_granted", "False") == "True" + self.total_trips = int(state.get("total_trips", "0")) + self.last_trip_time = float(state.get("last_trip_time", "0.0")) + except Exception as e: + logger.warning(f"Valkey circuit_breaker [{self.name}] sync failed: {e}") + + async def save_to_valkey(self, redis_client) -> None: + """Persist circuit breaker state to Valkey.""" + if not redis_client: + return + try: + key = f"circuit_breaker:{self.name}" + state = { + "tier": str(self.tier), + "cooldown_until": str(self.cooldown_until), + "probe_granted": "True" if self.probe_granted else "False", + "total_trips": str(self.total_trips), + "last_trip_time": str(self.last_trip_time), + } + await redis_client.hset(key, mapping=state) + now = time.time() + ttl = int(max(3600.0, self.cooldown_until - now + 3600.0)) + await redis_client.expire(key, ttl) + except Exception as e: + logger.warning(f"Valkey circuit_breaker [{self.name}] save failed: {e}") + + class DualCircuitBreaker: """ @@ -162,6 +198,23 @@ def status(self) -> dict: "vendor": self.vendor.status(), } + async def sync_from_valkey(self, redis_client) -> None: + """Synchronize both sub-breakers from Valkey.""" + import asyncio + await asyncio.gather( + self.google.sync_from_valkey(redis_client), + self.vendor.sync_from_valkey(redis_client) + ) + + async def save_to_valkey(self, redis_client) -> None: + """Persist both sub-breakers to Valkey.""" + import asyncio + await asyncio.gather( + self.google.save_to_valkey(redis_client), + self.vendor.save_to_valkey(redis_client) + ) + + # Module-level singleton _breaker = DualCircuitBreaker() diff --git a/router/free_models_roster.json b/router/free_models_roster.json index f7cedb5b..9795cea4 100644 --- a/router/free_models_roster.json +++ b/router/free_models_roster.json @@ -42,6 +42,12 @@ "score": 28.0, "context_length": 256000 }, + { + "id": "cohere/north-mini-code:free", + "name": "Cohere: North Mini Code (free)", + "score": 25.0, + "context_length": 256000 + }, { "id": "nex-agi/nex-n2-pro:free", "name": "Nex AGI: Nex-N2-Pro (free)", @@ -139,6 +145,6 @@ "context_length": 32768 } ], - "updated_at": "2026-06-17T20:17:56.981195Z", - "count": 23 + "updated_at": "2026-06-20T10:30:31.109687Z", + "count": 24 } \ No newline at end of file diff --git a/router/main.py b/router/main.py index 83c4127c..71de563b 100644 --- a/router/main.py +++ b/router/main.py @@ -9,6 +9,7 @@ import tempfile import yaml import httpx +import redis.asyncio as aioredis from contextlib import asynccontextmanager from fastapi import FastAPI, Request, HTTPException, Response from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse @@ -18,6 +19,122 @@ from pydantic import BaseModel from typing import Dict, Optional, Union +_redis_client = None +_redis_last_init_attempt = 0.0 +_REDIS_RETRY_INTERVAL_SECONDS = 5.0 + +def get_redis(): + """Lazily initialize and return the async Redis/Valkey client. + Returns None if connection fails or is disabled (non-fatal fallback).""" + global _redis_client, _redis_last_init_attempt + if _redis_client is None: + now = time.monotonic() + if now - _redis_last_init_attempt < _REDIS_RETRY_INTERVAL_SECONDS: + return None + _redis_last_init_attempt = now + try: + host = os.getenv("VALKEY_HOST", "127.0.0.1") + port = int(os.getenv("VALKEY_PORT", "6379")) + _redis_client = aioredis.Redis(host=host, port=port, decode_responses=True, socket_timeout=1.0) + logger.info(f"Valkey client initialized at {host}:{port}") + except Exception as e: + logger.warning(f"Failed to initialize Valkey client: {e} — falling back to local memory") + _redis_client = None + return _redis_client + + +_http_client = None + +def get_http_client(): + """Return the shared global httpx.AsyncClient singleton.""" + global _http_client + if _http_client is None: + _http_client = httpx.AsyncClient(timeout=3600.0) + return _http_client + + +def estimate_prompt_tokens(body: dict) -> int: + """Estimate prompt tokens by counting characters in message contents (1 token ~= 4 chars) + to avoid inflating metrics with large tool/schema declarations. + """ + tokens = 0 + for msg in body.get("messages", []): + if not isinstance(msg, dict): + continue + content = msg.get("content") or "" + if isinstance(content, str): + tokens += len(content) // 4 + elif isinstance(content, list): + for block in content: + if isinstance(block, dict) and block.get("type") == "text": + tokens += len(block.get("text") or "") // 4 + # Include a flat estimate for system prompt / metadata overhead + tokens += 50 + return max(1, tokens) + + +async def sync_cooldowns_from_valkey() -> None: + """Sync Ollama cooldown and circuit breaker states from Valkey to local memory.""" + redis = get_redis() + if not redis: + return + try: + val = await redis.get("cooldown:ollama") + global _ollama_cooldown_until + if val is not None: + epoch_until = float(val) + remaining = epoch_until - time.time() + if remaining > 0: + _ollama_cooldown_until = time.monotonic() + remaining + else: + _ollama_cooldown_until = 0.0 + else: + _ollama_cooldown_until = 0.0 + + breaker = get_breaker() + await breaker.sync_from_valkey(redis) + except Exception as e: + logger.warning(f"Failed to sync cooldowns from Valkey: {e}") + global _redis_client, _redis_last_init_attempt + _redis_client = None + _redis_last_init_attempt = time.monotonic() + + +async def save_cooldowns_to_valkey() -> None: + """Save local Ollama cooldown and circuit breaker states to Valkey.""" + redis = get_redis() + if not redis: + return + try: + global _ollama_cooldown_until + now_mono = time.monotonic() + if _ollama_cooldown_until > now_mono: + remaining = _ollama_cooldown_until - now_mono + epoch_until = time.time() + remaining + ttl = int(max(1.0, remaining)) + await redis.set("cooldown:ollama", str(epoch_until), ex=ttl) + else: + await redis.delete("cooldown:ollama") + + breaker = get_breaker() + await breaker.save_to_valkey(redis) + except Exception as e: + logger.warning(f"Failed to save cooldowns to Valkey: {e}") + global _redis_client, _redis_last_init_attempt + _redis_client = None + _redis_last_init_attempt = time.monotonic() + + +class ValkeyCooldownPersistence: + """Persistence provider mapping Valkey/Redis client synchronization to the global handlers.""" + async def sync(self) -> None: + await sync_cooldowns_from_valkey() + + async def save(self) -> None: + await save_cooldowns_to_valkey() + + + # Configure logging — respect LOG_LEVEL env var (default: WARNING) _log_level_str = os.getenv("LOG_LEVEL", "WARNING").upper() _log_level = getattr(logging, _log_level_str, logging.WARNING) @@ -130,6 +247,21 @@ async def push_aggregate_scores(): "timeline": [] } +# --------------------------------------------------------------------------- +# OLLAMA COOLDOWN — router-side cooldown for the Ollama backend. +# LiteLLM Community Edition's deployment cooldown is unreliable for single- +# deployment model groups (it bypasses cooldown when there's only 1 deployment) +# and doesn't reliably cooldown fallback-target model groups. Instead, the +# triage router tracks Ollama failures itself and returns 429 immediately +# during the cooldown window, skipping the LiteLLM call entirely. +# --------------------------------------------------------------------------- +_ollama_cooldown_until: float = 0.0 # monotonic timestamp when cooldown expires +try: + OLLAMA_COOLDOWN_SECONDS: int = int(os.getenv("OLLAMA_COOLDOWN_SECONDS", "300")) # 5 min default +except (TypeError, ValueError): + logger.warning("Invalid OLLAMA_COOLDOWN_SECONDS value; defaulting to 300") + OLLAMA_COOLDOWN_SECONDS = 300 + STATS_JSON_PATH = "/config/router_dir/router_stats.json" # Module-level set to hold references to fire-and-forget background tasks, @@ -225,6 +357,18 @@ async def save_persisted_stats(force=False): CACHE_TTL_SECONDS = 86400 # Decisions cached for 24 hours classification_lock = asyncio.Lock() +async def _purge_stale_deployments(db_url: str, pattern: str): + """Purge stale deployments matching the pattern from LiteLLM's DB.""" + import asyncpg + conn = await asyncpg.connect(db_url) + try: + await conn.execute( + 'DELETE FROM "LiteLLM_ProxyModelTable" WHERE model_name LIKE $1', + pattern + ) + finally: + await conn.close() + async def sync_adaptive_router_roster(master_key: str): """Fetch free OpenRouter models and register them as deployments in LiteLLM.""" if not master_key: @@ -243,6 +387,8 @@ async def sync_adaptive_router_roster(master_key: str): logger.warning(f"Failed to fetch OpenRouter models: {e}") return free_models = [] + model_contexts = {} + model_supported_params = {} for m in all_models: mid = m.get("id", "") # Skip internal OpenRouter encoded IDs that LiteLLM can't map to a provider @@ -272,6 +418,8 @@ async def sync_adaptive_router_roster(master_key: str): except Exception: score = 25.0 # conservative fallback for unparseable models free_models.append((score, mid)) + model_contexts[mid] = m.get("context_length") or 262144 + model_supported_params[mid] = supported_params free_models.sort(reverse=True) if not free_models: logger.warning("No free models found — skipping roster sync") @@ -332,14 +480,8 @@ def norm(s: float) -> float: # in 24h), bloating the DB and slowing LiteLLM startup. Each sync now # starts clean — delete all, then register only the current roster. try: - db_url = os.getenv("DATABASE_URL", "postgresql://postgres@127.0.0.1:5432/postgres") - import asyncpg - conn = await asyncpg.connect(db_url) - await conn.execute( - 'DELETE FROM "LiteLLM_ProxyModelTable" WHERE model_name LIKE $1', - 'agent-%' - ) - await conn.close() + db_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres-local-pw-2026@127.0.0.1:5432/postgres") + await _purge_stale_deployments(db_url, 'agent-%') logger.info("🧹 Purged stale agent-* deployments before roster sync") except Exception as e: logger.warning(f"Failed to purge stale deployments (non-fatal): {e}") @@ -348,9 +490,20 @@ def norm(s: float) -> float: failed = 0 for tier_name, model_ids in tier_assignments.items(): for mid in model_ids: + ctx_len = model_contexts.get(mid, 262144) + sp = model_supported_params.get(mid, []) payload = { "model_name": tier_name, - "litellm_params": {"model": f"openrouter/{mid}", "request_timeout": 120} + "litellm_params": {"model": f"openrouter/{mid}", "request_timeout": 20}, + "model_info": { + "supports_vision": "vision" in sp, + "supports_reasoning": True, # OpenRouter API has no "reasoning" param; assume all modern LLMs support it + "supports_function_calling": "tools" in sp, + "mode": "chat", + "max_tokens": ctx_len, + "max_input_tokens": ctx_len, + "is_public_model_group": True + } } try: r = await client.post(f"{admin_url}/model/new", headers=headers, json=payload) @@ -364,9 +517,128 @@ def norm(s: float) -> float: logger.warning(f"Failed to register {mid} under {tier_name}: {e}") logger.info(f"📊 Roster sync: registered {registered} deployments ({failed} failed) across 5 tiers — {sum(len(v) for v in tier_assignments.values())} attempted") +async def _register_ollama_models_in_db(master_key: str): + """Register static ollama models via /model/new so they become DB models. + + LiteLLM's /model_group/info endpoint aggregates model info using its internal + model cost map for known providers. For ollama_chat models not in the map, + capabilities (vision, reasoning, function_calling) and token limits come back + as null/false. Registering them as DB models ensures our model_info wins. + """ + if not master_key: + logger.warning("No LiteLLM master key provided — skipping Ollama DB registration") + return + + admin_url = os.getenv("LITELLM_ADMIN_URL", "http://127.0.0.1:4000") + headers = {"Authorization": f"Bearer {master_key}", "Content-Type": "application/json"} + + ollama_models = [] + litellm_config_path = os.getenv("LITELLM_CONFIG_PATH", "/config/litellm_dir/config.yaml") + + config_paths_to_try = [ + litellm_config_path, + str(Path(__file__).resolve().parent.parent / "litellm" / "config.yaml"), + "./litellm/config.yaml" + ] + + loaded_from_config = False + for path in config_paths_to_try: + if path and os.path.exists(path): + try: + with open(path, "r") as f: + litellm_config = yaml.safe_load(f) + if isinstance(litellm_config, dict) and isinstance(litellm_config.get("model_list"), list): + for item in litellm_config["model_list"]: + if isinstance(item, dict): + model_name = item.get("model_name", "") + if isinstance(model_name, str) and model_name.startswith("ollama-deepseek-"): + # Create a clean deep copy to avoid mutating configuration structures + ollama_models.append(copy.deepcopy(item)) + if ollama_models: + logger.info(f"Loaded {len(ollama_models)} Ollama model configurations dynamically from {path}") + loaded_from_config = True + break + except Exception as e: + logger.warning(f"Failed to load/parse LiteLLM config at {path}: {e}") + + if not loaded_from_config: + logger.warning("Could not load Ollama models from config.yaml, falling back to static definitions") + ollama_models = [ + { + "model_name": "ollama-deepseek-v4-pro", + "litellm_params": { + "model": "ollama_chat/deepseek-v4-pro", + "api_base": "https://api.ollama.com", + "api_key": "os.environ/OLLAMA_API_KEY", + "request_timeout": 120, + }, + "model_info": { + "supports_vision": True, + "supports_reasoning": True, + "supports_function_calling": True, + "mode": "chat", + "max_tokens": 524288, + "max_input_tokens": 524288, + "input_cost_per_token": 0.00000174, + "output_cost_per_token": 0.00000348, + "is_public_model_group": True, + }, + }, + { + "model_name": "ollama-deepseek-v4-flash", + "litellm_params": { + "model": "ollama_chat/deepseek-v4-flash", + "api_base": "https://api.ollama.com", + "api_key": "os.environ/OLLAMA_API_KEY", + "request_timeout": 120, + }, + "model_info": { + "supports_vision": True, + "supports_reasoning": True, + "supports_function_calling": True, + "mode": "chat", + "max_tokens": 524288, + "max_input_tokens": 524288, + "input_cost_per_token": 0.00000014, + "output_cost_per_token": 0.00000028, + "is_public_model_group": True, + }, + }, + ] + + # Purge stale ollama-deepseek DB entries before re-registering. + # Mirrors the agent-* purge pattern above — delete all, then register fresh. + try: + db_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres-local-pw-2026@127.0.0.1:5432/postgres") + await _purge_stale_deployments(db_url, 'ollama-deepseek-%') + logger.info("🧹 Purged stale ollama-deepseek-* DB entries before registration") + except Exception as e: + logger.warning(f"Failed to purge stale ollama DB entries (non-fatal): {e}") + + async with httpx.AsyncClient(timeout=10.0) as client: + registered = 0 + failed = 0 + for payload in ollama_models: + try: + r = await client.post(f"{admin_url}/model/new", headers=headers, json=payload) + if r.status_code in (200, 201): + registered += 1 + else: + failed += 1 + logger.warning(f"model/new {payload['model_name']}: HTTP {r.status_code} — {r.text[:200]}") + except Exception as e: + failed += 1 + logger.warning(f"Failed to register {payload['model_name']}: {e}") + logger.info(f"📊 Ollama DB registration: {registered} registered, {failed} failed") + + @asynccontextmanager async def lifespan(app: FastAPI): """Startup: wait for LiteLLM readiness, then sync free-model roster.""" + # Initialize shared HTTPX client and sync cooldowns from Redis/Valkey + get_http_client() + await sync_cooldowns_from_valkey() + litellm_ready_url = "http://127.0.0.1:4000/health/readiness" litellm_master_key = os.getenv("LITELLM_MASTER_KEY", "") max_wait = 180 @@ -391,6 +663,15 @@ async def lifespan(app: FastAPI): except Exception as e: logger.error(f"Roster sync failed: {e}") + # Register static ollama models via /model/new so they become DB models. + # The ollama_chat provider's internal model lookup overrides static config + # model_info at the group aggregation level (/model_group/info), causing + # features and token limits to show as null/false. DB models get priority. + try: + await _register_ollama_models_in_db(litellm_master_key) + except Exception as e: + logger.warning(f"Ollama DB registration failed (non-fatal): {e}") + # Start background task before yield so it runs during app lifetime task = asyncio.create_task(push_aggregate_scores()) @@ -404,6 +685,16 @@ async def lifespan(app: FastAPI): except asyncio.CancelledError: pass + # Close shared HTTPX client + global _http_client + if _http_client is not None: + await _http_client.aclose() + + # Close Redis client + global _redis_client + if _redis_client is not None and _redis_client is not False: + await _redis_client.aclose() + # Flush any buffered stats/timeline on clean shutdown (always runs) await save_persisted_stats(force=True) try: @@ -633,6 +924,8 @@ def detect_active_tool(body: dict) -> str: for idx in range(len(messages) - 1, -1, -1): msg = messages[idx] + if not isinstance(msg, dict): + continue role = msg.get("role") if role in ("tool", "function"): name = msg.get("name") @@ -641,11 +934,15 @@ def detect_active_tool(body: dict) -> str: tool_call_id = msg.get("tool_call_id") if tool_call_id: for prev_msg in reversed(messages[:idx]): + if not isinstance(prev_msg, dict): + continue if prev_msg.get("role") == "assistant": tcalls = prev_msg.get("tool_calls") or [] for tc in tcalls: - if tc.get("id") == tool_call_id: - name = tc.get("function", {}).get("name") + if isinstance(tc, dict) and tc.get("id") == tool_call_id: + fn = tc.get("function") + if isinstance(fn, dict): + name = fn.get("name") break if name: break @@ -656,11 +953,15 @@ def detect_active_tool(body: dict) -> str: tool_calls = msg.get("tool_calls") if tool_calls and isinstance(tool_calls, list): for tc in tool_calls: - name = tc.get("function", {}).get("name") or "other" - return map_tool_to_category(name) + if isinstance(tc, dict): + fn = tc.get("function") + name = (fn.get("name") if isinstance(fn, dict) else None) or "other" + return map_tool_to_category(name) # Fallback to keyphrase scanning in the user message for msg in reversed(messages): + if not isinstance(msg, dict): + continue if msg.get("role") == "user": content = str(msg.get("content", "")).lower() if "tree" in content or "files" in content: @@ -1051,15 +1352,17 @@ async def proxy_models(): data = r.json() # Inject llm-routing-* models at the top of the list. # Auto models (classifier pipeline) first, then direct models. - # context_length: 262144 (256K) — the downstream models (gemini-3.5-flash, - # claude-opus-4.6, deepseek-v4-pro/flash) all support 256K+ context. + # Context lengths aligned with downstream model targets: + # - auto-free / auto-agy: 262144 (262K) + # - auto-ollama / auto-agy-ollama / llm-routing-ollama: 524288 (512K) + # - llm-routing-agy: 1048576 (1M) routing_models = [ {"id": "llm-routing-auto-free", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 262144}, {"id": "llm-routing-auto-agy", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 262144}, - {"id": "llm-routing-auto-ollama", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 262144}, - {"id": "llm-routing-auto-agy-ollama", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 262144}, - {"id": "llm-routing-agy", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 262144}, - {"id": "llm-routing-ollama", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 262144}, + {"id": "llm-routing-auto-ollama", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 524288}, + {"id": "llm-routing-auto-agy-ollama", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 524288}, + {"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}, ] for entry in reversed(routing_models): data["data"].insert(0, entry) @@ -1079,6 +1382,8 @@ async def chat_completions(request: Request): except Exception: raise HTTPException(status_code=400, detail="Invalid JSON payload") + await sync_cooldowns_from_valkey() + messages = body.get("messages", []) if not messages: raise HTTPException(status_code=400, detail="Empty messages list") @@ -1089,8 +1394,13 @@ async def chat_completions(request: Request): # Extract last user message for complexity triage last_user_message = "" for msg in reversed(messages): + if not isinstance(msg, dict): + continue if msg.get("role") == "user": - last_user_message = msg.get("content", "") + content = msg.get("content") or "" + if isinstance(content, list): + content = "".join(block.get("text") or "" for block in content if isinstance(block, dict) and block.get("type") == "text") + last_user_message = content break # Known tier names that can be routed directly (bypass classifier) @@ -1098,7 +1408,7 @@ async def chat_completions(request: Request): "agent-simple-core", "agent-medium-core", "agent-complex-core", "agent-reasoning-core", "agent-advanced-core", - "llm-routing-agy", "llm-routing-ollama", + "llm-routing-agy", } AUTO_MODELS = { @@ -1126,13 +1436,13 @@ async def chat_completions(request: Request): langfuse_trace_id = None parent_obs = None - if client_model in AUTO_MODELS: + if client_model in AUTO_MODELS or client_model == "llm-routing-ollama": # Full pipeline: classify → route to best tier bypass_cache = request.headers.get("x-bypass-cache") == "true" target_model, triage_latency, was_cache_hit, raw_classification = await classify_request( last_user_message, bypass_cache=bypass_cache, langfuse_trace_id=langfuse_trace_id ) - logger.info(f"Triage decision (auto): Routing to -> '{target_model}'") + logger.info(f"Triage decision (auto/gated): Routing to -> '{target_model}'") elif client_model in DIRECT_TIERS: # Direct routing: client knows what tier they want, skip classifier target_model = client_model @@ -1202,19 +1512,25 @@ async def chat_completions(request: Request): or (client_model in ("llm-routing-auto-agy", "llm-routing-auto-agy-ollama") and target_model in ("agent-advanced-core", "agent-reasoning-core")) ) should_try_ollama = ( - client_model == "llm-routing-ollama" # direct — always try - or (client_model in ("llm-routing-auto-ollama", "llm-routing-auto-agy-ollama") and target_model in ("agent-advanced-core", "agent-reasoning-core")) + client_model == "llm-routing-ollama" # always try (will map to flash for complex/below) + or (client_model in ("llm-routing-auto-ollama", "llm-routing-auto-agy-ollama") and target_model in ("agent-advanced-core", "agent-reasoning-core", "agent-complex-core")) ) # --- AGY PROXY --- if should_try_agy: + agy_span_obj = None try: from agy_proxy import try_agy_proxy last_prompt = "" for msg in reversed(messages): + if not isinstance(msg, dict): + continue if msg.get("role") == "user": - last_prompt = msg.get("content", "") + content = msg.get("content") or "" + if isinstance(content, list): + content = "".join(block.get("text") or "" for block in content if isinstance(block, dict) and block.get("type") == "text") + last_prompt = content break session_id = None @@ -1222,15 +1538,18 @@ async def chat_completions(request: Request): import hashlib fingerprint_parts = [] for msg in messages[:4]: - c = msg.get("content", "") or "" - if c: + if not isinstance(msg, dict): + continue + c = msg.get("content") or "" + if isinstance(c, list): + c = "".join(block.get("text") or "" for block in c if isinstance(block, dict) and block.get("type") == "text") + if isinstance(c, str) and c: fingerprint_parts.append(c[:200]) fingerprint = "|".join(fingerprint_parts) session_id = hashlib.md5(fingerprint.encode()).hexdigest() if last_prompt: # --- Langfuse child span: agy proxy --- - agy_span_obj = None if langfuse_trace_id: lf_agy = get_langfuse() if lf_agy: @@ -1252,7 +1571,9 @@ async def chat_completions(request: Request): session_id=session_id, total_timeout=300.0, stream=is_stream_requested, - target_tier=target_model + target_tier=target_model, + client=get_http_client(), + cooldown_persistence=ValkeyCooldownPersistence() ) if agy_response: model_name = agy_response.get("model", "gemini-3.5-flash (via agy)") @@ -1300,9 +1621,7 @@ async def native_agy_stream_generator(stream_gen, model_name): # 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) + approx_prompt_tokens = estimate_prompt_tokens(body) record_tool_usage( active_tool, approx_prompt_tokens, token_count, @@ -1352,7 +1671,7 @@ async def native_agy_stream_generator(stream_gen, model_name): 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", "") + content = (agy_response.get("choices") or [{}])[0].get("message", {}).get("content") or "" async def agy_stream_generator(): """Asynchronous generator yielding simulated OpenAI-compatible streaming chunks from a static agy response.""" import uuid @@ -1412,173 +1731,266 @@ async def agy_stream_generator(): pass logger.error(f"agy proxy failed: {e}, falling back to LiteLLM") + original_target_model = target_model + # --- OLLAMA (via LiteLLM) --- # LiteLLM's ollama_chat provider handles the native Ollama API call. # We just proxy to LiteLLM with the appropriate model name. - # Reasoning tier → deepseek-v4-flash (lighter, faster) - # Advanced tier → deepseek-v4-pro (full power) # LiteLLM's fallback chain handles failures. if should_try_ollama: - if target_model == "agent-reasoning-core": - target_model = "ollama-deepseek-v4-flash" + if client_model in ("llm-routing-auto-ollama", "llm-routing-auto-agy-ollama"): + if target_model in ("agent-advanced-core", "agent-reasoning-core"): + target_model = "ollama-deepseek-v4-pro" + elif target_model == "agent-complex-core": + target_model = "ollama-deepseek-v4-flash" + elif client_model == "llm-routing-ollama": + if target_model in ("agent-advanced-core", "agent-reasoning-core"): + target_model = "ollama-deepseek-v4-pro" + else: + target_model = "ollama-deepseek-v4-flash" else: - target_model = "ollama-deepseek-v4-pro" + # Fallback (e.g. if LiteLLM fallback loops back with model: llm-routing-ollama) + if target_model in ("agent-advanced-core", "agent-reasoning-core"): + target_model = "ollama-deepseek-v4-pro" + else: + target_model = "ollama-deepseek-v4-flash" logger.info(f"Ollama route: proxying to LiteLLM as model={target_model}") - # Resolve backend connection parameters - backend_conf = backends.get(target_model) - if not backend_conf: - logger.error(f"Backend '{target_model}' not found in configuration backends.") - raise HTTPException(status_code=500, detail=f"Backend {target_model} misconfigured") - - backend_api_base = backend_conf["api_base"] - backend_api_key = backend_conf["api_key"] - if backend_api_key == "DYNAMIC_LITELLM_MASTER_KEY_PLACEHOLDER": - backend_api_key = os.getenv("LITELLM_MASTER_KEY", backend_api_key) - - # Delegate to LiteLLM which handles internal fallback chain - # Router sends model=agent-complex-core (or agent-simple-core) - # LiteLLM maps this to Nemotron → Kimi → GPT-OSS → local Qwen - logger.info(f"Proxying to LiteLLM as model={target_model}") - - # --- Langfuse child span: LiteLLM proxy --- - litellm_span_obj = None - if langfuse_trace_id: - lf_litellm = get_langfuse() - if lf_litellm: + async def execute_proxy(model_name: str): + # Resolve backend connection parameters + backend_conf = backends.get(model_name) + if not backend_conf: + logger.error(f"Backend '{model_name}' not found in configuration backends.") + raise HTTPException(status_code=500, detail=f"Backend {model_name} misconfigured") + + backend_api_base = backend_conf["api_base"] + backend_api_key = backend_conf["api_key"] + if backend_api_key == "DYNAMIC_LITELLM_MASTER_KEY_PLACEHOLDER": + backend_api_key = os.getenv("LITELLM_MASTER_KEY", backend_api_key) + + logger.info(f"Proxying to LiteLLM as model={model_name}") + + # --- Langfuse child span: LiteLLM proxy --- + litellm_span_obj = None + if langfuse_trace_id: + lf_litellm = get_langfuse() + if lf_litellm: + try: + litellm_span_obj = lf_litellm.start_observation( + trace_context={"trace_id": langfuse_trace_id}, + name="litellm-proxy", + input=model_name, + metadata={"model": model_name}, + level="DEFAULT", + ) + except Exception: + pass + + client = get_http_client() + try: + headers = {"Authorization": f"Bearer {backend_api_key}"} + if langfuse_trace_id: + headers["X-Langfuse-Trace-Id"] = langfuse_trace_id + + # Handle streaming vs non-streaming proxying (LiteLLM handles fallback internally) + proxy_start = time.time() + + # --- Pre-screening: clamp max_tokens to fit within downstream model context limits --- try: - litellm_span_obj = lf_litellm.start_observation( - trace_context={"trace_id": langfuse_trace_id}, - name="litellm-proxy", - input=target_model, - metadata={"model": target_model}, - level="DEFAULT", - ) - except Exception: - pass + body_to_send = body.copy() + body_to_send["model"] = model_name + requested_max_tokens = body_to_send.get("max_tokens", 4096) + + # Tier-aware minimum context length (from actual roster data): + # - agent-simple-core: 32K (includes tiny liquid/dolphin models) + # - agent-medium-core+: 256K (smallest non-tiny model is nemotron-nano-omni at 256K) + # - ollama-deepseek-v4-*: 1M (DeepSeek V4 native context) + _tier_min_ctx = { + "agent-simple-core": 32768, + "ollama-deepseek-v4-pro": 524288, + "ollama-deepseek-v4-flash": 524288, + } + _min_ctx = _tier_min_ctx.get(model_name, 262144) + _est_input = estimate_prompt_tokens(body_to_send) + _safe_max = max(1024, _min_ctx - _est_input - 2048) # 2K safety margin + if requested_max_tokens > _safe_max: + logger.warning( + f"⛔ Clamping max_tokens: {requested_max_tokens} → {_safe_max} " + f"(est_input={_est_input}, min_ctx={_min_ctx}, tier={model_name})" + ) + body_to_send["max_tokens"] = _safe_max + except Exception as e: + logger.warning(f"Pre-screening failed (non-fatal): {e}") + body_to_send = body.copy() + body_to_send["model"] = model_name + if "metadata" not in body_to_send or not isinstance(body_to_send["metadata"], dict): + body_to_send["metadata"] = {} + body_to_send["metadata"]["trace_name"] = "agent-completion" + + if body.get("stream", False): + logger.info(f"Proxying streaming to LiteLLM as model={model_name}") + req = client.build_request("POST", f"{backend_api_base}/chat/completions", json=body_to_send, headers=headers) + 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.""" + import codecs + completion_chars = 0 + request_tokens = estimate_prompt_tokens(body_to_send) + sse_buffer = "" + decoder = codecs.getincrementaldecoder("utf-8")() + try: + async for chunk in r.aiter_bytes(): + yield chunk + try: + sse_buffer += decoder.decode(chunk) + while "\n" in sse_buffer: + line, sse_buffer = sse_buffer.split("\n", 1) + line = line.strip() + if line.startswith("data:"): + data_str = line[5:].strip() + if data_str == "[DONE]": + continue + try: + data_json = json.loads(data_str) + choices = data_json.get("choices", []) + if choices: + delta = choices[0].get("delta", {}) + content = delta.get("content") or "" + completion_chars += len(content) + except Exception: + pass + except Exception: + pass + proxy_latency = (time.time() - proxy_start) * 1000.0 + stats["total_proxy_time_ms"] += proxy_latency + stats["avg_proxy_latency_ms"] = stats["total_proxy_time_ms"] / stats["total_requests"] + record_tool_usage(active_tool, request_tokens, completion_chars // 4, model_name, proxy_latency, route="litellm_fallback") + # Finalize LiteLLM span (streaming path) + if litellm_span_obj: + try: + litellm_span_obj.end( + output={"model": model_name, "stream": True}, + metadata={"latency_ms": proxy_latency, "tokens": completion_chars // 4}, + ) + except Exception: + pass + except Exception as ex: + logger.error(f"Stream error: {ex}") + finally: + await r.aclose() + return StreamingResponse(stream_generator(), media_type="text/event-stream") + else: + error_body = await r.aread() if r else b"" + logger.warning(f"LiteLLM stream failed ({r.status_code}): {error_body[:300]}") + await r.aclose() + raise HTTPException(status_code=r.status_code, detail="LiteLLM upstream request failed") + else: + logger.info(f"Proxying to LiteLLM as model={model_name}") + response = await client.post(f"{backend_api_base}/chat/completions", json=body_to_send, headers=headers) + if response.status_code == 200: + proxy_latency = (time.time() - proxy_start) * 1000.0 + stats["total_proxy_time_ms"] += proxy_latency + stats["avg_proxy_latency_ms"] = stats["total_proxy_time_ms"] / stats["total_requests"] + resp_json = response.json() + usage = resp_json.get("usage", {}) + prompt_tokens = usage.get("prompt_tokens", estimate_prompt_tokens(body_to_send)) + completion_tokens = usage.get("completion_tokens", len(json.dumps(resp_json)) // 4) + record_tool_usage(active_tool, prompt_tokens, completion_tokens, model_name, proxy_latency, route="litellm_fallback") + # Finalize LiteLLM span (non-streaming path) + if litellm_span_obj: + try: + litellm_span_obj.end( + output={"model": model_name, "tokens": completion_tokens}, + metadata={"latency_ms": proxy_latency}, + ) + except Exception: + pass + return resp_json + else: + logger.warning(f"LiteLLM failed ({response.status_code}): {response.text[:300]}") + raise HTTPException(status_code=response.status_code, detail="LiteLLM upstream request failed") + except HTTPException: + raise + except Exception as exc: + logger.error(f"httpx call failed: {exc}") + raise HTTPException(status_code=502, detail=f"Proxy call failed: {exc}") from exc - # Set up outgoing proxy request - client = httpx.AsyncClient(timeout=3600.0) - headers = {"Authorization": f"Bearer {backend_api_key}"} - if langfuse_trace_id: - headers["X-Langfuse-Trace-Id"] = langfuse_trace_id + if should_try_ollama: + # Sync state from Valkey first + await sync_cooldowns_from_valkey() + + # --- Router-side Ollama cooldown check --- + global _ollama_cooldown_until + now_mono = time.monotonic() + if now_mono < _ollama_cooldown_until: + remaining = int(_ollama_cooldown_until - now_mono) + logger.warning( + f"⏳ Ollama cooldown active ({remaining}s remaining), " + f"skipping {target_model}" + ) + if client_model in ("llm-routing-auto-ollama", "llm-routing-auto-agy-ollama"): + # Auto mode: silently fall through to the free tier + logger.info(f"Auto-mode fallback: {target_model} → {original_target_model} (Ollama cooled down)") + return await execute_proxy(original_target_model) + else: + # Direct/fallback llm-routing-ollama: return 429 so LiteLLM + # skips this model group and moves to openrouter-auto + raise HTTPException( + status_code=429, + detail=f"Ollama backend cooled down ({remaining}s remaining)" + ) - # Handle streaming vs non-streaming proxying (LiteLLM handles fallback internally) - proxy_start = time.time() - model_name = target_model # LiteLLM handles fallback internally - - # --- Pre-screening: clamp max_tokens to fit within downstream model context limits --- - # Hermes agents set max_tokens=65536 unconditionally, but some free models have - # context limits as low as 32K. Without clamping, input+output exceeds the limit - # and the request fails with a 400 BadRequest (context_length exceeded). - # We estimate input tokens and clamp max_tokens to leave a 2K safety margin. - try: - body_to_send = body.copy() - body_to_send["model"] = model_name - requested_max_tokens = body_to_send.get("max_tokens", 4096) - if requested_max_tokens > 32768: # Only intervene for unusually large max_tokens - # Tier-aware minimum context length (from actual roster data): - # - agent-simple-core: 32K (includes tiny liquid/dolphin models) - # - agent-medium-core+: 256K (smallest non-tiny model is nemotron-nano-omni at 256K) - # - ollama-deepseek-v4-*: 1M (DeepSeek V4 native context) - _tier_min_ctx = { - "agent-simple-core": 32768, - "ollama-deepseek-v4-pro": 1000000, - "ollama-deepseek-v4-flash": 1000000, - } - _min_ctx = _tier_min_ctx.get(model_name, 262144) - # Rough input token estimate: 1 token ≈ 4 chars of JSON - _est_input = len(json.dumps(body_to_send)) // 4 - _safe_max = max(1024, _min_ctx - _est_input - 2048) # 2K safety margin - if requested_max_tokens > _safe_max: - logger.warning( - f"⛔ Clamping max_tokens: {requested_max_tokens} → {_safe_max} " - f"(est_input={_est_input}, min_ctx={_min_ctx}, tier={model_name})" + try: + result = await execute_proxy(target_model) + return result + except HTTPException as e: + is_transient = e.status_code in (429, 500, 502, 503, 504) + if is_transient: + # Ollama failure — activate router-side cooldown + _ollama_cooldown_until = time.monotonic() + OLLAMA_COOLDOWN_SECONDS + await save_cooldowns_to_valkey() + logger.error( + f"🧊 Ollama failed ({e.status_code}), activating {OLLAMA_COOLDOWN_SECONDS}s cooldown" ) - body_to_send["max_tokens"] = _safe_max - except Exception as e: - logger.warning(f"Pre-screening failed (non-fatal): {e}") - body_to_send = body.copy() - body_to_send["model"] = model_name - - try: - if "metadata" not in body_to_send or not isinstance(body_to_send["metadata"], dict): - body_to_send["metadata"] = {} - body_to_send["metadata"]["trace_name"] = "agent-completion" - - if body.get("stream", False): - logger.info(f"Proxying streaming to LiteLLM as model={model_name}") - req = client.build_request("POST", f"{backend_api_base}/chat/completions", json=body_to_send, headers=headers) - 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: - async for chunk in r.aiter_bytes(): - completion_chars += len(chunk) - yield chunk - proxy_latency = (time.time() - proxy_start) * 1000.0 - stats["total_proxy_time_ms"] += proxy_latency - stats["avg_proxy_latency_ms"] = stats["total_proxy_time_ms"] / stats["total_requests"] - record_tool_usage(active_tool, request_tokens, completion_chars // 4, model_name, proxy_latency, route="litellm_fallback") - # Finalize LiteLLM span (streaming path) - if litellm_span_obj: - try: - litellm_span_obj.end( - output={"model": model_name, "stream": True}, - metadata={"latency_ms": proxy_latency, "tokens": completion_chars // 4}, - ) - except Exception: - pass - except Exception as ex: - logger.error(f"Stream error: {ex}") - finally: - await r.aclose() - await client.aclose() - return StreamingResponse(stream_generator(), media_type="text/event-stream") + if client_model in ("llm-routing-auto-ollama", "llm-routing-auto-agy-ollama"): + if is_transient: + logger.warning(f"Ollama proxy failed ({e.detail}), falling back to free tier {original_target_model}") + return await execute_proxy(original_target_model) + else: + raise e else: - error_body = await r.aread() if r else b"" - logger.warning(f"LiteLLM stream failed ({r.status_code}): {error_body[:300]}") - await r.aclose(); await client.aclose() - raise HTTPException(status_code=502, detail=f"LiteLLM failed: {r.status_code}") - else: - logger.info(f"Proxying to LiteLLM as model={model_name}") - response = await client.post(f"{backend_api_base}/chat/completions", json=body_to_send, headers=headers) - await client.aclose() - if response.status_code == 200: - proxy_latency = (time.time() - proxy_start) * 1000.0 - stats["total_proxy_time_ms"] += proxy_latency - stats["avg_proxy_latency_ms"] = stats["total_proxy_time_ms"] / stats["total_requests"] - resp_json = response.json() - usage = resp_json.get("usage", {}) - prompt_tokens = usage.get("prompt_tokens", len(json.dumps(body_to_send)) // 4) - completion_tokens = usage.get("completion_tokens", len(json.dumps(resp_json)) // 4) - record_tool_usage(active_tool, prompt_tokens, completion_tokens, model_name, proxy_latency, route="litellm_fallback") - # Finalize LiteLLM span (non-streaming path) - if litellm_span_obj: - try: - litellm_span_obj.end( - output={"model": model_name, "tokens": completion_tokens}, - metadata={"latency_ms": proxy_latency}, - ) - except Exception: - pass - return resp_json + # Direct/fallback llm-routing-ollama request + if is_transient: + logger.error(f"Ollama proxy failed ({e.detail}) for direct/fallback request, returning 429") + raise HTTPException( + status_code=429, + detail="Ollama backend rate limited/unavailable" + ) from e + else: + raise e + except Exception as e: + # Unexpected error (timeouts, connection issues) — also cooldown to prevent hammering + _ollama_cooldown_until = time.monotonic() + OLLAMA_COOLDOWN_SECONDS + await save_cooldowns_to_valkey() + logger.error( + f"🧊 Ollama unexpected error ({e}), activating {OLLAMA_COOLDOWN_SECONDS}s cooldown" + ) + if client_model in ("llm-routing-auto-ollama", "llm-routing-auto-agy-ollama"): + logger.warning(f"Ollama proxy error ({e}), falling back to free tier {original_target_model}") + return await execute_proxy(original_target_model) else: - logger.warning(f"LiteLLM failed ({response.status_code}): {response.text[:300]}") - raise HTTPException(status_code=502, detail=f"LiteLLM failed: {response.status_code}") - except HTTPException: - raise - except Exception as e: - logger.error(f"Exception during LiteLLM proxy: {e}") - await client.aclose() - raise HTTPException(status_code=502, detail="LiteLLM upstream failed") + raise HTTPException( + status_code=429, + detail="Ollama backend rate limited/unavailable" + ) from e + else: + return await execute_proxy(target_model) @app.get("/metrics") async def metrics(): """Expose triage and circuit breaker metrics in Prometheus format.""" + await sync_cooldowns_from_valkey() breaker = get_breaker() breaker_status = breaker.status() @@ -1645,12 +2057,23 @@ async def metrics(): lines.append("# HELP circuit_breaker_total_trips Total trips across both breakers") lines.append("# TYPE circuit_breaker_total_trips counter") lines.append(f"circuit_breaker_total_trips {google['total_trips'] + vendor['total_trips']}") + + # Ollama router-side cooldown metrics + _now_mono = time.monotonic() + _ollama_remaining = max(0.0, _ollama_cooldown_until - _now_mono) + lines.append("# HELP ollama_cooldown_active Whether Ollama is in router-side cooldown (1=active)") + lines.append("# TYPE ollama_cooldown_active gauge") + lines.append(f"ollama_cooldown_active {int(_ollama_remaining > 0)}") + lines.append("# HELP ollama_cooldown_remaining_seconds Seconds remaining in Ollama cooldown") + lines.append("# TYPE ollama_cooldown_remaining_seconds gauge") + lines.append(f"ollama_cooldown_remaining_seconds {_ollama_remaining:.0f}") return Response(content="\n".join(lines), media_type="text/plain; version=0.0.4") @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.""" + await sync_cooldowns_from_valkey() # 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/") @@ -1658,7 +2081,7 @@ async def get_dashboard(): langfuse_status = await check_http_endpoint("http://127.0.0.1:3001") # 1c. Check Gemini OAuth token status - oauth_status = get_gemini_oauth_status() + oauth_status = await asyncio.to_thread(get_gemini_oauth_status) # Pre-compute oauth_banner_html to avoid nested f-string and JavaScript bracket escaping issues oauth_banner_html = "" @@ -2637,9 +3060,11 @@ 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()) + content = await asyncio.to_thread(vis_path.read_text, encoding="utf-8") + return HTMLResponse(content) 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 @@ -2647,6 +3072,15 @@ class AnnotationItem(BaseModel): ts: Optional[str] = None VALID_TIERS = {"agent-simple-core", "agent-medium-core", "agent-complex-core", "agent-reasoning-core", "agent-advanced-core"} +# NOTE: annotations_lock (asyncio.Lock) only provides concurrency protection within +# a single Python process. In multi-worker uvicorn deployments, concurrent requests +# across different workers can still race. Eventual consistency is maintained via +# the atomic file-replace mechanism, which is acceptable for this dashboard feature. +annotations_lock = asyncio.Lock() + +def _read_annotations_sync(path) -> dict: + 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]): @@ -2696,19 +3130,19 @@ async def save_annotations(payload: Dict[str, AnnotationItem]): try: ann_path = DATA_DIR / "annotations.json" existing = {} - if ann_path.exists(): - try: - import json - with open(ann_path, "r", encoding="utf-8") as f: - existing = json.load(f) - except Exception as read_err: - logger.warning(f"Could not read existing annotations: {read_err}. Overwriting.") - - # Merge new annotations into existing - for k, item in payload.items(): - existing[k] = item.model_dump() if hasattr(item, "model_dump") else item.dict() + async with annotations_lock: + if ann_path.exists(): + try: + existing = await asyncio.to_thread(_read_annotations_sync, str(ann_path)) + except Exception as read_err: + logger.warning(f"Could not read existing annotations: {read_err}. Overwriting.") - await _atomic_write_json_async(str(ann_path), existing) + # Merge new annotations into existing + for k, item in payload.items(): + existing[k] = item.model_dump() if hasattr(item, "model_dump") else item.dict() + + await _atomic_write_json_async(str(ann_path), existing) + return JSONResponse({"status": "ok", "saved": len(payload)}) except Exception as e: logger.error(f"Failed to save annotations: {e}") diff --git a/scripts/README.md b/scripts/README.md new file mode 100644 index 00000000..992479c8 --- /dev/null +++ b/scripts/README.md @@ -0,0 +1,70 @@ +# Automation, Testing, and Verification Scripts + +This directory and the repository root contain various scripts used for stack orchestration, database backups, routing verification, classifier training/benchmarking, and system integration testing. + +--- + +## 1. Stack Orchestration & Backups + +### `start-stack.sh` (Root Directory) +Unified startup and credential extraction script for the Podman Kubernetes container stack. +- **Usage**: + - `./start-stack.sh` (Restart existing pod — fast, preserves logs) + - `./start-stack.sh --replace` (Stop + clean ports + redeploy pod from `pod.yaml`) + - `./start-stack.sh --full-rebuild` (Same as `--replace` + rebuild the triage router image; required for code changes in `router/`) + +### `scripts/backup.sh` +Automated database backup script that runs before every stack deployment. Uses `pg_isready` to safely wait for database connections and manages timestamped backups under `backups/`. + +--- + +## 2. Routing & Cooldown Verification Scripts + +These scripts are located in `scripts/verification/` and are used to assert that the router-side Ollama cooldowns and prompt-classification gating function correctly: + +### `scripts/verification/verify_ollama_routing.py` +Sends sample prompts of varying complexity to `llm-routing-auto-ollama` and `llm-routing-ollama` to verify correct gating and routing: +- **Expected Routing (`llm-routing-auto-ollama`)**: + - Simple $\rightarrow$ `agent-simple-core` + - Complex $\rightarrow$ `ollama-deepseek-v4-flash` + - Reasoning $\rightarrow$ `ollama-deepseek-v4-pro` +- **Expected Routing (`llm-routing-ollama`)**: + - Simple/Complex $\rightarrow$ `ollama-deepseek-v4-flash` + - Reasoning/Advanced $\rightarrow$ `ollama-deepseek-v4-pro` + +### `scripts/verification/verify_ollama_cooldown.py` +Simulates fallback cascades to verify that failed Ollama requests activate the 5-minute router-side cooldown and correctly bypass LiteLLM to prevent crash loops. + +### `scripts/verification/verify_direct_ollama_cooldown.py` +Asserts that direct requests to `llm-routing-ollama` immediately trigger the cooldown response without hammering downstream endpoints. + +### `scripts/verification/mock_rate_limit_server.py` +A simple HTTP server that returns `429 Rate Limit Exceeded` to simulate rate limits when testing cooldowns. +- **Usage**: `python3 scripts/verification/mock_rate_limit_server.py` (Runs on `127.0.0.1:9999`) + +--- + +## 3. Classifier & Dataset Maintenance (`scripts/`) + +These tools are used to benchmark the prompt classifier and extract datasets from Langfuse traces: + +- **`benchmark_classifier.py`**: Benchmarks latency and precision metrics of the Ryzen PRO APU-offloaded classifier. +- **`classify_direct.py`**: Takes a string prompt argument and prints the classification decision directly. +- **`extract_prompts.py` / `extract_complex.py` / `extract_gapfill.py`**: Mines prompt datasets from Langfuse PG/ClickHouse database traces for fine-tuning. +- **`reclassify_all.py`**: Re-evaluates prompt classifications against updated models. +- **`retry_errors.py`**: Retries failed queries. + +--- + +## 4. Integration Test Suite (Root Directory) + +- **`test_circuit_breaker.py`**: Unit/integration tests for the dual circuit breaker (`router/circuit_breaker.py`), covering independent Google/Vendor tiers and probe-granting logic. +- **`test_classifier_accuracy.py`**: Accuracy evaluation suite covering 25 system prompts. +- **`test_agy_tiers.py`**: Validates `agy` proxy model tier routing. +- **`test_agy_behavior.py`**: Asserts the behavior of the `agy` CLI client under quota limits. +- **`test_a2_verify.py`**: Quick sanity integration check for the agy proxy circuit breaker. +- **`test_antigravity.py`**: Tests the connection to the host Antigravity CLI daemon (`agentapi`). +- **`test_stream_latency.py`**: Measures Time-To-First-Token (TTFT) and token generation speed. +- **`test_quota_reset.sh`**: Simulates/triggers quota reset conditions. +- **`verify_breaker.py`**: Sanity verification check for the circuit breaker. +- **`watch_quota.sh`**: Watch/polling script for observing quota status. diff --git a/scripts/backup.sh b/scripts/backup.sh index 2cd26151..2000ddb7 100755 --- a/scripts/backup.sh +++ b/scripts/backup.sh @@ -7,8 +7,9 @@ set -e # Scheduled via: systemctl --user enable llm-backup.timer # ============================================================ +WORKDIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" TIMESTAMP=$(date +%Y%m%d_%H%M%S) -BACKUP_DIR="/home/gpav/Vrac/LAB/AI/LLM-Routing/backups" +BACKUP_DIR="${WORKDIR}/backups" RETENTION_DAYS=14 LOG_FILE="/tmp/llm-backup-${TIMESTAMP}.log" @@ -53,7 +54,7 @@ fi # ---- Config Files (lightweight copy) ---- CONFIG_SNAPSHOT="${BACKUP_DIR}/configs_${TIMESTAMP}.tar.gz" tar czf "$CONFIG_SNAPSHOT" \ - -C /home/gpav/Vrac/LAB/AI/LLM-Routing \ + -C "$WORKDIR" \ litellm/config.yaml \ router/config.yaml \ router/main.py \ diff --git a/scripts/extract_gapfill.py b/scripts/extract_gapfill.py index e5b41fd3..b01a7bcd 100644 --- a/scripts/extract_gapfill.py +++ b/scripts/extract_gapfill.py @@ -46,8 +46,10 @@ def extract_user_prompt(obs): if not inp: return None if isinstance(inp, str): - try: inp = json.loads(inp) - except: return None + try: + inp = json.loads(inp) + except json.JSONDecodeError: + return None if not isinstance(inp, dict): return None messages = inp.get('messages', []) diff --git a/scripts/extract_prompts.py b/scripts/extract_prompts.py index 028baa8d..789da769 100644 --- a/scripts/extract_prompts.py +++ b/scripts/extract_prompts.py @@ -143,7 +143,7 @@ def extract_first_user_prompt(obs): lengths = [len(p['prompt']) for p in prompts] 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):") + print(f"Short (<100 chars): {sum(1 for length in lengths if length < 100)}") + print("\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 index 91ea6b61..c2bce257 100644 --- a/scripts/reclassify_all.py +++ b/scripts/reclassify_all.py @@ -38,14 +38,6 @@ def classify(prompt): 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)...") diff --git a/scripts/retry_errors.py b/scripts/retry_errors.py index eff8be8c..5568c698 100644 --- a/scripts/retry_errors.py +++ b/scripts/retry_errors.py @@ -1,5 +1,5 @@ """Retry the 94 failed prompts with 800-char truncation (safe for 4096-ctx model).""" -import json, urllib.request, time, subprocess +import json, urllib.request, time, subprocess, tempfile, os from pathlib import Path from collections import Counter @@ -65,8 +65,6 @@ def classify(prompt): 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) def is_error_val(val): if not val: @@ -110,8 +108,11 @@ def is_error_val(val): 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) +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(dataset, tmp_f, indent=2, ensure_ascii=False) + tmp_name = tmp_f.name +os.replace(tmp_name, str(dest_path)) print(f"\nDone. Fixed: {fixed}, Errors: {errors}") for tier in sorted(new_counts.keys()): diff --git a/scripts/verification/mock_rate_limit_server.py b/scripts/verification/mock_rate_limit_server.py new file mode 100644 index 00000000..02222361 --- /dev/null +++ b/scripts/verification/mock_rate_limit_server.py @@ -0,0 +1,27 @@ +import http.server +import sys + +class MyHandler(http.server.BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + # Print to stdout/stderr so we can see it in logs + sys.stderr.write("%s - - [%s] %s\n" % (self.address_string(), self.log_date_time_string(), fmt%args)) + + def do_POST(self): + print(f"Mock server: received POST request to {self.path}", flush=True) + self.send_response(429) + self.send_header('Content-Type', 'application/json') + self.end_headers() + self.wfile.write(b'{"error":{"message":"Rate limit exceeded","type":"rate_limit_error","param":null,"code":null}}') + +def main(): + port = 9999 + print(f"Starting mock 429 rate limit server on 127.0.0.1:{port}...", flush=True) + server = http.server.HTTPServer(('127.0.0.1', port), MyHandler) + try: + server.serve_forever() + except KeyboardInterrupt: + pass + print("Stopping mock server...", flush=True) + +if __name__ == "__main__": + main() diff --git a/scripts/verification/verify_direct_ollama_cooldown.py b/scripts/verification/verify_direct_ollama_cooldown.py new file mode 100644 index 00000000..682610d9 --- /dev/null +++ b/scripts/verification/verify_direct_ollama_cooldown.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +import json +import time +import os +import uuid +import sys +import httpx + +# Resolve the absolute path to .env file in the workspace +workspace_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +env_path = os.path.join(workspace_dir, ".env") + +litellm_key = "gateway-pass" +if os.path.exists(env_path): + with open(env_path, "r") as f: + for line in f: + if line.startswith("LITELLM_MASTER_KEY="): + litellm_key = line.split("=", 1)[1].strip().strip('"').strip("'") + break + +LITELLM_URL = "http://localhost:4000/v1/chat/completions" +METRICS_URL = "http://localhost:5000/metrics" + +def get_triage_request_count(): + try: + response = httpx.get(METRICS_URL, timeout=5.0) + response.raise_for_status() + lines = response.text.splitlines() + for line in lines: + if line.startswith("triage_requests_total"): + return int(float(line.split()[1])) + except (httpx.HTTPError, ValueError) as e: + print(f"Error fetching metrics: {e}") + return 0 + +def send_litellm_request(model: str, prompt: str): + unique_prompt = f"{prompt} [id: {uuid.uuid4()}]" + payload = { + "model": model, + "messages": [ + {"role": "user", "content": unique_prompt} + ], + "temperature": 0.0, + "max_tokens": 10 + } + start_time = time.time() + try: + response = httpx.post( + LITELLM_URL, + json=payload, + headers={"Authorization": f"Bearer {litellm_key}"}, + timeout=30.0 + ) + response.raise_for_status() + result = response.json() + model_returned = result.get("model", "unknown") + text = (result["choices"][0]["message"].get("content") or "").strip() + print(f"Success in {time.time() - start_time:.1f}s: model={model_returned}, text='{text[:40]}'") + return True, model_returned + except httpx.HTTPStatusError as e: + err_msg = f"{e} - {e.response.text}" + print(f"Failed in {time.time() - start_time:.1f}s: {err_msg}") + return False, err_msg + except httpx.HTTPError as e: + err_msg = str(e) + print(f"Failed in {time.time() - start_time:.1f}s: {err_msg}") + return False, err_msg + except (KeyError, IndexError, ValueError) as e: + err_msg = f"Parse error: {e}" + print(f"Failed in {time.time() - start_time:.1f}s: {err_msg}") + return False, err_msg + +def main(): + print("--- Verifying Direct llm-routing-ollama Cooldown ---") + print(f"Using LiteLLM Master Key: {'set' if litellm_key else 'missing'}") + + # 1. Get initial triage request count + count_init = get_triage_request_count() + print(f"Initial triage requests count: {count_init}") + + # 2. Send first request directly to llm-routing-ollama. + # Since Ollama deepseek-v4-pro is offline/unauthorized, it will fail, which should return an error + # to LiteLLM, triggering immediate cooldown for llm-routing-ollama. + print("\nSending first request to llm-routing-ollama...") + send_litellm_request("llm-routing-ollama", "Design a distributed pub/sub system with Valkey and describe failover states") + + # 3. Check triage requests count. + count_after_1 = get_triage_request_count() + print(f"Triage requests count after 1st request: {count_after_1}") + + # 4. Send second request to llm-routing-ollama. + # Since llm-routing-ollama should now be on cooldown in LiteLLM, LiteLLM should reject it immediately + # without proxying to the triage router. + print("\nSending second request to llm-routing-ollama (should be skipped / fail immediately via cooldown)...") + send_litellm_request("llm-routing-ollama", "Design a distributed pub/sub system with Valkey and describe failover states") + + # 5. Check triage requests count. + count_after_2 = get_triage_request_count() + print(f"Triage requests count after 2nd request: {count_after_2}") + + diff = count_after_2 - count_after_1 + + if count_after_1 > count_init: + print("✓ First request successfully reached the triage router.") + if diff == 0: + print("✅ SUCCESS: llm-routing-ollama was successfully cooled down and skipped on the second request!") + else: + print(f"❌ FAILURE: llm-routing-ollama was NOT cooled down (count increased by {diff})!") + sys.exit(1) + else: + print("❌ FAILURE: First request did not even reach the triage router.") + sys.exit(1) + +if __name__ == "__main__": + main() diff --git a/scripts/verification/verify_ollama_cooldown.py b/scripts/verification/verify_ollama_cooldown.py new file mode 100644 index 00000000..991c4049 --- /dev/null +++ b/scripts/verification/verify_ollama_cooldown.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +import json +import time +import os +import uuid +import sys +import httpx + +# Resolve the absolute path to .env file in the workspace +workspace_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +env_path = os.path.join(workspace_dir, ".env") + +# Read LITELLM_MASTER_KEY from .env +litellm_key = "gateway-pass" +if os.path.exists(env_path): + with open(env_path, "r") as f: + for line in f: + if line.startswith("LITELLM_MASTER_KEY="): + # extract value inside quotes + litellm_key = line.split("=", 1)[1].strip().strip('"').strip("'") + break + +LITELLM_URL = "http://localhost:4000/v1/chat/completions" +METRICS_URL = "http://localhost:5000/metrics" + +def get_triage_request_count(): + try: + response = httpx.get(METRICS_URL, timeout=5.0) + response.raise_for_status() + lines = response.text.splitlines() + for line in lines: + if line.startswith("triage_requests_total"): + return int(float(line.split()[1])) + except (httpx.HTTPError, ValueError) as e: + print(f"Error fetching metrics: {e}") + return 0 + +def send_litellm_request(model: str, prompt: str): + unique_prompt = f"{prompt} [id: {uuid.uuid4()}]" + payload = { + "model": model, + "messages": [ + {"role": "user", "content": unique_prompt} + ], + "temperature": 0.0, + "max_tokens": 10 + } + start_time = time.time() + try: + response = httpx.post( + LITELLM_URL, + json=payload, + headers={"Authorization": f"Bearer {litellm_key}"}, + timeout=30.0 + ) + response.raise_for_status() + result = response.json() + model_returned = result.get("model", "unknown") + text = (result["choices"][0]["message"].get("content") or "").strip() + print(f"Success in {time.time() - start_time:.1f}s: model={model_returned}, text='{text[:40]}'") + return True, model_returned + except httpx.HTTPStatusError as e: + err_msg = f"{e} - {e.response.text}" + print(f"Failed in {time.time() - start_time:.1f}s: {err_msg}") + return False, err_msg + except httpx.HTTPError as e: + err_msg = str(e) + print(f"Failed in {time.time() - start_time:.1f}s: {err_msg}") + return False, err_msg + except (KeyError, IndexError, ValueError) as e: + err_msg = f"Parse error: {e}" + print(f"Failed in {time.time() - start_time:.1f}s: {err_msg}") + return False, err_msg + +def main(): + print("--- Verifying Ollama Cooldown and Skip Behavior ---") + print(f"Using LiteLLM Master Key: {'set' if litellm_key else 'missing'}") + + # 1. Get initial triage request count + count_init = get_triage_request_count() + print(f"Initial triage requests count: {count_init}") + + # 2. Send first request to agent-advanced-core. + print("\nSending first request to agent-advanced-core...") + send_litellm_request("agent-advanced-core", "Design a distributed pub/sub system with Valkey and describe failover states") + + # 3. Check triage requests count. + count_after_1 = get_triage_request_count() + print(f"Triage requests count after 1st request: {count_after_1}") + + # 4. Send second request to agent-advanced-core. + print("\nSending second request to agent-advanced-core (llm-routing-ollama should be skipped)...") + send_litellm_request("agent-advanced-core", "Design a distributed pub/sub system with Valkey and describe failover states") + + # 5. Check triage requests count. + count_after_2 = get_triage_request_count() + print(f"Triage requests count after 2nd request: {count_after_2}") + + diff = count_after_2 - count_after_1 + + # Verify by checking if the count incremented on the first request and stayed constant on the second + if count_after_1 > count_init: + print("✓ First request successfully reached the triage router via fallback!") + if diff == 0: + print("✅ SUCCESS: llm-routing-ollama was successfully skipped (cooled down) on the second request!") + else: + print(f"❌ FAILURE: llm-routing-ollama was NOT skipped (count increased by {diff})!") + sys.exit(1) + else: + print("❌ FAILURE: First request did not even reach the triage router (check if all free models failed immediately without fallback).") + sys.exit(1) + +if __name__ == "__main__": + main() diff --git a/scripts/verification/verify_ollama_routing.py b/scripts/verification/verify_ollama_routing.py new file mode 100644 index 00000000..b873d82d --- /dev/null +++ b/scripts/verification/verify_ollama_routing.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +import json +import sys +import os +import httpx + +URL = "http://localhost:5000/v1/chat/completions" + +# Resolve the absolute path to .env file in the workspace +workspace_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +env_path = os.path.join(workspace_dir, ".env") + +# Read LITELLM_MASTER_KEY from .env +litellm_key = "gateway-pass" +if os.path.exists(env_path): + with open(env_path, "r") as f: + for line in f: + if line.startswith("LITELLM_MASTER_KEY="): + # extract value inside quotes + litellm_key = line.split("=", 1)[1].strip().strip('"').strip("'") + break + +def send_request(model: str, prompt: str, expected_model: str): + payload = { + "model": model, + "messages": [ + {"role": "user", "content": prompt} + ], + "temperature": 0.0, + "max_tokens": 10 + } + try: + response = httpx.post( + URL, + json=payload, + headers={"Authorization": f"Bearer {litellm_key}"}, + timeout=30.0 + ) + response.raise_for_status() + result = response.json() + model_returned = result.get("model", "unknown") + text = (result["choices"][0]["message"].get("content") or "").strip() + print(f"Request: model={model}, prompt='{prompt[:40]}...'") + print(f"Response: model={model_returned}, text='{text[:60]}...'") + if model_returned != expected_model: + print(f"❌ FAILURE: Expected model '{expected_model}', but got '{model_returned}'", file=sys.stderr) + sys.exit(1) + print("✓ SUCCESS: Routed correctly!") + except httpx.HTTPStatusError as e: + print(f"❌ HTTP ERROR: Request to model={model} failed with status {e.response.status_code}: {e}\nResponse body:\n{e.response.text}", file=sys.stderr) + sys.exit(1) + except httpx.HTTPError as e: + print(f"❌ HTTP ERROR: Request to model={model} failed: {e}", file=sys.stderr) + sys.exit(1) + except (json.JSONDecodeError, KeyError, ValueError) as e: + print(f"❌ PARSE ERROR: Failed to parse response for model={model}: {e}", file=sys.stderr) + sys.exit(1) + +def main(): + print("--- 1. Testing llm-routing-auto-ollama ---") + # Simple prompt -> should route to agent-simple-core + send_request("llm-routing-auto-ollama", "Write a hello world in Python", "agent-simple-core") + + # Complex prompt -> should route to ollama-deepseek-v4-flash + send_request("llm-routing-auto-ollama", "Implement a custom memory-efficient Trie in C++", "ollama-deepseek-v4-flash") + + # Reasoning prompt -> should route to ollama-deepseek-v4-pro + send_request("llm-routing-auto-ollama", "Design a distributed pub/sub system with Valkey and describe failover states", "ollama-deepseek-v4-pro") + + print("\n--- 2. Testing llm-routing-ollama ---") + # Simple prompt -> should route to ollama-deepseek-v4-flash + send_request("llm-routing-ollama", "Write a hello world in Python", "ollama-deepseek-v4-flash") + + # Complex prompt -> should route to ollama-deepseek-v4-flash + send_request("llm-routing-ollama", "Implement a custom memory-efficient Trie in C++", "ollama-deepseek-v4-flash") + + # Reasoning prompt -> should route to ollama-deepseek-v4-pro + send_request("llm-routing-ollama", "Design a distributed pub/sub system with Valkey and describe failover states", "ollama-deepseek-v4-pro") + +if __name__ == "__main__": + main() diff --git a/start-stack.sh b/start-stack.sh index 21ac9496..f9d98463 100755 --- a/start-stack.sh +++ b/start-stack.sh @@ -6,10 +6,10 @@ set -e # ./start-stack.sh --replace → Graceful stop + clean ports + redeploy pod # (for pod.yaml changes: ports, probes, env vars) # ./start-stack.sh --full-rebuild → Same as --replace + rebuild router image -# (for router/Containerfile changes) +# (for router/Dockerfile changes) # Set working directory -WORKDIR="/home/gpav/Vrac/LAB/AI/LLM-Routing" +WORKDIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" cd "$WORKDIR" # Ensure local volume directories exist on the host for Podman mounts @@ -43,7 +43,7 @@ if [ -z "$OPENROUTER_API_KEY" ]; then fi # 2. Sync Gemini OAuth token (skip if <15 min old) -OAUTH_CREDS="/home/gpav/.gemini/oauth_creds.json" +OAUTH_CREDS="$HOME/.gemini/oauth_creds.json" NEED_SYNC=true if [ -f "$OAUTH_CREDS" ]; then CREDS_AGE=$(($(date +%s) - $(stat -c %Y "$OAUTH_CREDS" 2>/dev/null || echo 0))) @@ -79,13 +79,17 @@ else echo "⚠️ Warning: Host agy daemon not responding on port 5005" fi -# 3. Use LiteLLM master key from .env if present, otherwise generate a random one if [ -z "$LITELLM_MASTER_KEY" ]; then LITELLM_MASTER_KEY="sk-litellm-$(openssl rand -hex 16)" echo "LITELLM_MASTER_KEY=\"$LITELLM_MASTER_KEY\"" >> "$ENV_FILE" echo "✓ Generated new LiteLLM master key and saved to $ENV_FILE" fi +if [ -z "$LITELLM_MASTER_KEY" ]; then + echo "❌ Error: LITELLM_MASTER_KEY is not set and could not be generated." + exit 1 +fi + # DYNAMIC_LITELLM_MASTER_KEY_PLACEHOLDER in router config is resolved at runtime from env FULL_REBUILD=false @@ -291,23 +295,36 @@ safe_pod_teardown() { if podman pod exists agent-router-pod 2>/dev/null; then echo "💾 Taking pre-deploy database backup..." bash scripts/backup.sh && echo "✓ Pre-deploy backup saved" || echo "⚠️ Pre-deploy backup skipped" -else - echo "⚠️ Pod not running — skipping pre-deploy backup" fi +render_pod_yaml() { + export WORKDIR HOME LITELLM_MASTER_KEY + python3 - "$WORKDIR/pod.yaml" <<'PY' +import os, sys +uid = os.getuid() +with open(sys.argv[1], "r", encoding="utf-8") as f: + text = f.read() +text = text.replace("/home/gpav/Vrac/LAB/AI/LLM-Routing", os.environ["WORKDIR"]) +text = text.replace("/home/gpav/", os.environ["HOME"] + "/") +text = text.replace("/run/user/1000", f"/run/user/{uid}") +text = text.replace("sk-lit...33bf", os.environ["LITELLM_MASTER_KEY"]) +sys.stdout.write(text) +PY +} + if podman pod exists agent-router-pod 2>/dev/null; then if $FULL_REBUILD; then echo "🔨 Building custom local triage router image..." - podman build -t localhost/llm-triage-router:latest -f router/Containerfile router + podman build -t localhost/llm-triage-router:latest -f router/Dockerfile router safe_pod_teardown echo "🚀 Deploying fresh triage pod..." - podman play kube "$WORKDIR/pod.yaml" + render_pod_yaml | podman play kube - setup_minio_buckets verify_stack_health elif $REPLACE_MODE; then safe_pod_teardown echo "🚀 Deploying replacement pod from YAML..." - podman play kube "$WORKDIR/pod.yaml" + render_pod_yaml | podman play kube - setup_minio_buckets verify_stack_health else @@ -330,10 +347,10 @@ else # First deploy — no pod exists, clean ports just in case cleanup_zombie_ports echo "🔨 Building custom local triage router image..." - podman build -t localhost/llm-triage-router:latest -f router/Containerfile router + podman build -t localhost/llm-triage-router:latest -f router/Dockerfile router echo "🚀 No existing pod found. Deploying fresh triage pod..." - podman play kube "$WORKDIR/pod.yaml" + render_pod_yaml | podman play kube - setup_minio_buckets verify_stack_health fi diff --git a/sync_gemini_token.py b/sync_gemini_token.py index 5d852a21..42e987a0 100755 --- a/sync_gemini_token.py +++ b/sync_gemini_token.py @@ -6,7 +6,7 @@ import os from datetime import datetime -TARGET_PATH = "/home/gpav/.gemini/oauth_creds.json" +TARGET_PATH = os.path.expanduser("~/.gemini/oauth_creds.json") def main(): try: diff --git a/test_antigravity.py b/test_antigravity.py index a8e7a226..4244b232 100644 --- a/test_antigravity.py +++ b/test_antigravity.py @@ -11,12 +11,20 @@ def test_antigravity_connection(): print("--- Testing antigravity-cli connection with current OAuth ---") - # We simulate a request to Gemini 3.5 Flash via antigravity-cli - # Using the agentapi binary located at /home/gpav/.gemini/antigravity-cli/bin/agentapi + # Using the agentapi binary located at ~/.gemini/antigravity-cli/bin/agentapi + agentapi_path = os.path.expanduser("~/.gemini/antigravity-cli/bin/agentapi") + if not os.path.exists(agentapi_path): + try: + import pytest + pytest.skip(f"agentapi binary not found at {agentapi_path}; skipping health check") + except ImportError: + print(f"agentapi binary not found at {agentapi_path}; skipping health check") + return + try: # Testing non-interactive print mode result = subprocess.run( - ["/home/gpav/.gemini/antigravity-cli/bin/agentapi", "--print", "Hello, who are you?"], + [agentapi_path, "--print", "Hello, who are you?"], capture_output=True, text=True, timeout=20 diff --git a/test_classifier_accuracy.py b/test_classifier_accuracy.py index a3a803a4..018082a6 100644 --- a/test_classifier_accuracy.py +++ b/test_classifier_accuracy.py @@ -8,7 +8,7 @@ import urllib.error # Load config to get system prompt -CONFIG_PATH = os.getenv("CONFIG_PATH", "/home/gpav/Vrac/LAB/AI/LLM-Routing/router/config.yaml") +CONFIG_PATH = os.getenv("CONFIG_PATH", os.path.join(os.path.dirname(os.path.abspath(__file__)), "router", "config.yaml")) system_prompt = "" if os.path.exists(CONFIG_PATH): try: diff --git a/test_goose.py b/test_goose.py deleted file mode 100644 index 7df869a1..00000000 --- a/test_goose.py +++ /dev/null @@ -1 +0,0 @@ -print("Hello, World!")