import json import re from fastapi import HTTPException import httpx from app.core.config import get_settings _REASONING_MODEL_MIN_COMPLETION_TOKENS = 1024 def build_llm_headers() -> dict[str, str]: settings = get_settings() headers = {"Content-Type": "application/json", "User-Agent": "PowerUp/1.0"} if settings.llm_api_key: headers["Authorization"] = f"Bearer {settings.llm_api_key}" return headers def build_llm_timeout() -> httpx.Timeout: settings = get_settings() timeout_seconds = max(settings.llm_request_timeout_seconds, 5.0) connect_timeout = min(timeout_seconds, 10.0) return httpx.Timeout(timeout_seconds, connect=connect_timeout) def _normalize_error_text(value: str) -> str: return " ".join((value or "").split()).strip() def _looks_like_html_error(value: str) -> bool: normalized = value.casefold() return ( "" in normalized or bool(re.search(r"<[a-z][^>]*>", normalized)) ) def _fallback_http_detail(status_code: int) -> str: if status_code == 404: return "endpoint LLM non raggiungibile" if status_code == 401: return "autenticazione LLM non valida" if status_code == 403: return "accesso al servizio LLM negato" if status_code >= 500: return "servizio LLM temporaneamente non disponibile" return "risposta upstream non valida" def describe_llm_http_error(exc: httpx.HTTPError) -> str: if isinstance(exc, httpx.HTTPStatusError): response = exc.response detail = "" try: payload = response.json() if isinstance(payload, dict): error_payload = payload.get("error") if isinstance(error_payload, dict): detail = str(error_payload.get("message") or error_payload.get("detail") or "").strip() if not detail: detail = str(payload.get("detail") or payload.get("message") or "").strip() if not detail: detail = json.dumps(payload, ensure_ascii=True) except Exception: detail = response.text.strip() detail = _normalize_error_text(detail)[:400] if _looks_like_html_error(detail): detail = _fallback_http_detail(response.status_code) if detail: return f"HTTP {response.status_code}: {detail}" return f"HTTP {response.status_code}: {_fallback_http_detail(response.status_code)}" request = getattr(exc, "request", None) request_url = str(request.url) if request is not None else "" message = str(exc).strip() error_name = exc.__class__.__name__ if message and request_url: return f"{error_name} verso {request_url}: {message}" if message: return f"{error_name}: {message}" if request_url: return f"{error_name} verso {request_url}" return error_name async def resolve_llm_model(client: httpx.AsyncClient, base_url: str, headers: dict[str, str]) -> str: settings = get_settings() if settings.assistant_model: return settings.assistant_model response = await client.get(f"{base_url}/models", headers=headers) response.raise_for_status() payload = response.json() for item in payload.get("data", []): if isinstance(item, dict) and item.get("id"): return str(item["id"]) raise HTTPException(status_code=502, detail="Nessun modello disponibile sull'endpoint LLM configurato") def _clean_model_name(value: str | None) -> str | None: cleaned = (value or "").strip() return cleaned or None def _estimate_prompt_tokens(messages: list[dict[str, str]]) -> int: # Groq applies TPM to prompt tokens plus requested completion tokens. Keep this conservative. serialized = json.dumps(messages, ensure_ascii=False) return max(1, (len(serialized) // 3) + (len(messages) * 16)) def _model_request_token_limit(model: str) -> int | None: normalized = model.lower().strip() if "qwen/qwen3-32b" in normalized: return 6000 if "gpt-oss" in normalized: return 8000 return None def _strip_reasoning_blocks(reply: str) -> str: cleaned = re.sub(r".*?", "", reply, flags=re.IGNORECASE | re.DOTALL).strip() return cleaned or reply.strip() def _model_needs_reasoning_token_floor(model: str) -> bool: normalized = model.lower() return "gpt-oss" in normalized or "qwen/qwen3" in normalized def _effective_max_tokens( *, model: str, messages: list[dict[str, str]], requested_max_tokens: int | None, ) -> int | None: settings = get_settings() configured_max_tokens = settings.assistant_max_tokens if settings.assistant_max_tokens > 0 else None if requested_max_tokens is None: target_max_tokens = configured_max_tokens else: target_max_tokens = requested_max_tokens if _model_needs_reasoning_token_floor(model): target_max_tokens = max(target_max_tokens, _REASONING_MODEL_MIN_COMPLETION_TOKENS) if configured_max_tokens is not None: target_max_tokens = min(target_max_tokens, configured_max_tokens) if target_max_tokens is None: return None request_token_limit = _model_request_token_limit(model) if request_token_limit is None: return target_max_tokens prompt_tokens = _estimate_prompt_tokens(messages) safe_completion_tokens = request_token_limit - prompt_tokens - 256 if safe_completion_tokens <= 0: return 1 return max(1, min(target_max_tokens, safe_completion_tokens)) def _rate_limit_retry_max_tokens( *, model: str, messages: list[dict[str, str]], requested_max_tokens: int | None, ) -> int | None: current_max_tokens = _effective_max_tokens( model=model, messages=messages, requested_max_tokens=requested_max_tokens, ) if current_max_tokens is None or current_max_tokens <= _REASONING_MODEL_MIN_COMPLETION_TOKENS: return None return max(_REASONING_MODEL_MIN_COMPLETION_TOKENS, min(2048, current_max_tokens // 2)) def extract_llm_reply(payload: dict[str, object]) -> str: reply = payload.get("choices", [{}])[0].get("message", {}).get("content") # type: ignore[index] if not reply: reply = payload.get("choices", [{}])[0].get("text") # type: ignore[index] if not reply: finish_reason = payload.get("choices", [{}])[0].get("finish_reason") # type: ignore[index] if finish_reason == "length": raise HTTPException(status_code=502, detail="Risposta LLM vuota: limite output consumato dal reasoning") raise HTTPException(status_code=502, detail="Risposta LLM priva di contenuto utilizzabile") return _strip_reasoning_blocks(str(reply)) def _build_chat_payload( *, model: str, messages: list[dict[str, str]], temperature: float | None, max_tokens: int | None, ) -> dict[str, object]: settings = get_settings() payload: dict[str, object] = { "model": model, "temperature": settings.assistant_temperature if temperature is None else temperature, "messages": messages, } effective_max_tokens = _effective_max_tokens( model=model, messages=messages, requested_max_tokens=max_tokens, ) if effective_max_tokens is not None: if "gpt-oss" in model.lower(): payload["max_completion_tokens"] = effective_max_tokens else: payload["max_tokens"] = effective_max_tokens if "gpt-oss" in model.lower(): reasoning_effort = (settings.assistant_reasoning_effort or "").strip() if reasoning_effort: payload["reasoning_effort"] = reasoning_effort payload["include_reasoning"] = bool(settings.assistant_include_reasoning) payload["disable_tool_validation"] = True return payload async def request_llm_chat_completion( messages: list[dict[str, str]], *, temperature: float | None = None, max_tokens: int | None = None, model: str | None = None, ) -> tuple[str, str]: settings = get_settings() base_url = settings.llm_base_url.rstrip("/") headers = build_llm_headers() try: async with httpx.AsyncClient(timeout=build_llm_timeout()) as client: resolved_model = model or await resolve_llm_model(client, base_url, headers) async def _post_chat_completion(active_model: str, requested_max_tokens: int | None) -> httpx.Response: return await client.post( f"{base_url}/chat/completions", headers=headers, json=_build_chat_payload( model=active_model, messages=messages, temperature=temperature, max_tokens=requested_max_tokens, ), ) try: response = await _post_chat_completion(resolved_model, max_tokens) response.raise_for_status() except httpx.HTTPStatusError as exc: if exc.response.status_code == 429: reduced_max_tokens = _rate_limit_retry_max_tokens( model=resolved_model, messages=messages, requested_max_tokens=max_tokens, ) if reduced_max_tokens is not None: try: response = await _post_chat_completion(resolved_model, reduced_max_tokens) response.raise_for_status() except httpx.HTTPStatusError as retry_exc: exc = retry_exc else: data = response.json() return extract_llm_reply(data), resolved_model fallback_model = _clean_model_name(settings.assistant_fallback_model) if exc.response.status_code != 429 or not fallback_model or fallback_model == resolved_model: raise resolved_model = fallback_model fallback_max_tokens = ( _rate_limit_retry_max_tokens( model=resolved_model, messages=messages, requested_max_tokens=max_tokens, ) or max_tokens ) response = await _post_chat_completion(resolved_model, fallback_max_tokens) response.raise_for_status() except httpx.HTTPError as exc: raise HTTPException(status_code=502, detail=f"Errore comunicazione LLM: {describe_llm_http_error(exc)}") from exc data = response.json() return extract_llm_reply(data), resolved_model