"""LLM client adapters for different providers.""" import logging from typing import Dict, Any, List, Optional, Union from abc import ABC, abstractmethod from dataclasses import dataclass from langchain_openai.chat_models import ChatOpenAI from ..config.loader import load_config logger = logging.getLogger(__name__) class QuotaExhaustedError(RuntimeError): """Raised when the LLM provider reports a *permanent* failure that retrying cannot fix: account quota exhausted, billing disabled, or invalid credentials. The orchestrator catches this at the agent boundary to short-circuit the LangGraph state machine and return a user-visible message, instead of falling through with empty fallbacks and looping until ``GraphRecursionError`` (which the system previously did, costing ~100+ wasted HTTP calls per user query — see DEFERRED / CHANGELOG). Transient errors (rate-limit-without-quota-exhaustion, HTTP 5xx, network blips) deliberately do NOT raise this — they keep the current retry-and-fallback behaviour because they often recover on their own. """ def __init__(self, message: str, *, model: str = None, error_code: str = None, http_status: int = None): super().__init__(message) self.model = model self.error_code = error_code self.http_status = http_status # Load configuration once at module level _config = load_config() # Legacy client factory functions (inlined from auditqa_old.reader) def _create_inf_provider_client(): """Create INF_PROVIDERS client.""" reader_config = _config.get("reader", {}) inf_config = reader_config.get("INF_PROVIDERS", {}) api_key = inf_config.get("api_key") if not api_key: raise ValueError("INF_PROVIDERS api_key not found in configuration") provider = inf_config.get("provider") if not provider: raise ValueError("INF_PROVIDERS provider not found in configuration") from huggingface_hub import InferenceClient return InferenceClient( provider=provider, api_key=api_key, bill_to="GIZ", ) def _create_nvidia_client(): """Create NVIDIA client.""" from huggingface_hub import InferenceClient reader_config = _config.get("reader", {}) nvidia_config = reader_config.get("NVIDIA", {}) api_key = nvidia_config.get("api_key") if not api_key: raise ValueError("NVIDIA api_key not found in configuration") endpoint = nvidia_config.get("endpoint") if not endpoint: raise ValueError("NVIDIA endpoint not found in configuration") return InferenceClient( base_url=endpoint, api_key=api_key ) def _create_serverless_client(): """Create serverless API client.""" from huggingface_hub import InferenceClient reader_config = _config.get("reader", {}) serverless_config = reader_config.get("SERVERLESS", {}) api_key = serverless_config.get("api_key") if not api_key: raise ValueError("SERVERLESS api_key not found in configuration") model_id = serverless_config.get("model", "meta-llama/Meta-Llama-3-8B-Instruct") return InferenceClient( model=model_id, api_key=api_key, ) def _create_dedicated_endpoint_client(): """Create dedicated endpoint client.""" from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler from langchain_community.llms import HuggingFaceEndpoint from langchain_community.chat_models.huggingface import ChatHuggingFace reader_config = _config.get("reader", {}) dedicated_config = reader_config.get("DEDICATED", {}) api_key = dedicated_config.get("api_key") if not api_key: raise ValueError("DEDICATED api_key not found in configuration") endpoint = dedicated_config.get("endpoint") if not endpoint: raise ValueError("DEDICATED endpoint not found in configuration") max_tokens = dedicated_config.get("max_tokens", 768) callback = StreamingStdOutCallbackHandler() llm_qa = HuggingFaceEndpoint( endpoint_url=endpoint, max_new_tokens=int(max_tokens), repetition_penalty=1.03, timeout=70, huggingfacehub_api_token=api_key, streaming=True, callbacks=[callback] ) return ChatHuggingFace(llm=llm_qa) @dataclass class LLMResponse: """Standardized LLM response format.""" content: str model: str provider: str metadata: Dict[str, Any] = None class BaseLLMAdapter(ABC): """Base class for LLM adapters.""" def __init__(self, config: Dict[str, Any]): self.config = config def _log_usage( self, provider: str, model: str, messages: List[Dict[str, str]], usage: Optional[Dict[str, Any]] = None, prompt_name: Optional[str] = None, ) -> None: """ Lightweight usage logger for cost estimation. - Writes CSV lines to logs/llm_usage.csv (created if missing). - Records timestamp, provider, model, prompt_name, num_messages, input_chars, input_tokens/output_tokens if provided by the SDK. """ try: from datetime import datetime from pathlib import Path from src.config.paths import PROJECT_DIR log_dir = PROJECT_DIR / "logs" log_dir.mkdir(parents=True, exist_ok=True) log_file = log_dir / "llm_usage.csv" ts = datetime.utcnow().isoformat() num_messages = len(messages) if messages else 0 input_chars = sum(len(m.get("content", "")) for m in messages) if messages else 0 input_tokens = None output_tokens = None if usage: input_tokens = usage.get("input_tokens") or usage.get("prompt_tokens") output_tokens = usage.get("output_tokens") or usage.get("completion_tokens") line = [ ts, provider or "", model or "", prompt_name or "", str(num_messages), str(input_chars), "" if input_tokens is None else str(input_tokens), "" if output_tokens is None else str(output_tokens), ] header = "timestamp,provider,model,prompt_name,num_messages,input_chars,input_tokens,output_tokens\n" line_str = ",".join(item.replace(",", " ") for item in line) + "\n" if not log_file.exists(): log_file.write_text(header) with log_file.open("a") as f: f.write(line_str) except Exception: # Never break inference on logging failures pass @abstractmethod def generate(self, messages: List[Dict[str, str]], **kwargs) -> LLMResponse: """Generate response from messages.""" pass @abstractmethod def stream_generate(self, messages: List[Dict[str, str]], **kwargs): """Generate streaming response from messages.""" pass class MistralAdapter(BaseLLMAdapter): """Adapter for Mistral AI models.""" def __init__(self, config: Dict[str, Any]): super().__init__(config) from langchain_mistralai.chat_models import ChatMistralAI self.model = ChatMistralAI( model=config.get("model", "mistral-medium-latest") ) def generate(self, messages: List[Dict[str, str]], **kwargs) -> LLMResponse: """Generate response using Mistral.""" response = self.model.invoke(messages) usage_meta = getattr(response, 'usage_metadata', {}) or {} self._log_usage( provider="mistral", model=self.config.get("model", "mistral-medium-latest"), messages=messages, usage=usage_meta, prompt_name=kwargs.get("prompt_name"), ) return LLMResponse( content=response.content, model=self.config.get("model", "mistral-medium-latest"), provider="mistral", metadata={"usage": usage_meta} ) def stream_generate(self, messages: List[Dict[str, str]], **kwargs): """Generate streaming response using Mistral.""" for chunk in self.model.stream(messages): if chunk.content: yield chunk.content class OpenAIAdapter(BaseLLMAdapter): """Adapter for OpenAI models. On construction, the adapter sends ONE tiny probe call (1 input token, 1 output token, cost ~$0.0001) to the configured model with the configured kwargs. If OpenAI rejects any parameter — for example a reasoning-class model like ``gpt-5`` rejects custom ``temperature`` — the offending parameter is dropped (parsed from the structured error body) and the probe is retried. The first set of kwargs that the API accepts is cached per (model, requested_kwargs) tuple at the class level, so subsequent ``OpenAIAdapter`` instances for the same config reuse it without a second probe. This means the system self-tunes to any current or future OpenAI model without hardcoded "this model accepts X" knowledge living in our codebase, and per-query latency is unaffected (the probe runs once at startup, not per request). """ # Cache key: (model, sorted_kwargs_tuple) → dict of working kwargs. _PROBE_CACHE: Dict[tuple, Dict[str, Any]] = {} _MAX_PROBE_ATTEMPTS = 5 # The probe needs enough budget for reasoning-class models to do # their internal "thinking" before producing visible output. With a # tiny budget (e.g. 1), gpt-5 / o1 / o3 will return HTTP 400 # "Could not finish the message because max_tokens or model output # limit was reached". 256 tokens gives ~100 visible tokens of headroom # after reasoning, costs ~$0.0026 per probe at gpt-5 pricing. _PROBE_MAX_COMPLETION_TOKENS = 256 def __init__(self, config: Dict[str, Any]): super().__init__(config) model_name = config.get("model", "gpt-4o-mini") requested_kwargs: Dict[str, Any] = {"model": model_name} if "temperature" in config: requested_kwargs["temperature"] = config["temperature"] if "max_tokens" in config: requested_kwargs["max_tokens"] = config["max_tokens"] working_kwargs = self._probe_supported_kwargs(requested_kwargs) self.model = ChatOpenAI(**working_kwargs) @classmethod def _probe_supported_kwargs(cls, requested: Dict[str, Any]) -> Dict[str, Any]: """Self-tune kwargs by sending a 1-token probe and dropping any parameter the API rejects with ``unsupported_value``.""" cache_key = (requested.get("model"), tuple(sorted(requested.items()))) if cache_key in cls._PROBE_CACHE: return dict(cls._PROBE_CACHE[cache_key]) # Lazy import — only loaded when an OpenAIAdapter is actually built. import openai client = openai.OpenAI() kwargs = dict(requested) dropped_params = [] for attempt in range(cls._MAX_PROBE_ATTEMPTS): try: client.chat.completions.create( messages=[{"role": "user", "content": "ping"}], max_completion_tokens=cls._PROBE_MAX_COMPLETION_TOKENS, **{k: v for k, v in kwargs.items() if k != "max_tokens"}, ) if dropped_params: logger.warning( "OpenAIAdapter probe: model '%s' rejected parameter(s) %s; " "using model defaults for those. Working kwargs: %s", kwargs.get("model"), dropped_params, {k: v for k, v in kwargs.items() if k != "model"}, ) cls._PROBE_CACHE[cache_key] = dict(kwargs) return dict(kwargs) except openai.BadRequestError as exc: rejected = cls._extract_rejected_param(exc) if rejected and rejected in kwargs: dropped_params.append((rejected, kwargs.pop(rejected))) continue logger.error( "OpenAIAdapter probe failed with un-handleable 400 for model " "'%s': %s. Caller will likely hit the same error on first real " "request.", kwargs.get("model"), str(exc)[:300], ) break except Exception as exc: logger.warning( "OpenAIAdapter probe could not run for model '%s' (%s: %s). " "Proceeding with requested kwargs as-is; if they're rejected, " "actual queries will fail with the same error.", kwargs.get("model"), type(exc).__name__, str(exc)[:200], ) break cls._PROBE_CACHE[cache_key] = dict(kwargs) return dict(kwargs) @staticmethod def _extract_rejected_param(exc: Exception) -> Optional[str]: """Pull the parameter name out of an OpenAI ``unsupported_value`` error.""" body = getattr(exc, "body", None) if isinstance(body, dict): err = body.get("error", {}) if isinstance(err, dict): if err.get("code") == "unsupported_value" and err.get("param"): return err["param"] msg = str(exc).lower() for candidate in ("temperature", "top_p", "presence_penalty", "frequency_penalty", "max_tokens"): if candidate in msg and ("unsupported" in msg or "does not support" in msg): return candidate return None # OpenAI error.code values that we treat as PERMANENT (no point # retrying — raise QuotaExhaustedError so the orchestrator can # short-circuit instead of letting the graph loop). Transient # rate-limits (HTTP 429 WITHOUT one of these codes) deliberately # stay on the existing retry-with-fallback path. _PERMANENT_ERROR_CODES = frozenset({ "insufficient_quota", # account out of credits / billing issue "billing_hard_limit_reached", # explicit billing cap hit "billing_not_active", # account disabled "invalid_api_key", # bad key }) # HTTP status codes that are permanent (no point retrying) _PERMANENT_HTTP_STATUSES = frozenset({401, 403}) @classmethod def _classify_permanent_error(cls, exc: Exception) -> Optional[tuple]: """If ``exc`` is a permanent (don't-retry) OpenAI error, return ``(error_code, http_status)``. Otherwise return ``None`` (caller should let the existing retry/fallback path handle it). """ body = getattr(exc, "body", None) status = getattr(exc, "status_code", None) or getattr(exc, "http_status", None) # Structured-body check first (most reliable). if isinstance(body, dict): err = body.get("error") or {} if isinstance(err, dict): code = err.get("code") or err.get("type") if code in cls._PERMANENT_ERROR_CODES: return (code, status) # Then status-code check (auth errors). if status in cls._PERMANENT_HTTP_STATUSES: return ("http_" + str(status), status) # Fallback: substring match on the message (less reliable but # catches cases where openai library doesn't populate body). msg = str(exc).lower() for code in cls._PERMANENT_ERROR_CODES: if code in msg: return (code, status) return None def generate(self, messages: List[Dict[str, str]], **kwargs) -> LLMResponse: """Generate response using OpenAI. Raises: QuotaExhaustedError: if OpenAI reports a permanent error (quota exhausted, billing disabled, invalid key). Callers upstream catch this to short-circuit the LangGraph state machine. Transient errors (5xx, brief rate-limits) are NOT caught here — the SDK's built-in retry handles them. """ import time as _t t0 = _t.time() try: response = self.model.invoke(messages) except Exception as exc: classified = self._classify_permanent_error(exc) if classified is not None: code, status = classified model_name = self.config.get("model", "gpt-4o-mini") raise QuotaExhaustedError( f"OpenAI returned permanent error for model {model_name!r}: " f"code={code} status={status} message={str(exc)[:200]}", model=model_name, error_code=code, http_status=status, ) from exc raise elapsed = _t.time() - t0 usage_meta = getattr(response, 'usage_metadata', {}) or {} model_name = self.config.get("model", "gpt-4o-mini") prompt_name = kwargs.get("prompt_name") # CSV usage log on disk (legacy aggregation) self._log_usage( provider="openai", model=model_name, messages=messages, usage=usage_meta, prompt_name=prompt_name, ) # Concise one-line stdout log for prod observability — visible in # HF Space logs, no external service needed. self._log_call_stdout(model_name, prompt_name, usage_meta, elapsed) return LLMResponse( content=response.content, model=model_name, provider="openai", metadata={"usage": usage_meta}, ) # Per-1M-token USD pricing for the OpenAI models we use. Update when # OpenAI changes pricing. The dict is consulted only for the stdout # cost-estimate log line; missing models log without a cost figure. _PRICING_USD_PER_1M = { "gpt-4o-mini": {"prompt": 0.150, "completion": 0.600}, "gpt-4o": {"prompt": 2.500, "completion": 10.000}, "gpt-4.1": {"prompt": 2.000, "completion": 8.000}, "gpt-4.1-mini": {"prompt": 0.400, "completion": 1.600}, "gpt-5": {"prompt": 1.250, "completion": 10.000}, "gpt-5-mini": {"prompt": 0.250, "completion": 2.000}, "gpt-5-chat-latest": {"prompt": 1.250, "completion": 10.000}, } @classmethod def _log_call_stdout(cls, model: str, prompt_name: Optional[str], usage: Dict[str, Any], elapsed_s: float = 0.0) -> None: """One-line per-call summary. Designed to be readable at a glance in HF Space / Docker logs alongside the [L1.main] [L4.response] agent-level breadcrumbs.""" try: in_tok = (usage.get("input_tokens") or usage.get("prompt_tokens") or 0) out_tok = (usage.get("output_tokens") or usage.get("completion_tokens") or 0) details_in = usage.get("input_token_details") or {} details_out = usage.get("output_token_details") or {} cached = details_in.get("cache_read", 0) or 0 reasoning = details_out.get("reasoning", 0) or 0 prices = cls._PRICING_USD_PER_1M.get(model) cost_str = "" if prices and (in_tok or out_tok): non_cached = max(0, in_tok - cached) cost = ( non_cached / 1_000_000 * prices["prompt"] + cached / 1_000_000 * prices["prompt"] * 0.5 + out_tok / 1_000_000 * prices["completion"] ) cost_str = f" cost=${cost:.4f}" hit_str = "" if in_tok > 0: hit_pct = cached / in_tok * 100 hit_str = f" cache_hit={cached}/{in_tok}({hit_pct:.0f}%)" tag = f" tag={prompt_name}" if prompt_name else "" reasoning_str = f" reasoning={reasoning}" if reasoning else "" elapsed_str = f" took={elapsed_s:.1f}s" if elapsed_s > 0 else "" logger.info( "[LLM] model=%s%s in=%d out=%d%s%s%s%s", model, elapsed_str, in_tok, out_tok, reasoning_str, hit_str, cost_str, tag, ) except Exception: # Logging must never crash inference. pass def stream_generate(self, messages: List[Dict[str, str]], **kwargs): """Generate streaming response using OpenAI.""" for chunk in self.model.stream(messages): if chunk.content: yield chunk.content class OllamaAdapter(BaseLLMAdapter): """Adapter for Ollama models.""" def __init__(self, config: Dict[str, Any]): super().__init__(config) from langchain_ollama import ChatOllama self.model = ChatOllama( model=config.get("model", "mistral-small3.1:24b-instruct-2503-q8_0"), base_url=config.get("base_url", "http://localhost:11434/"), temperature=config.get("temperature", 0.8), num_predict=config.get("num_predict", 256) ) def generate(self, messages: List[Dict[str, str]], **kwargs) -> LLMResponse: """Generate response using Ollama.""" response = self.model.invoke(messages) usage_meta = getattr(response, 'usage_metadata', {}) or {} self._log_usage( provider="ollama", model=self.config.get("model", "mistral-small3.1:24b-instruct-2503-q8_0"), messages=messages, usage=usage_meta, prompt_name=kwargs.get("prompt_name"), ) return LLMResponse( content=response.content, model=self.config.get("model", "mistral-small3.1:24b-instruct-2503-q8_0"), provider="ollama", metadata=usage_meta ) def stream_generate(self, messages: List[Dict[str, str]], **kwargs): """Generate streaming response using Ollama.""" for chunk in self.model.stream(messages): if chunk.content: yield chunk.content class OpenRouterAdapter(BaseLLMAdapter): """Adapter for OpenRouter models.""" def __init__(self, config: Dict[str, Any]): super().__init__(config) # Prepare custom headers for OpenRouter (optional) headers = {} if config.get("site_url"): headers["HTTP-Referer"] = config["site_url"] if config.get("site_name"): headers["X-Title"] = config["site_name"] # Initialize ChatOpenAI with OpenRouter configuration self.model = ChatOpenAI( model=config.get("model", "openai/gpt-3.5-turbo"), api_key=config.get("api_key"), base_url=config.get("base_url", "https://openrouter.ai/api/v1"), default_headers= headers if headers else {}, temperature=config.get("temperature", 0.7), max_tokens=config.get("max_tokens", 1000) ) def generate(self, messages: List[Dict[str, str]], **kwargs) -> LLMResponse: """Generate response using OpenRouter.""" response = self.model.invoke(messages) usage_meta = getattr(response, 'usage_metadata', {}) or {} self._log_usage( provider="openrouter", model=self.config.get("model", "openai/gpt-3.5-turbo"), messages=messages, usage=usage_meta, prompt_name=kwargs.get("prompt_name"), ) return LLMResponse( content=response.content, model=self.config.get("model", "openai/gpt-3.5-turbo"), provider="openrouter", metadata={"usage": usage_meta} ) def stream_generate(self, messages: List[Dict[str, str]], **kwargs): """Generate streaming response using OpenRouter.""" for chunk in self.model.stream(messages): if chunk.content: yield chunk.content class LegacyAdapter(BaseLLMAdapter): """Adapter for legacy LLM clients (INF_PROVIDERS, NVIDIA, etc.).""" def __init__(self, config: Dict[str, Any], client_type: str): super().__init__(config) self.client_type = client_type self.client = self._create_client() def _create_client(self): """Create legacy client based on type.""" if self.client_type == "INF_PROVIDERS": return _create_inf_provider_client() elif self.client_type == "NVIDIA": return _create_nvidia_client() elif self.client_type == "DEDICATED": return _create_dedicated_endpoint_client() else: # SERVERLESS return _create_serverless_client() def generate(self, messages: List[Dict[str, str]], **kwargs) -> LLMResponse: """Generate response using legacy client.""" max_tokens = kwargs.get('max_tokens', self.config.get('max_tokens', 768)) if self.client_type == "INF_PROVIDERS": response = self.client.chat.completions.create( model=self.config.get("model"), messages=messages, max_tokens=max_tokens ) content = response.choices[0].message.content elif self.client_type == "NVIDIA": response = self.client.chat_completion( model=self.config.get("model"), messages=messages, max_tokens=max_tokens ) content = response.choices[0].message.content else: # DEDICATED or SERVERLESS response = self.client.chat_completion( messages=messages, max_tokens=max_tokens ) content = response.choices[0].message.content return LLMResponse( content=content, model=self.config.get("model", "unknown"), provider=self.client_type.lower(), metadata={} ) def stream_generate(self, messages: List[Dict[str, str]], **kwargs): """Generate streaming response using legacy client.""" # Legacy clients may not support streaming in the same way # This is a simplified implementation response = self.generate(messages, **kwargs) words = response.content.split() for word in words: yield word + " " class LLMRegistry: """Registry for managing different LLM adapters.""" def __init__(self): self.adapters = {} self.adapter_configs = {} def register_adapter(self, name: str, adapter_class: type, config: Dict[str, Any]): """Register an LLM adapter (lazy instantiation).""" self.adapter_configs[name] = (adapter_class, config) def get_adapter(self, name: str) -> BaseLLMAdapter: """Get an LLM adapter by name (lazy instantiation).""" if name not in self.adapter_configs: raise ValueError(f"Unknown LLM adapter: {name}") # Lazy instantiation - only create when needed if name not in self.adapters: adapter_class, config = self.adapter_configs[name] self.adapters[name] = adapter_class(config) return self.adapters[name] def list_adapters(self) -> List[str]: """List available adapter names.""" return list(self.adapter_configs.keys()) def create_llm_registry(config: Dict[str, Any]) -> LLMRegistry: """ Create and populate LLM registry from configuration. Args: config: Configuration dictionary Returns: Populated LLMRegistry """ registry = LLMRegistry() reader_config = config.get("reader", {}) # Register simple adapters if "MISTRAL" in reader_config: registry.register_adapter("mistral", MistralAdapter, reader_config["MISTRAL"]) if "OPENAI" in reader_config: registry.register_adapter("openai", OpenAIAdapter, reader_config["OPENAI"]) if "OPENAI_STRONG" in reader_config: registry.register_adapter("openai_strong", OpenAIAdapter, reader_config["OPENAI_STRONG"]) if "OPENAI_RESPONSE" in reader_config: registry.register_adapter("openai_response", OpenAIAdapter, reader_config["OPENAI_RESPONSE"]) if "OLLAMA" in reader_config: registry.register_adapter("ollama", OllamaAdapter, reader_config["OLLAMA"]) if "OPENROUTER" in reader_config: registry.register_adapter("openrouter", OpenRouterAdapter, reader_config["OPENROUTER"]) # Register legacy adapters # legacy_types = ["INF_PROVIDERS", "NVIDIA", "DEDICATED"] legacy_types = ["INF_PROVIDERS"] for legacy_type in legacy_types: if legacy_type in reader_config: registry.register_adapter( legacy_type.lower(), lambda cfg, lt=legacy_type: LegacyAdapter(cfg, lt), reader_config[legacy_type] ) return registry def get_llm_client(provider: str, config: Dict[str, Any]) -> BaseLLMAdapter: """ Get LLM client for specified provider. Args: provider: Provider name (mistral, openai, ollama, etc.) config: Configuration dictionary Returns: LLM adapter instance """ registry = create_llm_registry(config) return registry.get_adapter(provider)