From 06f20ae7522baae506d517ec2a7fb0795671cbe5 Mon Sep 17 00:00:00 2001 From: claude-noether Date: Mon, 20 Jul 2026 13:50:18 +0200 Subject: [PATCH] llm-proxy: per-backend admission gate (semaphore, bounded queue, 503 on saturation) ThreadingHTTPServer fanned unbounded concurrent forwards into single-slot llama.cpp -> pile-up (client-disconnect never cancels upstream, so contention cascades into a hang). Add a BoundedSemaphore per local backend, slots from local-backends.json (qwen3-4b=1, bosch-dspark=32), in-system cap = slots*4 -> fast 503 backend_busy on saturation, ~90s bounded wait otherwise. Gates local backends only. Verified: 6 concurrent -> 4 served serially + 2 fast-503, no pile-up. Deployed on hossenfelder. Co-Authored-By: Claude Opus 4.8 --- llm-proxy.py | 1319 ++++++++++++++++++++++++++++++++----------- local-backends.json | 4 + 2 files changed, 996 insertions(+), 327 deletions(-) create mode 100644 local-backends.json diff --git a/llm-proxy.py b/llm-proxy.py index 1ec5669..4d6d37a 100755 --- a/llm-proxy.py +++ b/llm-proxy.py @@ -1,203 +1,595 @@ #!/usr/bin/env python3 -"""Boltzmann-local LLM aggregator with compression + classifier middleware. +"""LLM router: model-prefix → OpenRouter (cloud) or local model-aware fanout. -Behaviour (unchanged from v2): - - GET /v1/models : merged model list from every reachable backend - - POST /v1/chat/completions : routes by model name; compresses gemma4 messages - - POST /v1/completions : same routing rule +Local backends: model-name routed when possible, first-up failover otherwise. + - On POST with a known model id, the request goes to the backend whose + /v1/models advertises it (cached 30s). + - On unknown model or empty body, falls back to first-up wins, + preserving legacy semantics. -New: - - POST /v1/classify : calls local Llama 3.2 1B classifier - - POST /v1/compress : token-budget compression of chat messages +Cloud routing: HTTPS to openrouter.ai, server-side Bearer auth from OPENROUTER_API_KEY. + +/v1/models: merged cloud catalog (live-filtered against OpenRouter's +/api/v1/models) + union of every reachable local backend's model list. +Entries are visually tagged in their id: + - "[$] vendor/model" paid cloud (manually curated or auto-discovered) + - "[free] vendor/model" free cloud (auto-detected by :free suffix) + - "[local] alias" local backend + +A periodic background thread (every 3 h) re-fetches the full OpenRouter model +list and: + a) adds newly free models >7 B parameters to the curated free catalog + b) removes models that are no longer free + c) probes each free model with a tiny 1-token request; if it returns 429 the + model id gets a "-na" suffix (not available). + d) auto-discovers cheap paid models (prompt < $0.50/MTok) from known + prefixes and adds them to the paid catalog. + +Paid models can also be added manually by editing cloud-catalog.json's "paid" +section — the background checker never removes them. + +Inbound requests have the tag + optional "-na" suffix stripped before routing, +so OWUI sending back the displayed id Just Works. + +Config: + /etc/llm-proxy/local-backends.json — list of {name, host, port} dicts + /etc/llm-proxy/cloud-catalog.json — persisted cloud model catalog + status + +Signals: + SIGHUP reload local-backends.json + SIGUSR1 drop discover caches + trigger an immediate background re-check """ import http.server import http.client -import json -import os +import ssl import sys +import os +import os.path import time -import urllib.request -import urllib.error +import json +import threading +import signal +import re +import copy +import traceback -BACKENDS = [ - ("gemma4-12b-q4km", "127.0.0.1", 8087), - ("classifier-1b", "127.0.0.1", 8089), - ("coder-1.5b", "127.0.0.1", 8081), - ("llama-3.1", "127.0.0.1", 8083), - ("nemo-12b", "127.0.0.1", 8084), - ("qwen3-30b-a3b", "127.0.0.1", 8085), - ("qwen3.5-9b-npu", "127.0.0.1", 8086), +LOCAL_BACKENDS_FILE = "/etc/llm-proxy/local-backends.json" +CLOUD_CATALOG_FILE = "/etc/llm-proxy/cloud-catalog.json" + +DEFAULT_LOCAL_BACKENDS = [ + ("clevo", "clevo.fritz.box", 8081), + ("dirac", "dirac.fritz.box", 8080), + ("boltzmann", "boltzmann.fritz.box", 8082), ] -CONNECT_TIMEOUT = 30 -READ_TIMEOUT = 1800 -LISTEN_PORT = int(os.getenv("LLM_PROXY_PORT", "8082")) -CLASSIFIER_URL = os.getenv("CLASSIFIER_URL", "http://127.0.0.1:8090/classify") -COMPRESS_URL = os.getenv("COMPRESS_URL", "http://127.0.0.1:8091/compress") +FRIENDLY_NAMES = { + "gemma4": "escher", + "gemma4-12b-q4km": "gemma4", + "qwen3.5-9b-npu": "qwen3.5", + "qwen3-30b-a3b": "qwen3.6", +} -# Gemma 4 12B context limit - reserve 4096 for completion -GEMMA_MAX_INPUT_TOKENS = 65536 - 4096 -FEW_SHOT_EXAMPLES = [ - ("What is 2+2?", "low"), - ("List the files in the current directory", "low"), - ("What is the weather today?", "low"), - ("Write a Python function to sort a list of dictionaries by a key", "medium"), - ("Explain how garbage collection works in Go", "medium"), - ("Debug this error: TypeError: 'NoneType' object is not subscriptable", "medium"), - ("Deploy the new release to production", "high"), - ("Roll back the last database migration", "high"), - ("Commit and push all changes to main branch", "high"), -] +CONNECT_TIMEOUT = 3 +READ_TIMEOUT = 1800 +DISCOVERY_TTL = 30 +CLOUD_DISCOVERY_TTL = 600 +CHECK_INTERVAL = 10800 +PROBE_TIMEOUT = 15 +NA_SUFFIX = "-na" +DOWN_SUFFIX = "-down" +LAST_KNOWN_FILE = "/etc/llm-proxy/last-known-models.json" + +OPENROUTER_HOST = "openrouter.ai" +OPENROUTER_PORT = 443 +CLOUD_PREFIXES = ( + "moonshotai/", "poolside/", "z-ai/", "cognitivecomputations/", "liquid/", + "openrouter/", "anthropic/", "openai/", "google/", "deepseek/", "meta-llama/", + "mistralai/", "qwen/", "x-ai/", "nvidia/", "cohere/", "perplexity/", + "amazon/", "microsoft/", "nousresearch/", +) + +PAID_DISCOVER_THRESHOLD = 0.50 # max prompt $/MTok for auto-discover + +TAG_PAID = "[$] " +TAG_FREE = "[free] " +TAG_LOCAL = "[local] " +TAGS = (TAG_PAID, TAG_FREE, TAG_LOCAL) + +OR_KEY = os.environ.get("OPENROUTER_API_KEY", "") +ZEN_HOST = "opencode.ai" +ZEN_PORT = 443 +LOG = sys.stdout + +_disc_cache = {} +_disc_lock = threading.Lock() + +_cloud_disc = {"ids": None, "expiry": 0.0} +_cloud_lock = threading.Lock() + +LOCAL_BACKENDS = [] +_backends_lock = threading.Lock() + +_free_catalog = {} +_paid_catalog = {} +_cloud_catalog_lock = threading.Lock() +_catalog_dirty = False +_model_metadata = {} # model_id -> {name, ctx, reasoning} +_backends_up = set() +_backends_up_lock = threading.Lock() +_last_known = {} # backend_name -> [model_ids] +_last_known_lock = threading.Lock() + +_checker_pending = threading.Event() +_checker_exit = threading.Event() + +# ---- admission control: protect single-slot local backends from pile-up ---- +ADMIT_TIMEOUT = 90 # max seconds a request waits for a backend slot +DEFAULT_SLOTS = 1 # local llama.cpp servers are single-slot by default +QUEUE_FACTOR = 4 # max in-system (waiting+running) per backend = slots*QUEUE_FACTOR +_backend_sems = {} # name -> BoundedSemaphore(slots) +_backend_slots = {} # name -> int +_backend_inflight = {} # name -> waiting+running count +_admit_lock = threading.Lock() + + +def _build_sems(slots_map): + global _backend_sems, _backend_slots + sems, slots = {}, {} + for name, n in slots_map.items(): + n = max(1, int(n)) + sems[name] = threading.BoundedSemaphore(n) + slots[name] = n + _backend_sems, _backend_slots = sems, slots + + +def _admit(name): + """Reserve a slot on backend `name`. True=admitted (caller must _release), + False=rejected (queue full or waited too long).""" + sem = _backend_sems.get(name) + if sem is None: + return True + cap = _backend_slots.get(name, DEFAULT_SLOTS) * QUEUE_FACTOR + with _admit_lock: + if _backend_inflight.get(name, 0) >= cap: + return False + _backend_inflight[name] = _backend_inflight.get(name, 0) + 1 + if sem.acquire(timeout=ADMIT_TIMEOUT): + return True + with _admit_lock: + _backend_inflight[name] = max(0, _backend_inflight.get(name, 0) - 1) + return False + + +def _release(name): + sem = _backend_sems.get(name) + if sem is None: + return + try: + sem.release() + except ValueError: + pass + with _admit_lock: + _backend_inflight[name] = max(0, _backend_inflight.get(name, 0) - 1) def log(msg): - sys.stdout.write(f"{time.strftime('%H:%M:%S')} {msg}\n") - sys.stdout.flush() + LOG.write(f"{time.strftime('%H:%M:%S')} {msg}\n") + LOG.flush() -# ── helpers ──────────────────────────────────────────────────────── - -def _simple_token_estimate(text): - """Rough token estimate: 4 chars per token (standard heuristic).""" - return len(text) // 4 - - -def compress_messages(messages, max_tokens=GEMMA_MAX_INPUT_TOKENS): - """Drop oldest non-system messages to stay within token budget.""" - if not messages: - return messages - - system_msgs = [m for m in messages if m.get("role") == "system"] - history = [m for m in messages if m.get("role") != "system"] - - # Count current tokens - total = sum(_simple_token_estimate(json.dumps(m)) for m in messages) - - if total <= max_tokens: - return messages - - # Drop oldest history messages, keep most recent - while history and total > max_tokens: - dropped = history.pop(0) - total -= _simple_token_estimate(json.dumps(dropped)) - log(f"compress: dropped {dropped.get('role','?')} msg ({total} est. tokens remain)") - - compressed = system_msgs + history - return compressed if compressed else messages - - -def classify_query(query): - """Send query to the local classifier service. Returns tier string.""" - body = json.dumps({"query": query}).encode() - req = urllib.request.Request( - CLASSIFIER_URL, data=body, - headers={"Content-Type": "application/json"}, - ) - try: - resp = urllib.request.urlopen(req, timeout=15) - data = json.loads(resp.read()) - return data.get("tier", "medium") - except (urllib.error.URLError, json.JSONDecodeError, OSError) as e: - log(f"classifier call failed: {e}") - return "medium" - - -def compress_via_service(messages, rate=0.4): - """Send messages to the ML compression service. Returns compressed messages or None.""" - body = json.dumps({"messages": messages, "rate": rate}).encode() - req = urllib.request.Request( - COMPRESS_URL, data=body, - headers={"Content-Type": "application/json"}, - ) - try: - resp = urllib.request.urlopen(req, timeout=5) # fast timeout, fall back to truncation - data = json.loads(resp.read()) - return data.get("compressed") - except (urllib.error.URLError, json.JSONDecodeError, OSError) as e: - log(f"compress service call failed: {e}") +def parse_param_count(text): + if not text: return None - - -def classify_local(query): - """Fallback: classify directly against Llama 3.2 1B if classifier service is unavailable.""" - url = "http://127.0.0.1:8089/v1/chat/completions" - msgs = [{"role": "system", "content": "Classify queries as low, medium, or high."}] - for q, a in FEW_SHOT_EXAMPLES: - msgs.append({"role": "user", "content": q}) - msgs.append({"role": "assistant", "content": a}) - msgs.append({"role": "user", "content": query}) - body = json.dumps({ - "model": "llama-3.2-1b-classifier", - "messages": msgs, - "max_tokens": 5, - "temperature": 0, - }).encode() - req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}) - try: - resp = urllib.request.urlopen(req, timeout=30) - data = json.loads(resp.read()) - content = data["choices"][0]["message"]["content"].strip().lower() - return content if content in ("low", "medium", "high") else "medium" - except Exception as e: - log(f"direct classify failed: {e}") - return "medium" - - -# ── model discovery ──────────────────────────────────────────────── - -def fetch_models(backend): - name, host, port = backend - try: - c = http.client.HTTPConnection(host, port, timeout=CONNECT_TIMEOUT) - c.request("GET", "/v1/models") - r = c.getresponse() - if r.status != 200: - return None - data = json.loads(r.read()) - c.close() - ids = [] - for m in data.get("data", []) or data.get("models", []): - mid = m.get("id") or m.get("name") or m.get("model") - if mid: - ids.append(mid) - return ids - except (OSError, ValueError, http.client.HTTPException): - return None - - -def discover(): - out = [] - for b in BACKENDS: - ids = fetch_models(b) - if ids is not None: - out.append((b, ids)) - return out - - -def find_backend_for_model(model_id): - for b in BACKENDS: - ids = fetch_models(b) - if ids and model_id in ids: - return b + m = re.search(r'(? now: + return cached[1] + ids = _fetch_backend_models(backend) + with _disc_lock: + _disc_cache[name] = (now + DISCOVERY_TTL, ids) + if ids is not None: + with _last_known_lock: + _last_known[name] = ids + _save_last_known() + with _backends_up_lock: + _backends_up.add(name) + else: + with _backends_up_lock: + _backends_up.discard(name) + with _last_known_lock: + ids = _last_known.get(name) + return ids + + +def _fetch_openrouter_models(): + if not OR_KEY: + return None + conn = None + try: + conn = http.client.HTTPSConnection( + OPENROUTER_HOST, OPENROUTER_PORT, + timeout=10, context=ssl.create_default_context(), + ) + conn.request("GET", "/api/v1/models", + headers={"Authorization": f"Bearer {OR_KEY}", + "Accept": "application/json"}) + resp = conn.getresponse() + if resp.status != 200: + log(f"openrouter discovery: HTTP {resp.status}") + return None + data = json.loads(resp.read()) + ids = frozenset(m["id"] for m in data.get("data", []) if "id" in m) + return ids, data.get("data", []) + except Exception as e: + log(f"openrouter discovery FAIL: {type(e).__name__}: {e}") + return None + finally: + try: + if conn: + conn.close() + except Exception: + pass + + +def discover_cloud(): + now = time.monotonic() + with _cloud_lock: + if _cloud_disc["ids"] is not None and _cloud_disc["expiry"] > now: + return _cloud_disc["ids"] + result = _fetch_openrouter_models() + with _cloud_lock: + if result is not None: + ids, _ = result + _cloud_disc["ids"] = ids + _cloud_disc["expiry"] = now + CLOUD_DISCOVERY_TTL + return _cloud_disc["ids"] + + +def find_backend_for_model(model_id): + backends = list(LOCAL_BACKENDS) + # 1. Check if it's a direct match in any backend + for b in backends: + ids = discover(b) + if ids and model_id in ids: + return b + + # 2. Check if it's a friendly name + friendly_to_backend = {v: k for k, v in FRIENDLY_NAMES.items()} + if model_id in friendly_to_backend: + target_name = friendly_to_backend[model_id] + for b in backends: + if b[0] == target_name: + ids = discover(b) + if ids: + for mid in ids: + if model_id in mid: + return b + return None + + +def _probe_model(api_key, model_id): + if not api_key: + return None + conn = None + try: + body = json.dumps({ + "model": model_id, + "messages": [{"role": "user", "content": "hi"}], + "max_tokens": 1, + "stream": False, + }).encode() + conn = http.client.HTTPSConnection( + OPENROUTER_HOST, OPENROUTER_PORT, + timeout=PROBE_TIMEOUT, + context=ssl.create_default_context(), + ) + conn.request("POST", "/api/v1/chat/completions", + body=body, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + "Accept": "application/json", + }) + resp = conn.getresponse() + resp.read() + if resp.status == 429: + return False + if resp.status == 200: + return True + return None + except Exception: + return None + finally: + try: + if conn: + conn.close() + except Exception: + pass + + +def _run_catalog_check(): + result = _fetch_openrouter_models() + if result is None: + log("check: openrouter fetch failed, skipping cycle") + return + _, raw_models = result + + # --- Free catalog (auto-discovered :free models) --- + free_ids = {} + paid_info = {} + new_metadata = {} + for m in raw_models: + mid = m["id"] + name = m.get("name", "") + ctx = m.get("context_length", 0) + pricing = m.get("pricing", {}) + supported_params = m.get("supported_parameters", []) + has_reasoning = "reasoning" in supported_params + meta_entry = {"name": name, "ctx": ctx, "reasoning": has_reasoning} + new_metadata[mid] = meta_entry + if mid.endswith(":free"): + free_ids[mid] = {"name": name, "ctx": ctx} + elif any(mid.startswith(p) for p in CLOUD_PREFIXES): + prompt_price = float(pricing.get("prompt", 999)) + if prompt_price <= PAID_DISCOVER_THRESHOLD: + paid_info[mid] = {"name": name, "ctx": ctx, "prompt_price": prompt_price} + + current_free = dict(_free_catalog) + new_free = {} + changes = [] + for mid, info in free_ids.items(): + param = parse_param_count(mid) or parse_param_count(info["name"]) + if param is not None and param <= 7.0: + continue + if param is None and info["ctx"] < 32768: + continue + status = current_free.get(mid, "ok") + new_free[mid] = status + if mid not in current_free: + changes.append(f"+{mid} (param={param}, ctx={info['ctx']})") + + for mid in current_free: + if mid not in new_free: + changes.append(f"-{mid}") + + # --- Paid catalog (auto-discover cheap known-prefix models) --- + current_paid = dict(_paid_catalog) + new_paid = dict(current_paid) # preserve manually-added paid entries + paid_changes = [] + for mid, info in paid_info.items(): + if mid not in new_paid: + paid_changes.append(f"+{mid} (${info['prompt_price']}/MTok)") + new_paid[mid] = "ok" + + if changes or paid_changes: + log(f"check: free changes: {', '.join(changes) if changes else 'none'}; " + f"paid changes: {', '.join(paid_changes) if paid_changes else 'none'}") + else: + log("check: no catalog changes") + + # Probe free models for rate-limit status + for mid in list(new_free.keys()): + if not OR_KEY: + break + old_status = new_free[mid] + available = _probe_model(OR_KEY, mid) + if available is False: + new_free[mid] = "na" + if old_status != "na": + log(f"check: {mid} -> RATE-LIMITED (na)") + elif available is True: + new_free[mid] = "ok" + if old_status != "ok": + log(f"check: {mid} -> recovered (ok)") + + with _cloud_catalog_lock: + _free_catalog.clear() + _free_catalog.update(new_free) + _paid_catalog.clear() + _paid_catalog.update(new_paid) + _model_metadata.clear() + _model_metadata.update(new_metadata) + _save_catalog() + n_na = sum(1 for v in new_free.values() if v == "na") + log(f"check: done ({len(new_free)} free/{n_na} na, {len(new_paid)} paid)") + + +def _checker_loop(): + _run_catalog_check() + while not _checker_exit.is_set(): + triggered = _checker_pending.wait(timeout=CHECK_INTERVAL) + if _checker_exit.is_set(): + break + if triggered: + _checker_pending.clear() + _run_catalog_check() + + +def _is_in_any_catalog(model): + with _cloud_catalog_lock: + return model in _free_catalog or model in _paid_catalog + + +def _get_model_status(model): + with _cloud_catalog_lock: + st = _free_catalog.get(model) + if st is not None: + return st + return _paid_catalog.get(model) + + +def _merge_catalogs(): + with _cloud_catalog_lock: + return {**_free_catalog, **_paid_catalog} + class Handler(http.server.BaseHTTPRequestHandler): - server_version = "boltzmann-llm-proxy/3" - def _read_body(self): - n = int(self.headers.get("Content-Length", 0)) - return self.rfile.read(n) if n else b"" + length = int(self.headers.get("Content-Length", 0)) + return self.rfile.read(length) if length else b"" - def _json_body(self): - try: - return json.loads(self._read_body()) - except (ValueError, TypeError): - return {} - - def _fwd_headers(self, body_len): + def _fwd_headers_local(self, body_len): h = {} for k in ("Content-Type", "Authorization", "Accept", "Accept-Encoding"): v = self.headers.get(k) @@ -207,173 +599,446 @@ class Handler(http.server.BaseHTTPRequestHandler): h["Content-Length"] = str(body_len) return h - def _json_response(self, code, data): - payload = json.dumps(data).encode() - self.send_response(code) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(payload))) - self.end_headers() - self.wfile.write(payload) + def _fwd_headers_cloud(self, body_len): + h = { + "Content-Type": self.headers.get("Content-Type", "application/json"), + "Authorization": f"Bearer {OR_KEY}", + "Accept": self.headers.get("Accept", "application/json"), + "Accept-Encoding": "identity", + "HTTP-Referer": "http://hossenfelder.lxd", + "X-Title": "hossenfelder", + } + if body_len: + h["Content-Length"] = str(body_len) + return h - def _proxy_to(self, backend, method, body, extra_headers=None): - name, host, port = backend - try: - headers = self._fwd_headers(len(body)) - if extra_headers: - headers.update(extra_headers) - c = http.client.HTTPConnection(host, port, timeout=READ_TIMEOUT) - c.request(method, self.path, body=body, headers=headers) - r = c.getresponse() - data = r.read() - self.send_response(r.status) - for hname in ("Content-Type", "Content-Encoding", "Cache-Control", "ETag", "Last-Modified"): - v = r.getheader(hname) + def _stream_response(self, resp, backend_name, t0, t_try, hint, method, path, is_streaming=False): + t_hdr = time.monotonic() + self.send_response(resp.status) + content_type = resp.getheader("Content-Type", "").lower() + # For SSE/streaming responses, NEVER forward Content-Length — + # it breaks chunked framing and causes clients to hang waiting + # for exactly N bytes that arrive incrementally. + if is_streaming or "text/event-stream" in content_type: + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.send_header("Connection", "close") + # Don't forward Content-Length or Transfer-Encoding + else: + for hname in ("Content-Type", "Content-Encoding", "Content-Length"): + v = resp.getheader(hname) if v: self.send_header(hname, v) - self.send_header("Content-Length", str(len(data))) - self.send_header("X-Backend", name) - self.end_headers() - self.wfile.write(data) - c.close() - log(f"{method} {self.path} -> {name} ({r.status})") - except (OSError, http.client.HTTPException) as e: - self.send_response(502) + self.send_header("X-LLM-Backend", backend_name) + self.end_headers() + bytes_out = 0 + try: + while True: + chunk = resp.read(65536) + if not chunk: + break + self.wfile.write(chunk) + self.wfile.flush() + bytes_out += len(chunk) + except (BrokenPipeError, ConnectionResetError): + pass # client disconnected, stop forwarding + except http.client.IncompleteRead as e: + if e.partial: + self.wfile.write(e.partial) + self.wfile.flush() + bytes_out += len(e.partial) + dt_total = time.monotonic() - t0 + dt_ttfb = t_hdr - t_try + log(f"{method} {path} -> {backend_name} status={resp.status} ttfb={dt_ttfb:.2f}s total={dt_total:.2f}s bytes={bytes_out} stream={is_streaming}{hint}") + + def _send_error(self, status, body): + try: + self.send_response(status) self.send_header("Content-Type", "application/json") - self.send_header("X-Backend", name) + self.send_header("Content-Length", str(len(body))) self.end_headers() - self.wfile.write(json.dumps({"error": f"{name} unreachable: {e}"}).encode()) - log(f"{method} {self.path} -> {name} FAILED ({e})") + self.wfile.write(body) + except (BrokenPipeError, ConnectionResetError): + pass - # ── GET ── - - def do_GET(self): - path = self.path.rstrip("/") - if path == "/v1/models": - merged = [] - seen = set() - for backend, ids in discover(): - name = backend[0] - for mid in ids: - if mid in seen: - continue - seen.add(mid) - merged.append({ - "id": mid, - "object": "model", - "owned_by": f"boltzmann/{name}", - }) - self._json_response(200, {"object": "list", "data": merged}) - log(f"GET /v1/models -> aggregator ({len(merged)} models)") - return - - if path == "/health": - self._json_response(200, {"status": "ok", "version": 3}) - return - - for b in BACKENDS: - if fetch_models(b) is not None: - return self._proxy_to(b, "GET", b"") - self._json_response(502, {"error": "no backend up"}) - - # ── POST ── - - def do_POST(self): - path = self.path.rstrip("/") - - # ── Classification endpoint ── - if path == "/v1/classify": - body = self._json_body() - query = body.get("query", "") - if not query: - return self._json_response(400, {"error": "missing query"}) - tier = classify_local(query) - log(f"classify {query[:60]!r} -> {tier}") - return self._json_response(200, {"tier": tier}) - - # ── Compression endpoint ── - if path == "/v1/compress": - body = self._json_body() - messages = body.get("messages", []) - max_tokens = body.get("max_tokens", GEMMA_MAX_INPUT_TOKENS) - compressed = compress_messages(messages, max_tokens) - ratio = 0 - if messages: - orig = sum(_simple_token_estimate(json.dumps(m)) for m in messages) - new = sum(_simple_token_estimate(json.dumps(m)) for m in compressed) - ratio = round((1 - new / orig) * 100, 1) if orig else 0 - return self._json_response(200, { - "messages": compressed, - "original_tokens": orig if messages else 0, - "compressed_tokens": sum(_simple_token_estimate(json.dumps(m)) for m in compressed), - "ratio_pct": ratio, - }) - - # ── Chat / Completions ── - if path in ("/v1/chat/completions", "/v1/completions"): - body = self._read_body() - model = None + def _try_backend(self, backend, method, path, body, fwd, t0, hint, is_last, is_streaming=False): + name, host, port = backend + if not _admit(name): + log(f"{method} {path} -> {name} BUSY (admission gate, slots={_backend_slots.get(name)}) total={time.monotonic()-t0:.2f}s") + return "busy" + t_try = time.monotonic() + conn = None + try: + conn = http.client.HTTPConnection(host, port, timeout=CONNECT_TIMEOUT if not is_last else READ_TIMEOUT) + conn.request(method, path, body=body, headers=fwd) + resp = conn.getresponse() + self._stream_response(resp, name, t0, t_try, hint, method, path, is_streaming=is_streaming) + conn.close() + return "ok" + except (ConnectionRefusedError, TimeoutError, OSError) as e: + dt = time.monotonic() - t_try + log(f"{method} {path} -> {name} FAIL {type(e).__name__} {dt:.2f}s") try: - model = json.loads(body).get("model") - except (ValueError, TypeError): + if conn: + conn.close() + except Exception: + pass + return "down" + finally: + _release(name) + + def _proxy_local(self, method, path, body, hint, t0, model=None, is_streaming=False): + fwd = self._fwd_headers_local(len(body)) + backends = list(LOCAL_BACKENDS) + + chosen = None + if model: + chosen = find_backend_for_model(model) + if chosen is None: + available = set() + for b in backends: + for m in (discover(b) or []): + available.add(m) + payload = json.dumps({"error": { + "message": f"model '{model}' not available on any local backend", + "type": "model_not_found", + "available": sorted(available), + }}).encode() + self._send_error(404, payload) + log(f"{method} {path} -> model={model!r} 404 (not in any local backend) total={time.monotonic()-t0:.2f}s") + return + if chosen is not None: + r = self._try_backend(chosen, method, path, body, fwd, t0, hint + " (routed)", is_last=True, is_streaming=is_streaming) + if r == "ok": + return + if r == "busy": + self._send_error(503, b'{"error":{"message":"backend busy (admission gate), retry shortly","type":"backend_busy"}}') + return + log(f"{method} {path} -> {chosen[0]} routed-FAIL, falling back to legacy chain") + + chain = [b for b in backends if b is not chosen] + busy_seen = False + for i, b in enumerate(chain): + is_last = (i == len(chain) - 1) + r = self._try_backend(b, method, path, body, fwd, t0, hint, is_last, is_streaming=is_streaming) + if r == "ok": + return + if r == "busy": + busy_seen = True + if busy_seen: + self._send_error(503, b'{"error":{"message":"all local backends busy (admission gate), retry shortly","type":"backend_busy"}}') + log(f"{method} {path} -> ALL_BUSY total={time.monotonic()-t0:.2f}s") + else: + self._send_error(502, b'{"error":"all local backends down"}') + log(f"{method} {path} -> ALL_DOWN total={time.monotonic()-t0:.2f}s") + + def _proxy_zen(self, method, path, body, hint, t0, is_streaming=False): + upstream_path = "/zen/v1/chat/completions" + try: + j = json.loads(body) + raw_model = j.get("model", "") + if raw_model.startswith("opencode/"): + j["model"] = raw_model[len("opencode/"):] + for key in ("max_tokens", "max_completion_tokens"): + val = j.get(key) + if val is not None and (not isinstance(val, int) or val < 1): + j.pop(key, None) + # Zen API does not support "developer" role; map to "system" + for msg in j.get("messages", []): + if msg.get("role") == "developer": + msg["role"] = "system" + body = json.dumps(j).encode() + except Exception: + pass + fwd = { + "Content-Type": "application/json", + "Accept": "application/json", + "Accept-Encoding": "identity", + } + t_try = time.monotonic() + conn = None + try: + conn = http.client.HTTPSConnection( + ZEN_HOST, ZEN_PORT, + timeout=READ_TIMEOUT, + context=ssl.create_default_context(), + ) + conn.request(method, upstream_path, body=body, headers=fwd) + resp = conn.getresponse() + if resp.status >= 400: + resp_body = resp.read() + sent = body[:500].decode("utf-8", errors="replace") + log(f"ZEN 400: status={resp.status} body={resp_body[:300]} sent={sent}") + self._send_error(502, resp_body) + log(f"{method} {path} -> opencode-zen status={resp.status} ttfb={time.monotonic()-t_try:.2f}s") + conn.close() + return + self._stream_response(resp, "opencode-zen", t0, t_try, hint, method, path, is_streaming=is_streaming) + conn.close() + except Exception as e: + err = json.dumps({"error": f"opencode-zen: {type(e).__name__}: {e}"}).encode() + self._send_error(502, err) + log(f"{method} {path} -> opencode-zen FAIL {type(e).__name__}: {e}") + try: + if conn: + conn.close() + except Exception: pass - target = None - if model: - target = find_backend_for_model(model) - if target is None: - available = sorted({m for b in BACKENDS for m in (fetch_models(b) or [])}) - return self._json_response(404, {"error": { - "message": f"model '{model}' not available on this aggregator", - "type": "model_not_found", - "available": available, - }}) + def _proxy_cloud(self, method, path, body, hint, t0, is_streaming=False): + if not OR_KEY: + self._send_error(500, b'{"error":"OPENROUTER_API_KEY not set on hossenfelder"}') + log(f"{method} {path} -> openrouter NO_KEY") + return + upstream_path = "/api" + path if path.startswith("/v1") else path + fwd = self._fwd_headers_cloud(len(body)) + t_try = time.monotonic() + conn = None + try: + conn = http.client.HTTPSConnection( + OPENROUTER_HOST, OPENROUTER_PORT, + timeout=READ_TIMEOUT, + context=ssl.create_default_context(), + ) + conn.request(method, upstream_path, body=body, headers=fwd) + resp = conn.getresponse() + self._stream_response(resp, "openrouter", t0, t_try, hint, method, path, is_streaming=is_streaming) + conn.close() + except Exception as e: + err = json.dumps({"error": f"openrouter: {type(e).__name__}: {e}"}).encode() + self._send_error(502, err) + log(f"{method} {path} -> openrouter FAIL {type(e).__name__}: {e}") + try: + if conn: + conn.close() + except Exception: + pass - if target is None: - for b in BACKENDS: - if fetch_models(b) is not None: - target = b - break - if target is None: - return self._json_response(502, {"error": "no backend up"}) + def _handle_models(self, t0): + backends = list(LOCAL_BACKENDS) + local_entries = [] + seen_local = set() + up_count = 0 + for backend in backends: + name = backend[0] + ids = discover(backend) + if ids is None: + continue + is_down = name not in _backends_up + if is_down: + ids = list(ids) + if not is_down: + up_count += 1 + # Filter out bogus deepseek-v4-pro alias on ds4-escher + if name == "ds4-escher": + ids = [m for m in ids if m != "deepseek-v4-pro"] + for mid in ids: + if mid in seen_local: + continue + if is_down: + display_id = mid + DOWN_SUFFIX + owned_by = f"local-{name} (temporarily unavailable)" + else: + display_id = mid + owned_by = "local-" + name + seen_local.add(mid) + local_entries.append({ + "id": TAG_LOCAL + display_id, + "object": "model", + "owned_by": owned_by, + }) - # Compression: try ML service (fast timeout), fall back to token-budget truncation - compressed_body = body - if path == "/v1/chat/completions" and "gemma4" in (model or ""): + live_or = discover_cloud() + catalog_snapshot = _merge_catalogs() + + cloud_models = [] + with _cloud_catalog_lock: + meta_snapshot = dict(_model_metadata) + for mid, status in catalog_snapshot.items(): + if live_or is not None and mid not in live_or: + continue + display = mid + if status == "na": + base, suffix = mid.rsplit(":", 1) if ":" in mid else (mid, "") + display = f"{base}-na:{suffix}" if suffix else f"{mid}{NA_SUFFIX}" + if mid.endswith(":free"): + tag = TAG_FREE + owned = "openrouter-free" + display = display[:-5] if display.endswith(":free") else display + else: + tag = TAG_PAID + owned = "openrouter-paid" + entry = { + "id": tag + display, + "object": "model", + "owned_by": owned, + } + meta_info = meta_snapshot.get(mid, {}) + if "reasoning" in meta_info: + entry["reasoning"] = meta_info["reasoning"] + cloud_models.append(entry) + + # OpenCode Zen models + cloud_models.append({ + "id": TAG_FREE + "opencode/big-pickle", + "object": "model", + "owned_by": "opencode-zen", + "reasoning": True, + }) + + n_na = sum(1 for v in catalog_snapshot.values() if v == "na") + merged = {"object": "list", "data": cloud_models + local_entries} + body = json.dumps(merged).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + or_state = "live" if live_or is not None else "cached/down" + log(f"GET /v1/models -> merged cloud={len(cloud_models)} (na={n_na}, OR={or_state}) local={len(local_entries)} (from {up_count} backends) total={time.monotonic()-t0:.2f}s") + + def _proxy(self, method): + t0 = time.monotonic() + path = self.path + try: + if method == "GET" and path in ("/health", "/v1/health"): + resp_body = json.dumps({"status":"ok"}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(resp_body))) + self.end_headers() + self.wfile.write(resp_body) + return + + if method == "GET" and path in ("/v1/models", "/models"): + self._handle_models(t0) + return + + body = self._read_body() + hint = "" + model = "" + stream = False + if body and b"messages" in body: try: - req = json.loads(body) - if "messages" in req and len(req["messages"]) > 2: - orig_n = len(req["messages"]) - orig_t = sum(_simple_token_estimate(json.dumps(m)) for m in req["messages"]) - compressed = compress_via_service(req["messages"], rate=0.4) - if compressed: - req["messages"] = compressed - compressed_body = json.dumps(req).encode() - new_t = sum(_simple_token_estimate(json.dumps(m)) for m in compressed) - log(f"ML compress {model}: ~{orig_t}->{new_t} tokens") - else: - truncated = compress_messages(req["messages"], GEMMA_MAX_INPUT_TOKENS) - if len(truncated) < len(req["messages"]): - req["messages"] = truncated - compressed_body = json.dumps(req).encode() - log(f"truncate compress {model}: {orig_n}->{len(truncated)} msgs") - except (ValueError, TypeError, json.JSONDecodeError) as e: - log(f"compression skipped (parse error): {e}") + j = json.loads(body) + raw_model = j.get("model", "") + model = strip_tag(raw_model) + model = re.sub(r"\-na(?=:|\$)", "", model) + # Restore :free suffix for free-tier models (catalog keys include it) + if ":" not in model: + with _cloud_catalog_lock: + if model + ":free" in _free_catalog: + model = model + ":free" + if model != raw_model: + j["model"] = model + body = json.dumps(j).encode() + n_msgs = len(j.get("messages", [])) + max_t = j.get("max_tokens") or j.get("max_completion_tokens") or "-" + total_chars = sum(len(str(m.get("content", ""))) for m in j.get("messages", [])) + stream = j.get("stream", False) + hint = f" model={model} msgs={n_msgs} in_chars={total_chars} max_tok={max_t} stream={stream}" + except Exception: + pass - return self._proxy_to(target, "POST", compressed_body) + # /v1/compress: proxy to compress-server backend + if method in ("POST",) and path in ("/v1/compress", "/compress"): + compress_backend = None + for b in LOCAL_BACKENDS: + if b[0] == "compress": + compress_backend = b + break + if compress_backend: + fwd = self._fwd_headers_local(len(body)) + name, host, port = compress_backend + t_try = time.monotonic() + conn = None + try: + conn = http.client.HTTPConnection(host, port, timeout=600) + conn.request("POST", "/compress", body=body, headers=fwd) + resp = conn.getresponse() + data = resp.read() + self.send_response(resp.status) + for hn in ("Content-Type",): + v = resp.getheader(hn) + if v: + self.send_header(hn, v) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + conn.close() + log(f"POST /v1/compress -> compress-server status={resp.status} total={time.monotonic()-t0:.2f}s") + return + except Exception as e: + log(f"POST /v1/compress -> compress-server FAIL: {e}") + self._send_error(502, b'{"error":"compress backend unavailable"}') + return - # ── Anything else: forward to default backend ── - body = self._read_body() - for b in BACKENDS: - if fetch_models(b) is not None: - return self._proxy_to(b, "POST" if body else "GET", body) - self._json_response(502, {"error": "no backend up"}) + if model and model.startswith("opencode/"): + self._proxy_zen(method, path, body, hint, t0, is_streaming=stream) + return - def log_message(self, fmt, *args): - return + if model and (is_cloud_model(model) or _is_in_any_catalog(model)): + is_na = _get_model_status(model) == "na" + if is_na: + payload = json.dumps({"error": { + "message": f"model '{model}' is rate-limited and currently not available", + "type": "rate_limited", + }}).encode() + self._send_error(429, payload) + log(f"{method} {path} -> model={model!r} 429 (na) total={time.monotonic()-t0:.2f}s") + return + self._proxy_cloud(method, path, body, hint, t0, is_streaming=stream) + else: + self._proxy_local(method, path, body, hint, t0, model=model or None, is_streaming=stream) + except BrokenPipeError: + log(f"{method} {path} client-disconnect total={time.monotonic()-t0:.2f}s") + except Exception as e: + log(f"{method} {path} EXC {type(e).__name__}: {e}") + traceback.print_exc(file=LOG) + try: + self._send_error(500, b'{"error":"internal"}') + except Exception: + pass + + def do_POST(self): self._proxy("POST") + def do_GET(self): self._proxy("GET") + def do_OPTIONS(self): + self.send_response(200) + self.send_header("Access-Control-Allow-Origin", "*") + self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + self.send_header("Access-Control-Allow-Headers", "*") + self.end_headers() + + def log_message(self, fmt, *args): pass if __name__ == "__main__": - log(f"boltzmann-llm-proxy v3 on :{LISTEN_PORT}") - log(f" classifier: {CLASSIFIER_URL}") - log(f" backends: {', '.join(n for n, h, p in BACKENDS)}") - http.server.ThreadingHTTPServer(("0.0.0.0", LISTEN_PORT), Handler).serve_forever() + port = int(sys.argv[1]) if len(sys.argv) > 1 else 8082 + + free_cat, paid_cat, meta = _load_catalog() + _free_catalog.clear() + _free_catalog.update(free_cat) + _paid_catalog.clear() + _paid_catalog.update(paid_cat) + _model_metadata.clear() + _model_metadata.update(meta) + _last_known = _load_last_known() + + reload_backends(None, None) # populates global LOCAL_BACKENDS + signal.signal(signal.SIGHUP, reload_backends) + signal.signal(signal.SIGUSR1, schedule_recheck) + + _checker_pending.clear() + _checker_exit.clear() + checker = threading.Thread(target=_checker_loop, daemon=True, name="catalog-checker") + checker.start() + + server = http.server.ThreadingHTTPServer(("0.0.0.0", port), Handler) + n_free = len(_free_catalog) + n_paid = len(_paid_catalog) + n_na = sum(1 for v in _free_catalog.values() if v == "na") + log(f"llm-proxy starting on :{port} local={[b[0] for b in LOCAL_BACKENDS]} " + f"cloud={'enabled' if OR_KEY else 'DISABLED (no key)'} " + f"catalog={n_free} free (na={n_na}) + {n_paid} paid " + f"routing=model-aware+first-up-fallback+live-cloud-filter+auto-check") + server.serve_forever() diff --git a/local-backends.json b/local-backends.json new file mode 100644 index 0000000..a3ed392 --- /dev/null +++ b/local-backends.json @@ -0,0 +1,4 @@ +[ + {"name":"qwen3-4b","host":"boltzmann.fritz.box","port":8091,"slots":1}, + {"name":"bosch-dspark","host":"bosch.fritz.box","port":8888,"slots":32} +]