"""LLM vía OpenAI Responses API (cloud) o Chat Completions (Ollama / proxies).""" from __future__ import annotations import json import logging import httpx from llm.protocols import ( ChatMessage, ConversationContext, LanguageModel, LlmFatalError, LlmQuotaExceeded, LlmReply, ) logger = logging.getLogger(__name__) DEFAULT_SYSTEM = ( "Sos un asistente telefónico en español rioplatense. " "Respondé de forma breve y clara, apto para voz (1-3 oraciones). " "No uses markdown ni listas largas." ) # Campos CRM que se envían al modelo (sin raw ni extras). _POLIZA_INPUT_KEYS = ( "id", "nombre", "apellido", "poliza", "producto", "plan_actual", "precio_actual", "plan_ofrecido", "precio_ofrecido", "plan_retencion", "monto_asegurado", "moneda", "creado", ) _QUOTA_CODES = frozenset( { "rate_limit_exceeded", "insufficient_quota", "quota_exceeded", "billing_hard_limit_reached", } ) # Structured Outputs (Responses: text.format.type=json_schema) al fin de llamada. RESULTADO_LLAMADA_NAME = "resultado_llamada" RESULTADO_LLAMADA_INSTRUCTIONS = ( "Extrae el resultado de la conversación de retención telefónica. " "Usá los datos de póliza conocidos cuando estén disponibles; no inventes id ni número de póliza. " "Clasificá motivo como uno de: Economicos, Desconocimiento, Fallecimiento, Otro. " 'resultado es "retencion" si el cliente aceptó cambio de plan o descuento; ' '"baja" si se confirmó la baja del seguro. ' "plan_anterior es el plan que tenía al inicio. " "plan_nuevo es el plan aceptado o null. " 'descuento_aceptado es true solo si aceptó la bonificación; descuento_aplicado es "20" en ese caso, si no null.' ) RESULTADO_LLAMADA_SCHEMA: dict = { "type": "object", "properties": { "id": {"type": "string"}, "poliza": {"type": "string"}, "motivo": { "type": "string", "enum": ["Economicos", "Desconocimiento", "Fallecimiento", "Otro"], }, "resultado": { "type": "string", "enum": ["baja", "retencion"], }, "plan_anterior": {"type": "string"}, "plan_nuevo": {"type": ["string", "null"]}, "descuento_aceptado": {"type": "boolean"}, "descuento_aplicado": {"type": ["string", "null"]}, }, "required": [ "id", "poliza", "motivo", "resultado", "plan_anterior", "plan_nuevo", "descuento_aceptado", "descuento_aplicado", ], "additionalProperties": False, } class OpenAiLanguageModel(LanguageModel): provider_name = "openai" # Circuito global del proceso: si OpenAI corta por cuota, no insistir. _quota_blocked: bool = False _quota_block_reason: str = "" def __init__( self, *, api_key: str, model: str = "gpt-4.1-mini", temperature: float = 0.7, max_tokens: int = 1000, top_p: float = 1.0, frequency_penalty: float = 0.0, presence_penalty: float = 0.0, assistant_id: str = "", system_prompt: str = "", base_url: str = "https://api.openai.com/v1", timeout: float = 60.0, provider_name: str = "openai", use_assistants: bool | None = None, use_responses: bool | None = None, ) -> None: self.provider_name = (provider_name or "openai").strip().lower() self.base_url = (base_url or "https://api.openai.com/v1").rstrip("/") key = (api_key or "").strip() if not key: if self.provider_name == "ollama": key = "ollama" else: raise ValueError( "OPENAI_API_KEY es obligatorio para LLM_PROVIDER=openai" ) self.api_key = key self.model = model self.temperature = temperature self.max_tokens = max_tokens self.top_p = top_p self.frequency_penalty = frequency_penalty self.presence_penalty = presence_penalty self.system_prompt = (system_prompt or DEFAULT_SYSTEM).strip() # Responses solo en API real de OpenAI (no Ollama / proxies locales) if use_responses is not None: self._use_responses = bool(use_responses) else: self._use_responses = ( self.provider_name == "openai" and "api.openai.com" in self.base_url ) # Assistants API deprecada: se ignora assistant_id (compat .env legacy) if assistant_id and assistant_id.strip(): logger.warning( "OPENAI_ASSISTANT_ID está definido pero se ignora: " "Assistants API deprecada; se usa Responses API. " "La personalidad va en BOT_SYSTEM_PROMPT / bot_system_prompt.txt." ) if use_assistants: logger.warning( "use_assistants=True ignorado: Assistants API reemplazada por Responses" ) self.assistant_id = "" # previous_response_id por call_id (historial stateful en OpenAI) self._response_ids: dict[str, str] = {} # call_ids a los que ya se inyectó el JSON de póliza en input self._poliza_seeded: set[str] = set() headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } if self._use_responses: logger.info( "LLM provider=%s modo=responses model=%s base=%s", self.provider_name, self.model, self.base_url, ) else: logger.info( "LLM provider=%s modo=chat.completions model=%s base=%s", self.provider_name, self.model, self.base_url, ) self._client = httpx.AsyncClient( base_url=self.base_url, headers=headers, timeout=timeout, ) # Ollama a veces rechaza penalties; se omiten fuera de OpenAI cloud self._send_penalties = "api.openai.com" in self.base_url and not self._use_responses # Si el proceso había cerrado el circuito por cuota OpenAI, Ollama puede usarse if self.provider_name == "ollama": OpenAiLanguageModel._quota_blocked = False OpenAiLanguageModel._quota_block_reason = "" async def warmup(self) -> None: """Ping corto a chat.completions para cargar el modelo en memoria (Ollama).""" if self.provider_name != "ollama": return try: response = await self._client.post( "/chat/completions", json={ "model": self.model, "messages": [{"role": "user", "content": "ok"}], "max_tokens": 1, "temperature": 0, }, ) response.raise_for_status() logger.info( "LLM warmup OK provider=ollama model=%s base=%s", self.model, self.base_url, ) except Exception as exc: logger.warning( "LLM warmup falló provider=ollama model=%s: %s", self.model, exc, ) def _build_instructions(self) -> str: """Solo comportamiento permanente (Responses: campo instructions).""" return self.system_prompt def _build_system(self, context: ConversationContext | None) -> str: """System message para Chat Completions (Ollama): prompt + datos dinámicos.""" parts = [self.system_prompt] context_block = self._context_input_text(context) if context_block: parts.append(context_block) return "\n".join(p for p in parts if p) @staticmethod def _poliza_payload(context: ConversationContext | None) -> dict | None: if not context or not context.poliza: return None data = context.poliza payload: dict = {} for key in _POLIZA_INPUT_KEYS: if key == "id": payload[key] = data.get("id") or context.document_id or "" else: payload[key] = data.get(key) return payload @classmethod def _context_input_text(cls, context: ConversationContext | None) -> str: """Texto de contexto dinámico (póliza / documento / llamante) para input o system.""" parts: list[str] = [] poliza = cls._poliza_payload(context) if poliza: parts.append( "Datos del cliente (fuente CRM; usá estos valores y no inventes otros):\n" + json.dumps(poliza, ensure_ascii=False, indent=2) ) elif context and context.document_id: parts.append( f"El documento/DNI asociado a esta llamada es {context.document_id}." ) if context and context.caller_number: parts.append(f"Número del llamante: {context.caller_number}.") return "\n".join(parts) @staticmethod def _latest_user_text(messages: list[ChatMessage]) -> str: for msg in reversed(messages): if msg.role == "user" and msg.content.strip(): return msg.content.strip() return "" @staticmethod def _history_as_input(messages: list[ChatMessage]) -> list[dict]: items: list[dict] = [] for msg in messages: if msg.role == "system": continue content = (msg.content or "").strip() if not content: continue if msg.role not in ("user", "assistant", "developer"): continue items.append({"role": msg.role, "content": content}) return items def _seed_context_input( self, call_id: str, context: ConversationContext | None, input_items: list[dict], ) -> list[dict]: """Inyecta el JSON de póliza una sola vez al inicio de la cadena Responses.""" if call_id in self._poliza_seeded: return input_items block = self._context_input_text(context) if not block: return input_items self._poliza_seeded.add(call_id) seed = { "role": "user", "content": ( "Contexto interno de la llamada (no es el cliente hablando):\n" f"{block}\n" "Cuando el cliente hable, usá estos datos." ), } return [seed, *input_items] @staticmethod def _extract_output_text(data: dict) -> str: direct = data.get("output_text") if isinstance(direct, str) and direct.strip(): return direct.strip() parts: list[str] = [] for item in data.get("output") or []: if not isinstance(item, dict): continue if item.get("type") != "message": continue for block in item.get("content") or []: if not isinstance(block, dict): continue if block.get("type") in ("output_text", "text"): text = block.get("text") or "" if text: parts.append(str(text)) return "\n".join(parts).strip() async def reply( self, messages: list[ChatMessage], *, context: ConversationContext | None = None, ) -> LlmReply: if OpenAiLanguageModel._quota_blocked: raise LlmQuotaExceeded( OpenAiLanguageModel._quota_block_reason or "OpenAI bloqueado por cuota agotada (circuito abierto)", ) try: if self._use_responses: return await self._reply_responses(messages, context=context) return await self._reply_chat(messages, context=context) except LlmFatalError: raise except httpx.HTTPStatusError as exc: self._raise_if_quota_http(exc) raise def _raise_if_quota_http(self, exc: httpx.HTTPStatusError) -> None: code = "" message = exc.response.text[:300] try: payload = exc.response.json() err = payload.get("error") or {} code = str(err.get("code") or "") message = str(err.get("message") or message) except Exception: pass lowered = f"{code} {message}".lower() if ( exc.response.status_code in (429, 402) or code in _QUOTA_CODES or "exceeded your current quota" in lowered or "insufficient_quota" in lowered ): OpenAiLanguageModel._quota_blocked = True OpenAiLanguageModel._quota_block_reason = message logger.error( "OpenAI CUOTA AGOTADA — circuito abierto, no más requests: %s", message, ) raise LlmQuotaExceeded(message, code=code or "rate_limit_exceeded") from exc async def _reply_responses( self, messages: list[ChatMessage], *, context: ConversationContext | None = None, ) -> LlmReply: """Responses API: instructions fijas + input dinámico + previous_response_id.""" call_id = (context.call_id if context else "") or "_default" previous_id = self._response_ids.get(call_id) if previous_id: user_text = self._latest_user_text(messages) if not user_text: return LlmReply(text="") input_payload: str | list[dict] = user_text else: input_items = self._history_as_input(messages) if not input_items: return LlmReply(text="") input_payload = self._seed_context_input(call_id, context, input_items) body: dict = { "model": self.model, "instructions": self._build_instructions(), "input": input_payload, "temperature": self.temperature, "max_output_tokens": self.max_tokens, "top_p": self.top_p, "store": True, } if previous_id: body["previous_response_id"] = previous_id data = await self._post_json("/responses", body) status = data.get("status") if status and status not in ("completed",): err = data.get("error") or {} raise LlmFatalError( f"OpenAI Responses status={status} " f"code={err.get('code')} msg={err.get('message')}", code=str(err.get("code") or status or "response_failed"), ) response_id = data.get("id") if isinstance(response_id, str) and response_id: self._response_ids[call_id] = response_id logger.debug( "OpenAI response chain call=%s response_id=%s prev=%s", call_id, response_id, previous_id or "-", ) text = self._extract_output_text(data) return LlmReply(text=text, raw=data) def _build_extract_input( self, messages: list[ChatMessage], *, context: ConversationContext | None = None, transcript: list[str] | None = None, ) -> str: parts: list[str] = [] poliza = self._poliza_payload(context) if poliza: parts.append( "Datos de póliza conocidos:\n" + json.dumps(poliza, ensure_ascii=False, indent=2) ) elif context and context.document_id: parts.append(f"Documento/DNI: {context.document_id}") if transcript: parts.append( "Transcripción de la llamada:\n" + "\n".join(transcript) ) else: lines: list[str] = [] for msg in messages: if msg.role == "system": continue content = (msg.content or "").strip() if not content: continue speaker = "cliente" if msg.role == "user" else "agente" lines.append(f"{speaker}: {content}") if lines: parts.append("Transcripción de la llamada:\n" + "\n".join(lines)) if not parts: return "No hay transcripción. Extraé el mejor resultado posible con campos vacíos o null según el esquema." parts.append("Extraé el resultado estructurado de esta llamada.") return "\n\n".join(parts) async def extract_call_result( self, messages: list[ChatMessage], *, context: ConversationContext | None = None, transcript: list[str] | None = None, ) -> dict | None: """Fin de llamada: Responses con text.format.type=json_schema.""" if OpenAiLanguageModel._quota_blocked: logger.warning( "extract_call_result omitido: circuito OpenAI abierto" ) return None if not messages and not transcript: return None input_text = self._build_extract_input( messages, context=context, transcript=transcript ) try: if self._use_responses: data = await self._extract_responses(input_text) text = self._extract_output_text(data) else: data = await self._extract_chat(input_text) text = ( data.get("choices", [{}])[0] .get("message", {}) .get("content", "") .strip() ) except LlmFatalError: raise except httpx.HTTPStatusError as exc: self._raise_if_quota_http(exc) logger.exception("extract_call_result HTTP error") return None except Exception: logger.exception("extract_call_result falló") return None if not text: return None try: parsed = json.loads(text) except json.JSONDecodeError: logger.warning( "extract_call_result: JSON inválido: %s", text[:300] ) return None if not isinstance(parsed, dict): return None return parsed async def _extract_responses(self, input_text: str) -> dict: """POST /responses con text.format json_schema (structured outputs).""" body: dict = { "model": self.model, "instructions": RESULTADO_LLAMADA_INSTRUCTIONS, "input": input_text, "temperature": 0, "max_output_tokens": min(500, self.max_tokens), "store": True, "text": { "format": { "type": "json_schema", "name": RESULTADO_LLAMADA_NAME, "strict": True, "schema": RESULTADO_LLAMADA_SCHEMA, } }, } data = await self._post_json("/responses", body) status = data.get("status") if status and status not in ("completed",): err = data.get("error") or {} raise LlmFatalError( f"OpenAI extract status={status} " f"code={err.get('code')} msg={err.get('message')}", code=str(err.get("code") or status or "response_failed"), ) return data async def _extract_chat(self, input_text: str) -> dict: """Chat Completions (Ollama): pide JSON; schema estricto si el proxy lo acepta.""" body: dict = { "model": self.model, "messages": [ {"role": "system", "content": RESULTADO_LLAMADA_INSTRUCTIONS}, {"role": "user", "content": input_text}, ], "temperature": 0, "max_tokens": min(500, self.max_tokens), "response_format": { "type": "json_schema", "json_schema": { "name": RESULTADO_LLAMADA_NAME, "strict": True, "schema": RESULTADO_LLAMADA_SCHEMA, }, }, } response = await self._client.post("/chat/completions", json=body) # Ollama / proxies viejos pueden rechazar json_schema → reintento json_object if response.status_code >= 400: body["response_format"] = {"type": "json_object"} response = await self._client.post("/chat/completions", json=body) if response.status_code >= 400: self._raise_if_quota_http( httpx.HTTPStatusError( f"HTTP {response.status_code}", request=response.request, response=response, ) ) response.raise_for_status() return response.json() async def _reply_chat( self, messages: list[ChatMessage], *, context: ConversationContext | None = None, ) -> LlmReply: """Chat Completions: Ollama y proxies compatibles OpenAI.""" payload_messages = [ {"role": "system", "content": self._build_system(context)}, ] for msg in messages: if msg.role == "system": continue payload_messages.append({"role": msg.role, "content": msg.content}) body: dict = { "model": self.model, "messages": payload_messages, "temperature": self.temperature, "max_tokens": self.max_tokens, "top_p": self.top_p, } if self._send_penalties: body["frequency_penalty"] = self.frequency_penalty body["presence_penalty"] = self.presence_penalty response = await self._client.post("/chat/completions", json=body) if response.status_code >= 400: self._raise_if_quota_http( httpx.HTTPStatusError( f"HTTP {response.status_code}", request=response.request, response=response, ) ) response.raise_for_status() data = response.json() text = ( data.get("choices", [{}])[0] .get("message", {}) .get("content", "") .strip() ) return LlmReply(text=text, raw=data) async def _post_json( self, path: str, body: dict, *, params: dict | None = None ) -> dict: response = await self._client.post(path, json=body, params=params) if response.status_code >= 400: logger.error( "OpenAI POST %s → %s %s", path, response.status_code, response.text[:500], ) try: self._raise_if_quota_http( httpx.HTTPStatusError( f"HTTP {response.status_code}", request=response.request, response=response, ) ) except LlmQuotaExceeded: raise response.raise_for_status() return response.json() def drop_thread(self, call_id: str) -> None: """Limpia la cadena previous_response_id de una llamada (nombre legacy).""" self._response_ids.pop(call_id, None) self._poliza_seeded.discard(call_id) def drop_response_chain(self, call_id: str) -> None: self._response_ids.pop(call_id, None) self._poliza_seeded.discard(call_id) async def close(self) -> None: self._response_ids.clear() self._poliza_seeded.clear() await self._client.aclose()