from __future__ import annotations import json from time import perf_counter from typing import Any from pydantic import BaseModel, ValidationError from app.core.logging import InstrumentedGroqClient, get_audit_logger from app.llm.models import ( LLMErrorKind, LLMProviderAvailability, LLMProviderCapabilities, LLMProviderError, LLMStructuredRequest, LLMStructuredResult, LLMTextRequest, LLMTextResult, ) def _text_size(value: Any) -> int: if value is None: return 0 if isinstance(value, str): return len(value) if isinstance(value, (list, tuple)): return sum(_text_size(item) for item in value) return len(str(value)) class GroqAdapter: def __init__( self, *, api_key: str | None, default_model: str, service: str = "llm_groq_adapter", client: Any | None = None, ) -> None: self._api_key = api_key self._default_model = default_model self._service = service self._client = client self._audit_logger = get_audit_logger() def provider_name(self) -> str: return "groq" def availability(self) -> LLMProviderAvailability: configured = bool(str(self._api_key or "").strip() or self._client is not None) reason = None if configured else "GROQ_API_KEY no configurada" return LLMProviderAvailability(configured=configured, reason=reason) def capabilities(self) -> LLMProviderCapabilities: return LLMProviderCapabilities( text_generation=True, structured_generation=True, native_structured_output=False, ) def close(self) -> None: self._client = None def _client_or_raise(self) -> Any: availability = self.availability() if not availability.configured: raise LLMProviderError( availability.reason or "Proveedor Groq no configurado", kind=LLMErrorKind.INVALID_CONFIGURATION, provider=self.provider_name(), model=self._default_model, ) if self._client is None: from groq import Groq self._client = InstrumentedGroqClient( Groq(api_key=self._api_key), provider=self.provider_name(), service=self._service, ) return self._client def _build_messages(self, request: LLMTextRequest) -> list[dict[str, str]]: messages: list[dict[str, str]] = [] if request.system_prompt: messages.append({"role": "system", "content": request.system_prompt}) messages.append({"role": "user", "content": request.prompt}) return messages def _classify_exception(self, exc: Exception, *, model: str) -> LLMProviderError: text = str(exc) if "429" in text: return LLMProviderError( text or "Groq alcanzó límite de cuota", kind=LLMErrorKind.RATE_LIMITED, provider=self.provider_name(), model=model, retryable=True, ) if "503" in text or "UNAVAILABLE" in text: return LLMProviderError( text or "Groq temporalmente no disponible", kind=LLMErrorKind.TRANSIENT, provider=self.provider_name(), model=model, retryable=True, ) return LLMProviderError( text or "Fallo del proveedor Groq", kind=LLMErrorKind.PROVIDER_UNAVAILABLE, provider=self.provider_name(), model=model, ) def generate_text(self, request: LLMTextRequest) -> LLMTextResult: model = request.model or self._default_model messages = self._build_messages(request) started_at = perf_counter() try: response = self._client_or_raise().chat.completions.create(model=model, messages=messages) text = response.choices[0].message.content or "" metrics = { "duration_ms": round((perf_counter() - started_at) * 1000, 3), "input_chars": _text_size(messages), "output_chars": _text_size(text), } return LLMTextResult( content=text, provider=self.provider_name(), model=model, metrics=metrics, ) except LLMProviderError: raise except Exception as exc: mapped = self._classify_exception(exc, model=model) self._audit_logger.llm_event( action="generate_text", outcome="error", service=self._service, provider=self.provider_name(), model=model, metrics={"duration_ms": round((perf_counter() - started_at) * 1000, 3)}, resource={"operation": "generate_text"}, error=mapped.to_error_payload(), ) raise mapped from exc def generate_structured(self, request: LLMStructuredRequest) -> LLMStructuredResult: model = request.model or self._default_model output_model = request.output_model if not isinstance(output_model, type) or not issubclass(output_model, BaseModel): raise LLMProviderError( "Groq structured output requiere un modelo Pydantic", kind=LLMErrorKind.INVALID_CONFIGURATION, provider=self.provider_name(), model=model, ) schema_json = json.dumps(output_model.model_json_schema(by_alias=True), ensure_ascii=False, separators=(",", ":")) prompt = ( f"{request.prompt}\n\n" "Responde exclusivamente con un único objeto JSON válido. " "No incluyas Markdown, comentarios ni texto adicional. " f"Esquema requerido: {schema_json}" ) system_prompt = ( f"{request.system_prompt}\n\n" "Tu respuesta DEBE ser JSON estricto y cumplir el esquema solicitado." if request.system_prompt else "Tu respuesta DEBE ser JSON estricto y cumplir el esquema solicitado." ) started_at = perf_counter() try: response = self._client_or_raise().chat.completions.create( model=model, temperature=0.0, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt}, ], ) text = response.choices[0].message.content or "" parsed_model = self._validate_json_output( text=text, output_model=output_model, model=model, prompt=prompt, system_prompt=system_prompt, ) metrics = { "duration_ms": round((perf_counter() - started_at) * 1000, 3), "input_chars": _text_size(system_prompt) + _text_size(prompt), "output_chars": _text_size(text), } return LLMStructuredResult( content=parsed_model.model_dump(by_alias=True, exclude_none=True), provider=self.provider_name(), model=model, output_kind=request.output_kind, metrics=metrics, ) except LLMProviderError: raise except Exception as exc: mapped = self._classify_exception(exc, model=model) raise mapped from exc def _validate_json_output( self, *, text: str, output_model: type[BaseModel], model: str, prompt: str, system_prompt: str, ) -> BaseModel: try: return output_model.model_validate_json(self._strip_json_fence(text)) except ValidationError as exc: retry_response = self._client_or_raise().chat.completions.create( model=model, temperature=0.0, messages=[ { "role": "system", "content": ( f"{system_prompt}\n\n" "Reintento obligatorio: responde únicamente JSON válido; no agregues texto fuera del objeto." ), }, {"role": "user", "content": prompt}, ], ) retry_text = retry_response.choices[0].message.content or "" try: return output_model.model_validate_json(self._strip_json_fence(retry_text)) except ValidationError as retry_exc: raise LLMProviderError( "Groq devolvió un JSON inválido para el esquema solicitado", kind=LLMErrorKind.UNSUPPORTED_CAPABILITY, provider=self.provider_name(), model=model, details={"validation_error": str(retry_exc)}, ) from exc @staticmethod def _strip_json_fence(value: str) -> str: text = str(value or "").strip() if text.startswith("```"): text = text.split("\n", 1)[1] if "\n" in text else text[3:] if text.endswith("```"): text = text[:-3] return text.strip()