1"""LLMClient — single entry point for every model call (OpenRouter, OpenAI-compatible).23Handles: streaming SSE parsing (keep-alive comments, fragmented tool_calls, final usage4chunk), native model fallback via `models`, retries on 429/5xx, attribution headers.5"""67from __future__ import annotations89import asyncio10import json11from collections.abc import AsyncIterator12from typing import Any1314import httpx1516from app.core.config import Settings, get_settings17from app.core.logging import get_logger18from app.llm.schemas import StreamEvent1920log = get_logger("llm.openrouter")2122RETRY_STATUS = {429, 500, 502, 503, 504}232425class LLMError(Exception):26 def __init__(self, message: str, status: int | None = None) -> None:27 super().__init__(message)28 self.status = status293031class LLMClient:32 def __init__(self, settings: Settings | None = None) -> None:33 self.settings = settings or get_settings()34 self._client: httpx.AsyncClient | None = None3536 def _headers(self) -> dict[str, str]:37 return {38 "Authorization": f"Bearer {self.settings.OPENROUTER_API_KEY.get_secret_value()}",39 "HTTP-Referer": self.settings.OPENROUTER_APP_URL,40 "X-Title": self.settings.OPENROUTER_APP_NAME,41 "Content-Type": "application/json",42 }4344 def client(self) -> httpx.AsyncClient:45 if self._client is None or self._client.is_closed:46 self._client = httpx.AsyncClient(47 base_url=self.settings.OPENROUTER_BASE_URL,48 timeout=httpx.Timeout(self.settings.LLM_TIMEOUT_SECONDS, connect=15),49 http2=True,50 )51 return self._client5253 async def aclose(self) -> None:54 if self._client and not self._client.is_closed:55 await self._client.aclose()5657 # ------------------------------------------------------------------ body58 def _body(59 self,60 messages: list[dict[str, Any]],61 tools: list[dict[str, Any]] | None,62 models: list[str],63 temperature: float,64 max_tokens: int,65 reasoning: dict[str, Any] | None,66 provider: dict[str, Any] | None,67 user_id_hash: str | None,68 stream: bool,69 response_format: dict[str, Any] | None,70 ) -> dict[str, Any]:71 body: dict[str, Any] = {72 "messages": messages,73 "stream": stream,74 "temperature": temperature,75 "max_tokens": max_tokens,76 "usage": {"include": True},77 "provider": provider or {"data_collection": "deny", "allow_fallbacks": True},78 }79 if len(models) == 1:80 body["model"] = models[0]81 else:82 body["model"] = models[0]83 body["models"] = models84 if tools:85 body["tools"] = tools86 body["tool_choice"] = "auto"87 if reasoning:88 body["reasoning"] = reasoning89 if user_id_hash:90 body["user"] = user_id_hash91 if response_format:92 body["response_format"] = response_format93 return body9495 # --------------------------------------------------------------- stream96 async def stream_chat(97 self,98 messages: list[dict[str, Any]],99 tools: list[dict[str, Any]] | None,100 models: list[str],101 temperature: float = 0.3,102 max_tokens: int | None = None,103 reasoning: dict[str, Any] | None = None,104 provider: dict[str, Any] | None = None,105 user_id_hash: str | None = None,106 response_format: dict[str, Any] | None = None,107 ) -> AsyncIterator[StreamEvent]:108 body = self._body(109 messages, tools, models, temperature,110 max_tokens or self.settings.LLM_MAX_OUTPUT_TOKENS,111 reasoning, provider, user_id_hash, True, response_format,112 )113 attempt = 0114 while True:115 attempt += 1116 try:117 async with self.client().stream(118 "POST", "/chat/completions", headers=self._headers(), json=body119 ) as resp:120 if resp.status_code in RETRY_STATUS and attempt < 3:121 await resp.aread()122 log.warning("llm_retry", status=resp.status_code, attempt=attempt)123 await asyncio.sleep(1.5 * attempt)124 continue125 if resp.status_code >= 400:126 text = (await resp.aread()).decode(errors="replace")127 raise LLMError(f"OpenRouter {resp.status_code}: {text[:400]}",128 resp.status_code)129 async for ev in self._parse_sse(resp):130 yield ev131 return132 except (httpx.TransportError, httpx.TimeoutException) as exc:133 if attempt >= 3:134 yield StreamEvent("error", {"code": "network", "message": str(exc)})135 return136 log.warning("llm_network_retry", attempt=attempt, error=str(exc))137 await asyncio.sleep(1.5 * attempt)138139 async def _parse_sse(self, resp: httpx.Response) -> AsyncIterator[StreamEvent]:140 """Parse OpenRouter SSE. Yields normalised StreamEvents."""141 tool_buf: dict[int, dict[str, Any]] = {}142 finish_reason: str | None = None143 model_used = ""144 gen_id = ""145 usage: dict[str, Any] | None = None146 async for raw_line in resp.aiter_lines():147 line = raw_line.strip()148 if not line or line.startswith(":"): # keep-alive ": OPENROUTER PROCESSING"149 continue150 if not line.startswith("data:"):151 continue152 data = line[5:].strip()153 if data == "[DONE]":154 break155 try:156 chunk = json.loads(data)157 except json.JSONDecodeError:158 continue159 if "error" in chunk and chunk["error"]:160 err = chunk["error"]161 yield StreamEvent("error", {"code": str(err.get("code", "upstream")),162 "message": str(err.get("message", err))})163 return164 model_used = chunk.get("model") or model_used165 gen_id = chunk.get("id") or gen_id166 if chunk.get("usage"):167 usage = chunk["usage"]168 for choice in chunk.get("choices", []) or []:169 delta = choice.get("delta") or {}170 if delta.get("content"):171 yield StreamEvent("text_delta", {"delta": delta["content"]})172 if delta.get("reasoning"):173 yield StreamEvent("reasoning_delta", {"delta": delta["reasoning"]})174 for tc in delta.get("tool_calls") or []:175 idx = int(tc.get("index", 0))176 buf = tool_buf.get(idx)177 if buf is None:178 buf = {"id": tc.get("id") or f"call_{idx}", "name": "", "args": ""}179 tool_buf[idx] = buf180 yield StreamEvent("tool_call_start", {"index": idx, "id": buf["id"]})181 if tc.get("id"):182 buf["id"] = tc["id"]183 fn = tc.get("function") or {}184 if fn.get("name") and not buf["name"]:185 buf["name"] = fn["name"]186 if fn.get("arguments"):187 buf["args"] += fn["arguments"]188 yield StreamEvent("tool_call_delta", {"index": idx, "delta": fn["arguments"]})189 if choice.get("finish_reason"):190 finish_reason = choice["finish_reason"]191 for idx in sorted(tool_buf):192 b = tool_buf[idx]193 yield StreamEvent("tool_call_end", {"index": idx, "id": b["id"], "name": b["name"],194 "arguments": b["args"]})195 if usage:196 yield StreamEvent("usage", {197 "input_tokens": usage.get("prompt_tokens", 0),198 "output_tokens": usage.get("completion_tokens", 0),199 "cost_usd": float(usage.get("cost", 0.0) or 0.0),200 "model": model_used,201 "generation_id": gen_id,202 })203 if finish_reason is None and tool_buf:204 finish_reason = "tool_calls"205 yield StreamEvent("done", {"finish_reason": finish_reason or "stop", "model": model_used,206 "generation_id": gen_id})207208 # ------------------------------------------------------------- complete209 async def complete(210 self,211 messages: list[dict[str, Any]],212 models: list[str],213 temperature: float = 0.0,214 max_tokens: int = 1024,215 response_format: dict[str, Any] | None = None,216 user_id_hash: str | None = None,217 ) -> tuple[str, dict[str, Any]]:218 """Non-streaming helper for small tasks (titles, classification, quiz JSON)."""219 body = self._body(messages, None, models, temperature, max_tokens, None, None,220 user_id_hash, False, response_format)221 for attempt in range(1, 4):222 try:223 resp = await self.client().post("/chat/completions", headers=self._headers(),224 json=body)225 except (httpx.TransportError, httpx.TimeoutException) as exc:226 if attempt == 3:227 raise LLMError(f"network: {exc}") from exc228 await asyncio.sleep(1.5 * attempt)229 continue230 if resp.status_code in RETRY_STATUS and attempt < 3:231 await asyncio.sleep(1.5 * attempt)232 continue233 if resp.status_code >= 400:234 raise LLMError(f"OpenRouter {resp.status_code}: {resp.text[:400]}", resp.status_code)235 data = resp.json()236 choice = (data.get("choices") or [{}])[0]237 text = (choice.get("message") or {}).get("content") or ""238 usage = data.get("usage") or {}239 return text, {240 "input_tokens": usage.get("prompt_tokens", 0),241 "output_tokens": usage.get("completion_tokens", 0),242 "cost_usd": float(usage.get("cost", 0.0) or 0.0),243 "model": data.get("model", ""),244 "generation_id": data.get("id", ""),245 }246 raise LLMError("OpenRouter: retries exhausted")247248 async def embed(self, texts: list[str]) -> list[list[float]] | None:249 """Optional embeddings via an OpenAI-compatible endpoint (EMBEDDINGS_BASE_URL)."""250 s = self.settings251 if not s.EMBEDDINGS_BASE_URL or not s.MODEL_EMBEDDINGS:252 return None253 headers = {"Content-Type": "application/json"}254 key = s.EMBEDDINGS_API_KEY.get_secret_value() or s.OPENROUTER_API_KEY.get_secret_value()255 if key:256 headers["Authorization"] = f"Bearer {key}"257 async with httpx.AsyncClient(timeout=60) as c:258 r = await c.post(f"{s.EMBEDDINGS_BASE_URL.rstrip('/')}/embeddings", headers=headers,259 json={"model": s.MODEL_EMBEDDINGS, "input": texts})260 r.raise_for_status()261 data = r.json()["data"]262 return [d["embedding"] for d in sorted(data, key=lambda d: d["index"])]263264265_client: LLMClient | None = None266267268def get_llm() -> LLMClient:269 global _client270 if _client is None:271 _client = LLMClient()272 return _client273