""" providers.py — Unified LLM provider configuration and client management. Single source of truth for all model definitions, provider configs, client creation, and API call logic. Used by benchmark.py, test_models.py, and web.py. """ import json import os import re import time import typing import json_storage def _configured_secrets() -> list[str]: """Every API key this process holds. The prefix patterns below only match the providers whose keys have a distinctive shape; Mistral, Cohere, Cloudflare and Deepgram keys are plain alphanumeric. Matching on the actual configured values catches those too, whatever format a provider picks next. """ names = [cfg["env_key"] for cfg in PROVIDER_CONFIG.values() if cfg.get("env_key")] names += ["DEEPGRAM_API_KEY", "CLOUDFLARE_ACCOUNT_ID", "HF_TOKEN"] return [value for value in (os.getenv(name, "").strip() for name in names) if len(value) >= 8] def _redact_secrets(msg: str) -> str: """Strip API keys and tokens from error messages.""" for secret in _configured_secrets(): msg = msg.replace(secret, "***") # Common API key patterns msg = re.sub(r'(sk-[a-zA-Z0-9_-]{10,})', 'sk-***', msg) msg = re.sub(r'(nvapi-[a-zA-Z0-9_-]{10,})', 'nvapi-***', msg) msg = re.sub(r'(gsk_[a-zA-Z0-9_-]{10,})', 'gsk_***', msg) msg = re.sub(r'(Bearer\s+)[a-zA-Z0-9_.-]{10,}', r'\1***', msg) msg = re.sub(r'(token[=:]\s*)[a-zA-Z0-9_.-]{10,}', r'\1***', msg, flags=re.IGNORECASE) msg = re.sub(r'(key[=:]\s*)[a-zA-Z0-9_.-]{10,}', r'\1***', msg, flags=re.IGNORECASE) msg = re.sub(r'(AIzaSy[a-zA-Z0-9_-]{10,})', 'AIza***', msg) return msg # ── Provider configuration ────────────────────────────────────────────── PROVIDER_CONFIG = { # "openai": {"env_key": "OPENAI_API_KEY", "base_url": None}, # never used — no credits "deepseek": {"env_key": "DEEPSEEK_API_KEY", "base_url": "https://api.deepseek.com"}, "groq": {"env_key": "GROQ_API_KEY", "base_url": "https://api.groq.com/openai/v1"}, "mistral": {"env_key": "MISTRAL_API_KEY", "base_url": "https://api.mistral.ai/v1"}, "gemini": {"env_key": "GEMINI_API_KEY", "base_url": None}, "anthropic": {"env_key": "ANTHROPIC_API_KEY", "base_url": None}, "proton": {"env_key": "LUMO_API_KEY", "base_url": "https://lumo-api.proton.me/ai/v1", "guest": True}, "cohere": {"env_key": "COHERE_API_KEY", "base_url": "https://api.cohere.com/v2"}, "nvidia": {"env_key": "NVIDIA_API_KEY", "base_url": "https://integrate.api.nvidia.com/v1"}, "cloudflare": {"env_key": "CLOUDFLARE_API_KEY", "base_url": None, "account_id_key": "CLOUDFLARE_ACCOUNT_ID"}, "openrouter": {"env_key": "OPENROUTER_API_KEY", "base_url": "https://openrouter.ai/api/v1"}, "ovhcloud": {"env_key": "OVHCLOUD_API_KEY", "base_url": "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1"}, "huggingface": {"env_key": "HF_API_KEY", "base_url": "https://router.huggingface.co/v1"}, } # ── Model registry ────────────────────────────────────────────────────── # provider: which PROVIDER_CONFIG entry to use # input/output: cost per million tokens (0.0 = free) # score: benchmark test pass rate (e.g. "24/31") # free: whether the model is free to use MODELS = { # OpenAI — paid (in chat UI) # "gpt-4.1-nano": {"provider": "openai", "input": 0.10, "output": 0.40, "score": "", "free": False}, # "gpt-4o-mini": {"provider": "openai", "input": 0.30, "output": 1.20, "score": "25/33", "free": False}, # "gpt-4.1-mini": {"provider": "openai", "input": 0.80, "output": 3.20, "score": "24/33", "free": False}, # OpenAI — paid (test-only, not in fallback chain) # "gpt-4o": {"provider": "openai", "input": 3.75, "output": 15.0, "score": "", "free": False}, # "gpt-4.1": {"provider": "openai", "input": 3.00, "output": 12.0, "score": "", "free": False}, # "gpt-5.4-nano": {"provider": "openai", "input": 0.10, "output": 0.40, "score": "", "free": False}, # "gpt-5.4-mini": {"provider": "openai", "input": 0.80, "output": 3.20, "score": "", "free": False}, # "gpt-5.4": {"provider": "openai", "input": 3.00, "output": 12.0, "score": "", "free": False}, # Groq — all free. Llama/Qwen3-32B/Kimi-K2 left the catalog (404, 2026-10-10). # Groq rejects chat_template_kwargs, so no disable_thinking here. "openai/gpt-oss-120b": {"provider": "groq", "input": 0, "output": 0, "score": "19/31", "free": True, "benchmark_tokens": 512}, "openai/gpt-oss-20b": {"provider": "groq", "input": 0, "output": 0, "score": "", "free": True}, # qwen3.8 answered "Co je Brno?" in Slovak (2026-10-10 probe) — registered, not a candidate "qwen/qwen3.8-27b": {"provider": "groq", "input": 0, "output": 0, "score": "", "free": True}, # Google Gemini — free tier, 15 rpm per model; 429s often (reliability.json). # All answer with thinking_level="minimal" (2026-10-24 probe); without it 3.x flash # spends the whole token budget thinking. "gemini-3.5-flash-lite": {"provider": "gemini", "input": 0, "output": 0, "score": "", "free": True}, "gemini-3.5-flash": {"provider": "gemini", "input": 0.30, "output": 2.50, "score": "10/18", "free": True}, "gemini-3.1-flash-lite": {"provider": "gemini", "input": 0, "output": 0, "score": "12/18", "free": True}, "gemini-3-flash-preview": {"provider": "gemini", "input": 0, "output": 0, "score": "10/18", "free": True}, # "gemini-3.1-flash-lite-preview": {"provider": "gemini", "input": 0, "output": 0, "score": "13/18", "free": True}, # Proton Lumo — paid, needs LUMO_API_KEY (per-token price not published here) "lumo": {"provider": "proton", "input": 0.25, "output": 1.5, "score": "24/31", "free": False}, # NVIDIA NIM — free tier (40 rpm) "meta/muse-glimmer-30b": {"provider": "nvidia", "input": 0, "output": 0, "score": "25/33", "free": True}, "nvidia/nemotron-3.5-lightning-30b-a3b": {"provider": "nvidia", "input": 0, "output": 0, "score": "23/33", "free": True, "disable_thinking": True}, # nemotron-3-super-120b: 404 "not found for account"; glm-5.3-flash: 3-min timeouts (2026-10-24) "nvidia/nemotron-3.5-super-vl-120b-a12b": {"provider": "nvidia", "input": 0, "output": 0, "score": "", "free": True, "disable_thinking": True}, "nvidia/nemotron-3-ultra-550b-a55b": {"provider": "nvidia", "input": 0, "output": 0, "score": "", "free": True, "disable_thinking": True}, "minimaxai/minimax-m3.1-flash": {"provider": "nvidia", "input": 0, "output": 0, "score": "", "free": True, "low_reasoning": True}, # Cloudflare Workers AI — free tier "@cf/qwen/qwen2.5-coder-32b-instruct": {"provider": "cloudflare", "input": 0, "output": 0, "score": "22/33", "free": True}, "@cf/deepseek-ai/deepseek-r1-distill-qwen-32b": {"provider": "cloudflare", "input": 0, "output": 0, "score": "21/33", "free": True}, "@cf/meta/llama-3.1-8b-instruct-fp8": {"provider": "cloudflare", "input": 0, "output": 0, "score": "20/33", "free": True}, "@cf/meta/llama-3.3-70b-instruct-fp8-fast": {"provider": "cloudflare", "input": 0, "output": 0, "score": "", "free": True}, "@cf/mistralai/mistral-small-3.1-24b-instruct": {"provider": "cloudflare", "input": 0, "output": 0, "score": "", "free": True}, "@cf/meta/llama-4-scout-17b-16e-instruct": {"provider": "cloudflare", "input": 0, "output": 0, "score": "", "free": True}, "@cf/aisingapore/gemma-sea-lion-v4-27b-it": {"provider": "cloudflare", "input": 0, "output": 0, "score": "", "free": True}, # qwen3 thinks first; CF ignores chat_template_kwargs, so give the benchmark room "@cf/qwen/qwen3-30b-a3b-fp8": {"provider": "cloudflare", "input": 0, "output": 0, "score": "", "free": True, "benchmark_tokens": 512}, "@cf/openai/gpt-oss-120b": {"provider": "cloudflare", "input": 0, "output": 0, "score": "", "free": True}, # reasoning — not a candidate "@cf/nvidia/nemotron-3-120b-a12b": {"provider": "cloudflare", "input": 0, "output": 0, "score": "", "free": True}, # reasoning, 2-3 tok/s effective — not a candidate # Mistral — free experiment tier # Small/Medium: 429 code 1300 on every call since 2026-09 (re-checked 2026-10-10) # "mistral-small-latest": {"provider": "mistral", "input": 0.20, "output": 0.60, "score": "23/31", "free": True}, # "mistral-medium-latest": {"provider": "mistral", "input": 0.40, "output": 2.00, "score": "19/31", "free": True}, # "mistral-large-latest": {"provider": "mistral", "input": 2.00, "output": 6.00, "score": "24/31", "free": True}, # 403 tier_not_allowed — not available in our subscription tier "open-mistral-nemo": {"provider": "mistral", "input": 0.02, "output": 0.04, "score": "21/31", "free": True}, "ministral-8b-latest": {"provider": "mistral", "input": 0.15, "output": 0.15, "score": "", "free": True}, "ministral-14b-latest": {"provider": "mistral", "input": 0.20, "output": 0.20, "score": "", "free": True}, "ministral-3b-latest": {"provider": "mistral", "input": 0, "output": 0, "score": "", "free": True}, # price not looked up # Hugging Face: free accounts get no inference credits (402, 2026-10-10) — not listed. } # ── Candidate models (15 curated, benchmarked at startup → top 5 active) ── # Balanced across providers, no OpenAI. Lumo is paid, so it sits last. CANDIDATE_MODELS = [ "ministral-14b-latest", # verified alternative to rate-limited Small # "mistral-large-latest", # mistral — 403 tier_not_allowed "ministral-8b-latest", # explicit successor to the old Nemo alias "nvidia/nemotron-3.5-lightning-30b-a3b", # nvidia — proven reliable "nvidia/nemotron-3.5-super-vl-120b-a12b", "minimaxai/minimax-m3.1-flash", "ministral-3b-latest", "openai/gpt-oss-120b", # groq # Muse is served but its reasoning exceeds the startup benchmark budget. "@cf/deepseek-ai/deepseek-r1-distill-qwen-32b", # cloudflare "@cf/meta/llama-3.1-8b-instruct-fp8", # cloudflare "@cf/qwen/qwen2.5-coder-32b-instruct", # cloudflare "@cf/meta/llama-3.3-70b-instruct-fp8-fast", "@cf/mistralai/mistral-small-3.1-24b-instruct", "@cf/qwen/qwen3-30b-a3b-fp8", "@cf/meta/llama-4-scout-17b-16e-instruct", "@cf/aisingapore/gemma-sea-lion-v4-27b-it", "gemini-3.5-flash-lite", # gemini — 15 rpm free tier per model "gemini-3.5-flash", "gemini-3.1-flash-lite", "lumo", # proton — paid, tried last ] # ── HF Spaces candidates (subset) ──────────────────────────────────── HF_CANDIDATES = [ "ministral-14b-latest", # "mistral-large-latest", # 403 tier_not_allowed "ministral-8b-latest", "nvidia/nemotron-3.5-lightning-30b-a3b", "nvidia/nemotron-3.5-super-vl-120b-a12b", "minimaxai/minimax-m3.1-flash", "ministral-3b-latest", "openai/gpt-oss-120b", "@cf/meta/llama-3.1-8b-instruct-fp8", "@cf/qwen/qwen2.5-coder-32b-instruct", "@cf/deepseek-ai/deepseek-r1-distill-qwen-32b", "@cf/meta/llama-3.3-70b-instruct-fp8-fast", "@cf/mistralai/mistral-small-3.1-24b-instruct", "@cf/qwen/qwen3-30b-a3b-fp8", "@cf/meta/llama-4-scout-17b-16e-instruct", "@cf/aisingapore/gemma-sea-lion-v4-27b-it", "gemini-3.5-flash-lite", "gemini-3.5-flash", "gemini-3.1-flash-lite", "lumo", # proton — paid, tried last ] # Active models (set at startup after benchmark — top 5 of candidates) ACTIVE_MODELS: list[str] = [] # Fallback chain (rebuilt at startup, ordered by speed) FALLBACK_CHAIN: list[str] = list(CANDIDATE_MODELS) # ── Reliability tracking ───────────────────────────────────────────────── _RELIABILITY_FILE = os.path.join(os.path.dirname(__file__), "data", "reliability.json") def _load_reliability(): try: with open(_RELIABILITY_FILE) as f: return json.load(f) except (OSError, ValueError): # missing/unreadable file or corrupt JSON return {} # The status, not any "400" in the text: "Error code: 400" (OpenAI SDK), # "HTTP Error 400" (urllib), "400 INVALID_ARGUMENT" (genai), "400 Client Error". _HTTP_400 = re.compile(r"^400\b|(?:Error(?: code:)?|HTTP) 400\b") def log_reliability(model, success, error_msg=None, ttft=None, tok_sec=None): """Record a success or failure for a model, with optional speed stats. Thread-safe. Skips logging for HTTP 400 errors (config bugs, not provider reliability issues).""" if not success and error_msg and (_HTTP_400.search(str(error_msg)) or "Unsupported parameter" in str(error_msg)): return # don't penalize provider for our config mistakes def record(data): r = data.setdefault(model, {"attempts": 0, "successes": 0, "errors": 0, "last_success": None, "last_error": None, "last_error_msg": None}) r["attempts"] += 1 ts = int(time.time()) if success: r["successes"] += 1 r["last_success"] = ts # Track real-world speed (rolling average) if ttft is not None: prev_ttft = r.get("avg_ttft") r["avg_ttft"] = round((prev_ttft * 0.7 + ttft * 0.3), 3) if prev_ttft else round(ttft, 3) if tok_sec is not None: prev_tps = r.get("avg_tok_sec") r["avg_tok_sec"] = round((prev_tps * 0.7 + tok_sec * 0.3), 1) if prev_tps else round(tok_sec, 1) else: r["errors"] += 1 r["last_error"] = ts if error_msg: r["last_error_msg"] = _redact_secrets(str(error_msg))[:500] json_storage.update_json(_RELIABILITY_FILE, dict, record) def get_reliability(model): """Return reliability % (0-100) for a model, or None if no data.""" data = _load_reliability() r = data.get(model) if not r or r["attempts"] == 0: return None return round(r["successes"] / r["attempts"] * 100) def get_all_reliability(): """Return dict of model -> reliability stats.""" return _load_reliability() # ── Client creation ───────────────────────────────────────────────────── def get_client(provider: str) -> tuple[typing.Any, str | None]: """Get API client for a provider. Returns (client, error_msg). The client is Any on purpose: it is an OpenAI, genai or Anthropic SDK client, or a plain dict for cohere/cloudflare. Callers pick the API by provider name, which a type checker cannot tie to the client's type. """ cfg = PROVIDER_CONFIG[provider] env_key = cfg["env_key"] api_key = os.getenv(env_key or "", "").strip() if cfg.get("guest") and not api_key: from openai import OpenAI, Timeout, omit class _GuestOpenAI(OpenAI): """Lumo serves guests unauthenticated; SDK 3.x insists on an auth method.""" def _validate_headers(self, headers, custom_headers) -> None: return # api_key is never sent — default_headers drops Authorization entirely. return _GuestOpenAI( api_key="guest", base_url=cfg["base_url"], # The SDK drops Omit-valued headers; its hint only allows str. default_headers={"Authorization": omit}, # pyright: ignore[reportArgumentType] timeout=Timeout(60, connect=5), ), None if not api_key: return None, f"Chybí {env_key} v .env" try: if provider == "cohere": return {"type": "cohere", "api_key": api_key}, None if provider == "cloudflare": account_id = os.getenv(cfg.get("account_id_key", ""), "").strip() if not account_id: return None, "Chybí CLOUDFLARE_ACCOUNT_ID v .env" return {"type": "cloudflare", "api_key": api_key, "account_id": account_id}, None if provider == "nvidia": from openai import OpenAI, Timeout return OpenAI(api_key=api_key, base_url="https://integrate.api.nvidia.com/v1", timeout=Timeout(60, connect=5)), None if provider == "gemini": from google import genai return genai.Client(api_key=api_key), None elif provider == "anthropic": import anthropic return anthropic.Anthropic(api_key=api_key), None else: from openai import OpenAI, Timeout kwargs = {"api_key": api_key, "timeout": Timeout(60, connect=5)} if cfg["base_url"]: kwargs["base_url"] = cfg["base_url"] return OpenAI(**kwargs), None except Exception as e: # noqa: BLE001 - any SDK import/setup failure is returned as the error return None, _redact_secrets(str(e))[:80] # ── Sync call (test_models.py) ────────────────────────────────────────── class _ChatRequestOptions(typing.TypedDict, total=False): extra_body: dict[str, dict[str, bool]] reasoning_effort: typing.Literal["low"] def _chat_request_options(model_name: str) -> _ChatRequestOptions: """Keep supported reasoning models within the visible-answer token budget.""" if MODELS[model_name].get("disable_thinking"): return {"extra_body": {"chat_template_kwargs": {"enable_thinking": False}}} if MODELS[model_name].get("low_reasoning"): return {"reasoning_effort": "low"} return {} # Gemini 3.x flash otherwise thinks through the whole max_output_tokens budget. _GEMINI_THINKING = {"thinking_level": "minimal"} def _cf_text(data: dict) -> str: """Workers AI answers either {"response": ...} or OpenAI-style choices (gpt-oss, nemotron, gemma-4).""" result = data.get("result") or {} if result.get("response"): return result["response"] choices = result.get("choices") or [] return (choices[0].get("message") or {}).get("content") or "" if choices else "" def call_model(model_name, messages, max_tokens=300, temperature=0.7): """Call a model synchronously. Returns (text, tokens_in, tokens_out).""" info = MODELS[model_name] provider = info["provider"] client, err = get_client(provider) if err: raise RuntimeError(err) if provider == "cloudflare": import urllib.request account_id = client["account_id"] payload = json.dumps({ "messages": messages, "max_tokens": max_tokens, }).encode() req = urllib.request.Request( f"https://api.cloudflare.com/client/v4/accounts/{account_id}/ai/run/{model_name}", data=payload, headers={"Authorization": f"Bearer {client['api_key']}", "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=60) as resp: data = json.loads(resp.read().decode()) text = _cf_text(data) tin = sum(len(m["content"]) // 4 for m in messages) tout = len(text) // 4 return text.strip(), tin, tout if provider == "cohere": import urllib.request # v2 API: system message goes in messages array, NOT as preamble cohere_msgs = [{"role": m["role"], "content": m["content"]} for m in messages] payload = json.dumps({ "model": model_name, "messages": cohere_msgs, "temperature": temperature, "max_tokens": max_tokens, }).encode() req = urllib.request.Request( "https://api.cohere.com/v2/chat", data=payload, headers={"Authorization": f"Bearer {client['api_key']}", "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode()) text = data.get("message", {}).get("content", [{}])[0].get("text", "") tin = data.get("usage", {}).get("tokens", {}).get("input_tokens", 0) tout = data.get("usage", {}).get("tokens", {}).get("output_tokens", 0) return text.strip(), tin, tout if provider == "gemini": from google.genai import types system_text, user_text = "", "" for m in messages: if m["role"] == "system": system_text = m["content"] elif m["role"] == "user": user_text = m["content"] response = client.models.generate_content( model=model_name, contents=user_text, config=types.GenerateContentConfig( system_instruction=system_text, temperature=temperature, max_output_tokens=max_tokens, thinking_config=types.ThinkingConfig.model_validate(_GEMINI_THINKING))) text = response.text or "" tin = getattr(response.usage_metadata, "prompt_token_count", 0) or 0 tout = getattr(response.usage_metadata, "candidates_token_count", 0) or 0 return text, tin, tout if provider == "anthropic": system_text = "" user_messages = [] for m in messages: if m["role"] == "system": system_text = m["content"] else: user_messages.append(m) r = client.messages.create(model=model_name, system=system_text, messages=user_messages, temperature=temperature, max_tokens=max_tokens) return r.content[0].text or "", r.usage.input_tokens, r.usage.output_tokens # OpenAI-compatible (openai, deepseek, groq, mistral) # GPT-5+ uses max_completion_tokens instead of max_tokens tok_kwarg = "max_completion_tokens" if model_name.startswith("gpt-5") else "max_tokens" r = client.chat.completions.create(model=model_name, messages=messages, temperature=temperature, **{tok_kwarg: max_tokens}, **_chat_request_options(model_name)) return r.choices[0].message.content or "", r.usage.prompt_tokens, r.usage.completion_tokens def _call_model_tracked(model_name, messages, **kwargs): """call_model with reliability logging.""" try: result = call_model(model_name, messages, **kwargs) log_reliability(model_name, success=True) return result except Exception as e: log_reliability(model_name, success=False, error_msg=str(e)) raise # ── Streaming call (benchmark.py speed test) ──────────────────────────── def stream_benchmark(model_name, system_prompt, user_prompt, max_tokens=150): """Stream a model for speed benchmarking. Returns (latency, ttft, tok_per_sec, error).""" latency, ttft, tok_sec, error = _stream_benchmark_inner( model_name, system_prompt, user_prompt, max_tokens) # Log reliability for benchmark runs too if error: log_reliability(model_name, success=False, error_msg=error) else: log_reliability(model_name, success=True) return latency, ttft, tok_sec, error def _stream_benchmark_inner(model_name, system_prompt, user_prompt, max_tokens=150): info = MODELS.get(model_name) if not info: return None, None, None, f"Unknown model: {model_name}" provider = info["provider"] messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}] max_tokens = max(max_tokens, info.get("benchmark_tokens", max_tokens)) _BENCH_TIMEOUT = 7 # aggressive timeout for benchmarks try: if provider in ("cohere", "cloudflare"): t0 = time.time() text, _, tout = call_model(model_name, messages, max_tokens=max_tokens) total = time.time() - t0 if not text or not text.strip(): return None, None, None, "Empty response (no visible output)" tok_sec = tout / total if total > 0 else 0 return total, total * 0.3, tok_sec, None if provider == "gemini": from google.genai import types client, err = get_client(provider) if err: return None, None, None, err t0 = time.time() ttft = None token_count = 0 stream = client.models.generate_content_stream( model=model_name, contents=user_prompt, config=types.GenerateContentConfig( system_instruction=system_prompt, max_output_tokens=max_tokens, thinking_config=types.ThinkingConfig.model_validate(_GEMINI_THINKING))) for chunk in stream: if chunk.text and chunk.text.strip(): if ttft is None: ttft = time.time() - t0 token_count += len(chunk.text.split()) total = time.time() - t0 if ttft is None: return None, None, None, "Empty response (no visible output)" return total, ttft, token_count / total if total > 0 else 0, None # OpenAI-compatible (openai, deepseek, groq, mistral) client, err = get_client(provider) if err: return None, None, None, err t0 = time.time() ttft = None token_count = 0 stream = client.chat.completions.create( model=model_name, messages=messages, max_tokens=max_tokens, stream=True, timeout=_BENCH_TIMEOUT, **_chat_request_options(model_name)) for chunk in stream: if time.time() - t0 > _BENCH_TIMEOUT: break if (chunk.choices and chunk.choices[0].delta.content and chunk.choices[0].delta.content.strip()): if ttft is None: ttft = time.time() - t0 token_count += 1 total = time.time() - t0 if ttft is None: return None, None, None, "Empty response (no visible output)" return total, ttft, token_count / total if total > 0 else 0, None except Exception as e: # noqa: BLE001 - any provider failure is a failed benchmark return None, None, None, _redact_secrets(str(e)) # ── Streaming generator (web.py SSE) ──────────────────────────────────── def stream_chat(model_name, messages, max_tokens=1024, temperature=0.7): """Stream a model response as (text_chunk) generator. For web SSE.""" info = MODELS.get(model_name) if not info: raise RuntimeError(f"Unknown model: {model_name}") provider = info["provider"] client, err = get_client(provider) if err: raise RuntimeError(err) if not client: raise RuntimeError(f"No client for {provider}") if provider == "cloudflare": # Cloudflare Workers AI doesn't stream; yield full response text, _, _ = call_model(model_name, messages, max_tokens=max_tokens, temperature=temperature) yield text return if provider == "cohere": import urllib.request # v2 API: system message goes in messages array cohere_msgs = [{"role": m["role"], "content": m["content"]} for m in messages] payload = json.dumps({ "model": model_name, "messages": cohere_msgs, "temperature": temperature, "max_tokens": max_tokens, "stream": True, }).encode() req = urllib.request.Request( "https://api.cohere.com/v2/chat", data=payload, headers={"Authorization": f"Bearer {client['api_key']}", "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=60) as resp: for raw_line in resp: line = raw_line.decode("utf-8").strip() if not line or not line.startswith("data:"): continue try: data = json.loads(line[5:].strip()) except json.JSONDecodeError: continue if data.get("type") == "content-delta": delta = data.get("delta", {}).get("message", {}).get("content", {}).get("text", "") if delta: yield delta return if provider == "gemini": from google.genai import types sys_text = "" contents = [] for m in messages: if m["role"] == "system": sys_text = m["content"] else: contents.append(types.Content( role="user" if m["role"] == "user" else "model", parts=[types.Part(text=m["content"])])) stream = client.models.generate_content_stream( model=model_name, contents=contents, config=types.GenerateContentConfig( system_instruction=sys_text, temperature=temperature, max_output_tokens=max_tokens, thinking_config=types.ThinkingConfig.model_validate(_GEMINI_THINKING))) for chunk in stream: if chunk.text: yield chunk.text return # OpenAI-compatible (openai, deepseek, groq, mistral) tok_kwarg = {"max_completion_tokens": max_tokens} if model_name.startswith("gpt-5") else {"max_tokens": max_tokens} stream = client.chat.completions.create( model=model_name, messages=messages, temperature=temperature, **tok_kwarg, stream=True, **_chat_request_options(model_name)) for chunk in stream: if chunk.choices and chunk.choices[0].delta.content: yield chunk.choices[0].delta.content # ── Error formatting ──────────────────────────────────────────────────── def friendly_error(emsg): """Convert raw API errors to human-readable Czech messages.""" el = emsg.lower() if "401" in emsg or "invalid_api_key" in el or "unauthorized" in el or "authentication" in el: return "Neplatný API klíč" if "402" in emsg or "insufficient" in el or "balance" in el: return "Nedostatečný kredit" if "429" in emsg or "quota" in el or "rate" in el or "resource_exhausted" in el: return "Rate limit / vyčerpaná kvóta" if "404" in emsg or "not found" in el or "does not exist" in el: return "Model nenalezen" if "403" in emsg or "forbidden" in el or "permission" in el: return "Přístup zamítnut" import re if "'message':" in emsg: m = re.search(r"'message':\s*'([^']+)'", emsg) if m: return m.group(1)[:80] return emsg[:80] # ── Dynamic free-model discovery ──────────────────────────────────────── # OVHcloud rotates its free catalog; OpenRouter discovery is limited to the # models verified to return visible chat output. AUTO_DISCOVER_PROVIDERS = ["openrouter", "ovhcloud"] OPENROUTER_MODELS = { name for name, config in MODELS.items() if config["provider"] == "openrouter" } def discover_free_models(provider: str, limit: int = 20) -> list[str]: """Query a provider's /models endpoint, return up to `limit` free-tier model IDs.""" client, err = get_client(provider) if err: return [] try: ids = [m.id for m in client.models.list().data] except Exception: # noqa: BLE001 - an unreachable catalog just yields no models return [] if provider == "openrouter": ids = [model_id for model_id in ids if model_id in OPENROUTER_MODELS] return ids[:limit] def register_discovered_models(limit_per_provider: int = 20) -> list[str]: """Add auto-discovered free models to MODELS + CANDIDATE_MODELS/HF_CANDIDATES. Returns the new IDs.""" added = [] for provider in AUTO_DISCOVER_PROVIDERS: for model_id in discover_free_models(provider, limit_per_provider): if model_id in MODELS: continue MODELS[model_id] = {"provider": provider, "input": 0, "output": 0, "score": "", "free": True} CANDIDATE_MODELS.append(model_id) HF_CANDIDATES.append(model_id) added.append(model_id) return added # ── Provider health check ─────────────────────────────────────────────── # Test model to use when validating each provider's API key _CHECK_MODELS = { "openai": "gpt-4.1-nano", "deepseek": "deepseek-chat", "groq": "llama-3.1-8b-instant", "mistral": "mistral-small-latest", "gemini": "gemini-3.5-flash", "anthropic": "claude-haiku-4-5-20251001", "nvidia": "nvidia/nemotron-3.5-lightning-30b-a3b", "proton": "lumo", "huggingface": "zai-org/GLM-5.3", } def check_provider(provider): """Validate API key with a minimal test call. Returns (ok, message).""" client, err = get_client(provider) if err: return False, err try: if provider in AUTO_DISCOVER_PROVIDERS: models = discover_free_models(provider) if models: return True, f"OK — {len(models)} free models: {', '.join(models[:3])}" return False, "Žádné free modely nenalezeny" if provider == "cloudflare": import urllib.request account_id = client["account_id"] payload = json.dumps({ "messages": [{"role": "user", "content": "Say ok."}], "max_tokens": 5, }).encode() req = urllib.request.Request( f"https://api.cloudflare.com/client/v4/accounts/{account_id}/ai/run/@cf/meta/llama-3.2-3b-instruct", data=payload, headers={"Authorization": f"Bearer {client['api_key']}", "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode()) text = _cf_text(data) return True, f"OK — {text.strip()[:20]}" if provider == "cohere": import urllib.request payload = json.dumps({ "model": "command-r7b-12-2024", "messages": [{"role": "user", "content": "Řekni 'ok'."}], "max_tokens": 5, }).encode() req = urllib.request.Request( "https://api.cohere.com/v2/chat", data=payload, headers={"Authorization": f"Bearer {client['api_key']}", "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=15) as resp: data = json.loads(resp.read().decode()) text = data.get("message", {}).get("content", [{}])[0].get("text", "") return True, f"OK — {text.strip()[:20]}" if provider == "gemini": response = client.models.generate_content( model="gemini-3.5-flash", contents="Řekni 'ok'.", config={"max_output_tokens": 5}) return True, f"OK — {response.text.strip()[:20]}" if provider == "anthropic": r = client.messages.create( model="claude-haiku-4-5-20251001", max_tokens=10, messages=[{"role": "user", "content": "Řekni 'ok'."}]) return True, f"OK — {r.content[0].text.strip()[:20]}" # OpenAI-compatible test_model = _CHECK_MODELS.get(provider, "gpt-4.1-nano") r = client.chat.completions.create( model=test_model, messages=[{"role": "user", "content": "Řekni 'ok'."}], max_tokens=5) return True, f"OK — {r.choices[0].message.content.strip()[:20]}" except Exception as e: # noqa: BLE001 - any provider failure is reported as not OK return False, friendly_error(str(e))