""" Local model router (stdlib-only; runs on any Python 3.9+, incl. 3.14). An OpenAI-compatible endpoint that picks the best local model per request and forwards straight to Ollama's own OpenAI API. Point every client at http://localhost:8080/v1 - model "auto" -> router decides (modality -> heuristics -> tiny classifier) - any real name -> passed through unchanged (manual override always wins) On upstream failure the router retries once with a mapped fallback model. Response headers x-router-model / x-router-decided-by report the pick. """ import os, re, json, time, base64, urllib.request, urllib.error, urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer OLLAMA = os.environ.get("ROUTER_OLLAMA", "http://127.0.0.1:11434").rstrip("/") # client target; NOT OLLAMA_HOST (that's the server bind addr, e.g. 0.0.0.0) UPSTREAM = os.environ.get("ROUTER_UPSTREAM", OLLAMA + "/v1").rstrip("/") HOST = os.environ.get("ROUTER_HOST", "0.0.0.0") # 0.0.0.0 = reachable on the LAN PORT = int(os.environ.get("ROUTER_PORT", "8080")) API_KEY = os.environ.get("ROUTER_API_KEY", "").strip() # if set, require Bearer token AUDIO_MODEL = os.environ.get("ROUTER_AUDIO_MODEL", "gemma4:e4b") # audio only works on official Gemma 4 # ComfyUI image generation (OpenAI /v1/images/generations -> ComfyUI submit/poll/fetch) COMFYUI = os.environ.get("ROUTER_COMFYUI", "http://127.0.0.1:8188").rstrip("/") COMFY_WORKFLOW = os.environ.get("ROUTER_COMFY_WORKFLOW", os.path.join(os.path.dirname(os.path.abspath(__file__)), "comfyui-workflow.json")) COMFY_CHECKPOINT = os.environ.get("ROUTER_COMFY_CHECKPOINT", "") # your ComfyUI checkpoint filename M = { "fast": "qwen3:8b", "general": "qwen3:14b", "code": "glm-4.7-flash", "code_heavy": "qwen3-coder:30b", "reason": "qwen3.6:35b-a3b", "agentic": "glm-4.7-flash", # tool-carrying requests (general agent): best hard-agentic (7/7, only one to recover from tool errors) "vision": "qwen3-vl:8b", "vision_heavy": "qwen3-vl:30b-a3b-instruct", } CLASSIFIER = "qwen3:8b" SHORT_CHARS = 240 # Uncensored fleet. A request routes here if it asks for model "auto-uncensored" # OR carries the dedicated famapp key (ROUTER_UNCENSORED_KEY) — in the latter case # EVERY request from that client is forced uncensored regardless of model asked for. UNCENSORED = { "general": "gemma4-uncensored:26b", "vision": "gemma4-uncensored:26b", # has vision "code": "glm-uncensored-code:30b", "fast": "gemma-e4b-uncensored", } UNCENSORED_KEY = os.environ.get("ROUTER_UNCENSORED_KEY", "").strip() FALLBACK = { "glm-4.7-flash": "qwen3-coder:30b", "qwen3-coder:30b": "glm-4.7-flash", "gemma4:26b": "qwen3.6:35b-a3b", "qwen3.6:35b-a3b": "gpt-oss:20b", "gpt-oss:20b": "qwen3:14b", "qwen3-vl:30b-a3b-instruct": "qwen3-vl:8b", "qwen3:14b": "qwen3:8b", "qwen3-vl:8b": "gemma4:12b", # uncensored models still downloading fall back to the one that's ready "glm-uncensored-code:30b": "gemma4-uncensored:26b", "gemma-e4b-uncensored": "gemma4-uncensored:26b", } CODE_RE = re.compile(r"```|\b(def |class |function |import |#include|refactor|debug|" r"stack ?trace|traceback|compile|unit test|npm |pip |git |bug|" r"exception|null pointer|segfault)\b|\.(py|js|ts|tsx|go|rs|java|" r"cpp|cc|c|rb|sh|sql)\b", re.I) CODE_HEAVY_RE = re.compile(r"\b(repo|repository|codebase|multiple files|entire project|" r"across files|whole project)\b", re.I) REASON_RE = re.compile(r"\b(prove|theorem|step[- ]by[- ]step|reason through|derive|" r"calculate|logic puzzle|why does|explain why|work through)\b", re.I) OCR_RE = re.compile(r"\b(ocr|document|invoice|receipt|table|form|handwrit|extract text|" r"scan|parse this (image|page|pdf))\b", re.I) def last_user_text(messages): for m in reversed(messages): if m.get("role") == "user": c = m.get("content") if isinstance(c, str): return c if isinstance(c, list): return " ".join(p.get("text", "") for p in c if isinstance(p, dict) and p.get("type") == "text") return "" def has_image(messages): for m in messages: c = m.get("content") if isinstance(c, list): for p in c: if isinstance(p, dict) and p.get("type") in ("image_url", "input_image", "image"): return True return False def has_audio(messages): for m in messages: c = m.get("content") if isinstance(c, list): for p in c: if isinstance(p, dict) and p.get("type") in ("input_audio", "audio"): return True return False def to_native_messages(messages): # OpenAI content -> Ollama native: text flattened, audio/image base64 into `images` out = [] for m in messages: role, c = m.get("role", "user"), m.get("content") if isinstance(c, str): out.append({"role": role, "content": c}) continue text_parts, media = [], [] if isinstance(c, list): for p in c: if not isinstance(p, dict): continue t = p.get("type") if t == "text": text_parts.append(p.get("text", "")) elif t in ("input_audio", "audio"): d = (p.get("input_audio") or p.get("audio") or {}).get("data") if d: media.append(d) elif t in ("image_url", "image", "input_image"): url = (p.get("image_url") or {}).get("url", "") if isinstance(p.get("image_url"), dict) else "" media.append(url.split(",", 1)[-1] if url.startswith("data:") else url) msg = {"role": role, "content": " ".join(tp for tp in text_parts if tp)} if media: msg["images"] = media out.append(msg) return out def parse_multipart(body, boundary): # minimal multipart/form-data parser -> {name: (filename_or_None, bytes)} fields = {} for part in body.split(b"--" + boundary.encode()): if part in (b"", b"--", b"--\r\n", b"\r\n"): continue if part.startswith(b"\r\n"): part = part[2:] if b"\r\n\r\n" not in part: continue head, _, data = part.partition(b"\r\n\r\n") if data.endswith(b"\r\n"): data = data[:-2] head_text = head.decode("utf-8", "replace") nm = re.search(r'name="([^"]*)"', head_text) if not nm: continue fm = re.search(r'filename="([^"]*)"', head_text) fields[nm.group(1)] = (fm.group(1) if fm else None, data) return fields def fill_workflow(node, subs): # recursively replace string sentinels ("{{PROMPT}}" etc.) with typed values if isinstance(node, dict): return {k: fill_workflow(v, subs) for k, v in node.items()} if isinstance(node, list): return [fill_workflow(x, subs) for x in node] if isinstance(node, str) and node in subs: return subs[node] return node def total_len(messages): n = 0 for m in messages: c = m.get("content") if isinstance(c, str): n += len(c) elif isinstance(c, list): for p in c: if isinstance(p, dict): n += len(p.get("text", "")) return n def classify(text): payload = json.dumps({ "model": CLASSIFIER, "prompt": ("Classify the user's task into exactly one label from: " "code, reason, vision, general. Reply with only the label.\n\n" "Task: " + text[:2000]), "stream": False, "keep_alive": -1, "options": {"temperature": 0, "num_predict": 3}, }).encode() try: req = urllib.request.Request(OLLAMA + "/api/generate", data=payload, headers={"content-type": "application/json"}) with urllib.request.urlopen(req, timeout=30) as r: out = json.loads(r.read()).get("response", "").strip().lower() for lab in ("code", "reason", "vision", "general"): if lab in out: return lab except Exception: pass return "general" def choose_model(body): messages = body.get("messages", []) text = last_user_text(messages) if has_image(messages): return M["vision_heavy"] if OCR_RE.search(text) else M["vision"] if body.get("tools"): # tool-carrying -> agentic (code agent vs general agent) return M["code_heavy"] if CODE_RE.search(text) else M["agentic"] if CODE_RE.search(text): return M["code_heavy"] if CODE_HEAVY_RE.search(text) else M["code"] if REASON_RE.search(text): return M["reason"] if total_len(messages) < SHORT_CHARS: return M["fast"] lab = classify(text) return {"code": M["code"], "reason": M["reason"], "vision": M["vision"], "general": M["general"]}[lab] def choose_uncensored(body): # same content heuristics, but only ever selects from the uncensored fleet messages = body.get("messages", []) text = last_user_text(messages) if has_image(messages): return UNCENSORED["vision"] if CODE_RE.search(text): return UNCENSORED["code"] if total_len(messages) < SHORT_CHARS: return UNCENSORED["fast"] return UNCENSORED["general"] def _open_upstream(body): data = json.dumps(body).encode() req = urllib.request.Request(UPSTREAM + "/chat/completions", data=data, headers={"content-type": "application/json"}) return urllib.request.urlopen(req, timeout=600) class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt, *args): # access log for EVERY request (incl 401/404) ua = self.headers.get("user-agent", "-")[:40] auth = "auth" if self.headers.get("authorization") else "noauth" print(f"[access] {time.strftime('%H:%M:%S')} {self.address_string()} " f"{auth} \"{ua}\" {fmt % args}", flush=True) def _authed(self): if not API_KEY: return True auth = self.headers.get("authorization", "") return auth.startswith("Bearer ") and auth[7:].strip() == API_KEY def _json(self, code, obj, extra=None): payload = json.dumps(obj).encode() self.send_response(code) self.send_header("content-type", "application/json") self.send_header("content-length", str(len(payload))) for k, v in (extra or {}).items(): self.send_header(k, v) self.end_headers() self.wfile.write(payload) def _text(self, code, s): data = s.encode() self.send_response(code) self.send_header("content-type", "text/plain") self.send_header("content-length", str(len(data))) self.end_headers() self.wfile.write(data) def _proxy_ollama(self, path, data=None): # read-only native Ollama endpoints for `ollama launch` probing. # NOTE: only metadata paths are wired up; inference stays on authed /v1/*. req = urllib.request.Request(OLLAMA + path, data=data, headers={"content-type": "application/json"}, method="POST" if data is not None else "GET") try: up = urllib.request.urlopen(req, timeout=30) body, code = up.read(), 200 ct = up.headers.get("content-type", "application/json") up.close() except urllib.error.HTTPError as e: body, code, ct = e.read(), e.code, "application/json" except Exception as e: return self._json(502, {"error": f"ollama unreachable: {e}"}) self.send_response(code) self.send_header("content-type", ct) self.send_header("content-length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): if self.path.rstrip("/") == "/healthz": return self._json(200, {"ok": True}) if self.path.rstrip("/") == "/v1/models": if not self._authed(): return self._json(401, {"error": "unauthorized"}) seen = list(dict.fromkeys(M.values())) return self._json(200, {"object": "list", "data": [{"id": "auto", "object": "model"}] + [{"id": m, "object": "model"} for m in seen]}) # native Ollama probe endpoints (read-only) so `ollama launch` accepts this host if self.path in ("/", ""): return self._text(200, "Ollama is running") if self.path.rstrip("/") in ("/api/version", "/api/tags", "/api/ps"): return self._proxy_ollama(self.path) return self._json(404, {"error": "not found"}) def _proxy_anthropic(self, path): # transparent proxy for Anthropic Messages API (Claude Code) -> LiteLLM /v1/messages if not self._authed(): return self._json(401, {"error": "unauthorized"}) try: length = int(self.headers.get("content-length", 0) or 0) body = self.rfile.read(length) except Exception as e: return self._json(400, {"error": f"bad request: {e}"}) url = UPSTREAM + path[len("/v1"):] # /v1/messages -> /v1/messages headers = {"content-type": "application/json"} for h in ("authorization", "anthropic-version", "anthropic-beta"): v = self.headers.get(h) if v: headers[h] = v try: up = urllib.request.urlopen( urllib.request.Request(url, data=body, headers=headers), timeout=600) except urllib.error.HTTPError as e: data = e.read() self.send_response(e.code) self.send_header("content-type", e.headers.get("content-type", "application/json")) self.send_header("content-length", str(len(data))) self.end_headers() self.wfile.write(data) return except Exception as e: return self._json(502, {"error": f"upstream unreachable: {e}"}) ctype = up.headers.get("content-type", "application/json") if "event-stream" in ctype: self.send_response(200) self.send_header("content-type", ctype) self.send_header("cache-control", "no-cache") self.send_header("connection", "close") self.end_headers() try: while True: chunk = up.read(2048) if not chunk: break self.wfile.write(chunk) self.wfile.flush() except Exception: pass finally: up.close() else: data = up.read() up.close() self.send_response(200) self.send_header("content-type", ctype) self.send_header("content-length", str(len(data))) self.end_headers() self.wfile.write(data) def _handle_transcription(self, translate=False): # OpenAI /v1/audio/transcriptions (Whisper-shaped) backed by Gemma 4 ct = self.headers.get("content-type", "") m = re.search(r"boundary=([^;]+)", ct) if "multipart/form-data" not in ct or not m: return self._json(400, {"error": "expected multipart/form-data with a boundary"}) boundary = m.group(1).strip().strip('"') length = int(self.headers.get("content-length", 0) or 0) fields = parse_multipart(self.rfile.read(length), boundary) if "file" not in fields or fields["file"][1] is None: return self._json(400, {"error": "missing 'file' form field (the audio)"}) audio_b64 = base64.b64encode(fields["file"][1]).decode() resp_format = (fields.get("response_format", (None, b"json"))[1] or b"json").decode().strip() prompt = ("Transcribe the speech in this audio and translate it into English. " "Output only the English text, nothing else." if translate else "Transcribe the speech in this audio verbatim. " "Output only the transcription text, nothing else.") native = {"model": AUDIO_MODEL, "stream": False, "keep_alive": "30m", "options": {"temperature": 0}, "messages": [{"role": "user", "content": prompt, "images": [audio_b64]}]} print(f"[router] stt -> {AUDIO_MODEL}", flush=True) try: req = urllib.request.Request(OLLAMA + "/api/chat", data=json.dumps(native).encode(), headers={"content-type": "application/json"}) with urllib.request.urlopen(req, timeout=600) as r: d = json.loads(r.read()) except urllib.error.HTTPError as e: return self._json(e.code, {"error": f"stt upstream {e.code}: {e.read().decode()[:300]}"}) except Exception as e: return self._json(502, {"error": f"stt upstream: {e}"}) text = ((d.get("message") or {}).get("content", "") or "").strip() if resp_format == "text": return self._text(200, text) return self._json(200, {"text": text}) def _handle_image(self, body): # OpenAI /v1/images/generations -> ComfyUI (submit workflow, poll, fetch images) if not self._authed(): return self._json(401, {"error": "unauthorized"}) prompt = (body.get("prompt") or "").strip() if not prompt: return self._json(400, {"error": "missing 'prompt'"}) n = max(1, int(body.get("n", 1) or 1)) try: w, h = (int(x) for x in str(body.get("size", "1024x1024")).lower().split("x")) except Exception: w, h = 1024, 1024 fmt = body.get("response_format", "b64_json") try: with open(COMFY_WORKFLOW) as f: wf = json.load(f) except Exception as e: return self._json(500, {"error": f"workflow '{COMFY_WORKFLOW}' unreadable: {e}"}) subs = {"{{PROMPT}}": prompt, "{{NEGATIVE}}": body.get("negative_prompt", ""), "{{WIDTH}}": w, "{{HEIGHT}}": h, "{{BATCH}}": n, "{{SEED}}": int(time.time() * 1000) % 2_000_000_000, "{{STEPS}}": int(body.get("steps", 25)), "{{CHECKPOINT}}": COMFY_CHECKPOINT or "{{CHECKPOINT}}"} wf = fill_workflow(wf, subs) print(f"[router] image -> comfyui ({w}x{h} n={n})", flush=True) try: req = urllib.request.Request(COMFYUI + "/prompt", data=json.dumps({"prompt": wf}).encode(), headers={"content-type": "application/json"}) pid = json.loads(urllib.request.urlopen(req, timeout=30).read()).get("prompt_id") if not pid: return self._json(502, {"error": "comfyui returned no prompt_id"}) images, deadline = [], time.time() + 600 while time.time() < deadline: time.sleep(1.5) hist = json.loads(urllib.request.urlopen(COMFYUI + "/history/" + pid, timeout=15).read()) entry = hist.get(pid) if entry and entry.get("outputs"): for node in entry["outputs"].values(): images.extend(node.get("images", [])) break if not images: return self._json(504, {"error": "comfyui timed out producing images"}) out = [] for im in images[:n]: q = urllib.parse.urlencode({"filename": im.get("filename", ""), "subfolder": im.get("subfolder", ""), "type": im.get("type", "output")}) data = urllib.request.urlopen(COMFYUI + "/view?" + q, timeout=60).read() b64 = base64.b64encode(data).decode() out.append({"b64_json": b64} if fmt == "b64_json" else {"url": "data:image/png;base64," + b64}) return self._json(200, {"created": int(time.time()), "data": out}, extra={"x-router-model": "comfyui"}) except urllib.error.HTTPError as e: return self._json(e.code, {"error": f"comfyui {e.code}: {e.read().decode()[:300]}"}) except Exception as e: return self._json(502, {"error": f"comfyui unreachable at {COMFYUI}: {e}"}) def _handle_audio(self, body): # OpenAI request carrying audio -> Ollama native /api/chat (images field), Gemma 4 model = AUDIO_MODEL native = {"model": model, "messages": to_native_messages(body.get("messages", [])), "stream": False, "keep_alive": "30m", "options": {}} if isinstance(body.get("options"), dict): native["options"].update(body["options"]) if body.get("max_tokens"): native["options"]["num_predict"] = body["max_tokens"] if body.get("temperature") is not None: native["options"]["temperature"] = body["temperature"] print(f"[router] audio -> {model}", flush=True) try: req = urllib.request.Request(OLLAMA + "/api/chat", data=json.dumps(native).encode(), headers={"content-type": "application/json"}) with urllib.request.urlopen(req, timeout=600) as r: d = json.loads(r.read()) except urllib.error.HTTPError as e: return self._json(e.code, {"error": f"audio upstream {e.code}: {e.read().decode()[:300]}"}) except Exception as e: return self._json(502, {"error": f"audio upstream: {e}"}) content = (d.get("message") or {}).get("content", "") or "" pc, ec = d.get("prompt_eval_count", 0) or 0, d.get("eval_count", 0) or 0 cid, created = "chatcmpl-audio", int(time.time()) hdr = {"x-router-model": model, "x-router-initial-model": model, "x-router-decided-by": "audio"} if body.get("stream"): self.send_response(200) self.send_header("content-type", "text/event-stream") self.send_header("cache-control", "no-cache") self.send_header("connection", "close") for k, v in hdr.items(): self.send_header(k, v) self.end_headers() first = {"id": cid, "object": "chat.completion.chunk", "created": created, "model": model, "choices": [{"index": 0, "delta": {"role": "assistant", "content": content}, "finish_reason": None}]} last = {"id": cid, "object": "chat.completion.chunk", "created": created, "model": model, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]} self.wfile.write(f"data: {json.dumps(first)}\n\n".encode()) self.wfile.write(f"data: {json.dumps(last)}\n\n".encode()) self.wfile.write(b"data: [DONE]\n\n") self.wfile.flush() else: self._json(200, {"id": cid, "object": "chat.completion", "created": created, "model": model, "choices": [{"index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": "stop"}], "usage": {"prompt_tokens": pc, "completion_tokens": ec, "total_tokens": pc + ec}}, extra=hdr) def do_POST(self): path = self.path.rstrip("/") if path == "/v1/audio/transcriptions": return self._handle_transcription() if path == "/v1/audio/translations": return self._handle_transcription(translate=True) if path == "/v1/images/generations": length = int(self.headers.get("content-length", 0) or 0) try: ibody = json.loads(self.rfile.read(length) or b"{}") except Exception as e: return self._json(400, {"error": f"bad request: {e}"}) return self._handle_image(ibody) if path.startswith("/v1/messages"): return self._proxy_anthropic(path) if path == "/api/show": # ollama launch model probe (read-only) length = int(self.headers.get("content-length", 0) or 0) return self._proxy_ollama("/api/show", self.rfile.read(length) or b"{}") if path != "/v1/chat/completions": return self._json(404, {"error": "not found"}) if not self._authed(): return self._json(401, {"error": "unauthorized: missing/invalid Bearer token"}) try: length = int(self.headers.get("content-length", 0)) body = json.loads(self.rfile.read(length) or b"{}") except Exception as e: return self._json(400, {"error": f"bad request: {e}"}) if has_audio(body.get("messages", [])): # OpenAI audio -> native Ollama path return self._handle_audio(body) requested = (body.get("model") or "auto").strip() auth = self.headers.get("authorization", "") token = auth[7:].strip() if auth.startswith("Bearer ") else "" force_uncensored = bool(UNCENSORED_KEY) and token == UNCENSORED_KEY if force_uncensored or requested in ("auto-uncensored", "uncensored"): model, decided_by = choose_uncensored(body), "uncensored" elif requested not in ("auto", "router", "router/auto"): model, decided_by = requested, "client" else: model, decided_by = choose_model(body), "router" initial_model = model # the routing decision, before any fallback stream = bool(body.get("stream")) tried = [] while True: body["model"] = model tried.append(model) print(f"[router] {decided_by:6s} -> {model}" f"{' (stream)' if stream else ''}", flush=True) try: up = _open_upstream(body) break except urllib.error.HTTPError as e: nxt = FALLBACK.get(model) if nxt and nxt not in tried: print(f"[router] {model} failed ({e.code}); fallback -> {nxt}", flush=True) model, decided_by = nxt, "fallback" continue detail = e.read().decode(errors="replace")[:500] return self._json(e.code, {"error": f"upstream {e.code}: {detail}", "model": model}) except Exception as e: nxt = FALLBACK.get(model) if nxt and nxt not in tried: model, decided_by = nxt, "fallback" continue return self._json(502, {"error": f"upstream unreachable: {e}", "model": model}) hdr = {"x-router-model": model, "x-router-initial-model": initial_model, "x-router-decided-by": decided_by} if stream: self.send_response(200) self.send_header("content-type", "text/event-stream") self.send_header("cache-control", "no-cache") self.send_header("connection", "close") for k, v in hdr.items(): self.send_header(k, v) self.end_headers() try: while True: chunk = up.read(2048) if not chunk: break self.wfile.write(chunk) self.wfile.flush() except Exception: pass finally: up.close() else: data = up.read() up.close() self.send_response(200) self.send_header("content-type", "application/json") self.send_header("content-length", str(len(data))) for k, v in hdr.items(): self.send_header(k, v) self.end_headers() self.wfile.write(data) if __name__ == "__main__": print(f">> llm-router on http://{HOST}:{PORT}/v1 (upstream {UPSTREAM}, " f"auth {'ON' if API_KEY else 'OFF'})", flush=True) ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()