Spaces:
Running
Running
| """Small, fail-closed provider adapter shared by REST and Gradio. | |
| Configured routes are not health claims. Explicit model choices never fall back | |
| to another provider; ``auto`` may choose from configured routes. | |
| """ | |
| import json | |
| import os | |
| import re | |
| import time | |
| import urllib.error | |
| import urllib.parse | |
| import urllib.request | |
| ROUTES = { | |
| "qwen": ("QWEN", "https://dashscope-intl.aliyuncs.com/compatible-mode/v1", "qwen-plus"), | |
| "llama": ("GROQ", "https://api.groq.com/openai/v1", "llama-3.3-70b-versatile"), | |
| "deepseek": ("DEEPSEEK", "https://api.deepseek.com", "deepseek-chat"), | |
| "gemini": ("GEMINI", "https://generativelanguage.googleapis.com/v1beta/openai", "gemini-2.5-flash"), | |
| "nemotron": ("OPENROUTER", "https://openrouter.ai/api/v1", "nvidia/nemotron-3-nano-30b-a3b:free"), | |
| "gpt-oss": ("OPENROUTER", "https://openrouter.ai/api/v1", "openai/gpt-oss-20b:free"), | |
| "lfm": ("OPENROUTER", "https://openrouter.ai/api/v1", "liquid/lfm-2.5-1.2b-instruct:free"), | |
| "openrouter-free": ("OPENROUTER", "https://openrouter.ai/api/v1", "openrouter/free"), | |
| } | |
| MODEL_ENV = { | |
| "nemotron": "OPENROUTER_NEMOTRON_MODEL_NAME", | |
| "gpt-oss": "OPENROUTER_GPT_OSS_MODEL_NAME", | |
| "lfm": "OPENROUTER_LFM_MODEL_NAME", | |
| "openrouter-free": "OPENROUTER_FREE_MODEL_NAME", | |
| } | |
| ERROR_TEXT = re.compile( | |
| r"^(?:AI[_ ]GENERATION[_ ]FAILED|AI generation failed|Transcription (?:failed|unavailable)|" | |
| r"Translation (?:failed|unavailable)|Error\s*:|Service unavailable)", re.I | |
| ) | |
| def failure(code, message): | |
| return {"ok": False, "error": {"code": code, "message": message}} | |
| def valid_content(text): | |
| return isinstance(text, str) and 0 < len(text.strip()) <= 4000 and not ERROR_TEXT.search(text.strip()) | |
| def _route(choice): | |
| prefix, default_url, default_model = ROUTES[choice] | |
| key = os.getenv(prefix + "_API_KEY", "").strip() | |
| if prefix == "GEMINI": | |
| key = key or os.getenv("GOOGLE_API_KEY", "").strip() | |
| if not key or key.lower() in {"your-api-key-here", "none", "null"}: | |
| return None | |
| model = os.getenv(MODEL_ENV.get(choice, prefix + "_MODEL_NAME"), default_model).strip() | |
| base_url = os.getenv(prefix + "_BASE_URL", default_url).strip().rstrip("/") | |
| parsed = urllib.parse.urlsplit(base_url) | |
| # Only operator-configured endpoints are used. Never put keys in URLs. | |
| if parsed.scheme not in {"https", "http"} or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment: | |
| return None | |
| if parsed.scheme == "http" and parsed.hostname not in {"localhost", "127.0.0.1", "::1"}: | |
| return None | |
| if not model: | |
| return None | |
| return {"key": key, "url": base_url + "/chat/completions", "model": model, "route": choice} | |
| def available_routes(): | |
| """Return non-secret configuration metadata, without claiming availability.""" | |
| return [ | |
| {"route": name, "model": route["model"], "configured": True, "verified": False} | |
| for name in ROUTES if (route := _route(name)) | |
| ] | |
| class _NoRedirect(urllib.request.HTTPRedirectHandler): | |
| def redirect_request(self, req, fp, code, msg, headers, newurl): | |
| # Do not forward Authorization to a redirected host. | |
| return None | |
| class _OutputFailure(Exception): | |
| def __init__(self, code, message): | |
| super().__init__(message) | |
| self.code = code | |
| self.message = message | |
| def _post(route, prompt, timeout=10): | |
| body = json.dumps({ | |
| "model": route["model"], | |
| "messages": [{"role": "user", "content": prompt}], | |
| "temperature": 0.3, | |
| "max_tokens": 768, | |
| }).encode("utf-8") | |
| request = urllib.request.Request(route["url"], data=body, headers={ | |
| "Authorization": "Bearer " + route["key"], | |
| "Content-Type": "application/json", | |
| }, method="POST") | |
| with urllib.request.build_opener(_NoRedirect).open(request, timeout=timeout) as response: | |
| raw = response.read(1024 * 1024 + 1) | |
| if len(raw) > 1024 * 1024: | |
| raise ValueError("Response exceeded limit") | |
| payload = json.loads(raw) | |
| choice = payload["choices"][0] | |
| finish_reason = choice.get("finish_reason") | |
| if finish_reason in {"length", "max_tokens"}: | |
| raise _OutputFailure("OUTPUT_TRUNCATED", "The model response was cut short. Shorten the text or retry.") | |
| if finish_reason != "stop" or choice.get("message", {}).get("tool_calls"): | |
| raise _OutputFailure("INVALID_OUTPUT", "The model did not return a complete usable response.") | |
| return choice["message"]["content"] | |
| def generate_text(prompt, model_choice="auto", *, deadline=None): | |
| if not isinstance(prompt, str) or not prompt.strip() or len(prompt) > 24000: | |
| return failure("INVALID_REQUEST", "A non-empty, bounded inference prompt is required.") | |
| choice = str(model_choice or "auto").strip().lower() | |
| if choice != "auto" and choice not in ROUTES: | |
| return failure("MODEL_UNAVAILABLE", "The selected model route is not supported.") | |
| names = list(ROUTES) if choice == "auto" else [choice] | |
| configured = [route for name in names if (route := _route(name))] | |
| if not configured: | |
| return failure("MODEL_UNAVAILABLE", "The selected model route is not configured.") | |
| deadline = min(deadline or (time.monotonic() + 30), time.monotonic() + 30) | |
| attempts = [] | |
| result = failure("PROVIDER_UNAVAILABLE", "The selected provider is unavailable. Please retry.") | |
| for route in configured[:3]: | |
| remaining = deadline - time.monotonic() | |
| if remaining <= 0: | |
| break | |
| try: | |
| content = _post(route, prompt, timeout=min(10, remaining)) | |
| if valid_content(content): | |
| return {"ok": True, "text": content.strip(), "model": route["model"], "route": route["route"], "attempts": attempts} | |
| result = failure("INVALID_OUTPUT", "The model did not return usable content.") | |
| except _OutputFailure as exc: | |
| result = failure(exc.code, exc.message) | |
| except urllib.error.HTTPError as exc: | |
| if exc.code in {401, 403}: | |
| result = failure("PROVIDER_AUTH_FAILED", "The selected provider rejected its credentials.") | |
| elif exc.code == 429: | |
| result = failure("PROVIDER_RATE_LIMITED", "The selected provider is rate limited or has no remaining quota.") | |
| elif exc.code in {400, 404, 422}: | |
| result = failure("PROVIDER_CONFIGURATION_ERROR", "The selected provider rejected the configured model or request.") | |
| else: | |
| result = failure("PROVIDER_UNAVAILABLE", "The selected provider is unavailable. Please retry.") | |
| except (TimeoutError, urllib.error.URLError): | |
| result = failure("PROVIDER_UNAVAILABLE", "The selected provider could not be reached. Please retry.") | |
| except Exception: | |
| result = failure("INVALID_OUTPUT", "The model did not return usable content.") | |
| attempts.append({"route": route["route"], "code": result["error"]["code"]}) | |
| return {**result, "attempts": attempts} | |