"""Orquestación conversacional: RTP → STT → LLM → TTS → RTP. Los proveedores STT/TTS/LLM se inyectan por factory; cambiar cloud↔local no requiere tocar este archivo. """ from __future__ import annotations import asyncio import audioop import logging from collections import deque from collections.abc import Awaitable, Callable from typing import Any from bot_actions import ( BotActionTrigger, find_trigger, load_bot_action_triggers, strip_phrase_from_speech, ) from call_metrics.session import utc_now from call_metrics.tracker import MetricsTracker, NullCallMetricsTracker from calls.models import CallState from config import Settings, get_settings from llm.factory import create_language_model from llm.protocols import ( ChatMessage, ConversationContext, LanguageModel, LlmFatalError, LlmQuotaExceeded, ) from media.ambiance import AmbienceLoop, get_ambience_loop from media.manager import MediaManager from media.rtp_session import SAMPLES_PER_PACKET, RtpSession from voice.factory import create_speech_to_text, create_text_to_speech from voice.protocols import AudioFormat, SpeechToText, TextToSpeech logger = logging.getLogger(__name__) NotifyFn = Callable[[CallState], Awaitable[None]] ActionFn = Callable[[CallState, BotActionTrigger], Awaitable[None]] TELEPHONY = AudioFormat(encoding="mulaw", sample_rate_hz=8000) # Tras TTS, descartar eco acústico del auricular antes del próximo turno. POST_SPEAK_COOLDOWN_MS = 600 # Tope de historial enviado al LLM (pares user/assistant). MAX_HISTORY_MESSAGES = 20 # Corte de seguridad: evita loops de eco / ruido infinito. MAX_LLM_TURNS = 20 # Si STT falla muchas veces seguidas, cortamos (sin gastar OpenAI). MAX_STT_ERRORS = 5 # Fallos LLM consecutivos (no-cuota) antes de abortar la conversación. MAX_LLM_ERRORS = 2 STT_ERROR_PROMPT = ( "Disculpá, no pude escucharte bien. ¿Podés repetir, por favor?" ) QUOTA_ERROR_PROMPT = ( "Disculpá, en este momento no puedo continuar con la atención automática. " "Por favor intentá más tarde." ) OPENING_PROMPT = ( "INICIO_LLAMADA: Generá el saludo inicial de esta llamada telefónica " "usando los datos de la póliza ya precargados. Sé breve (1-2 oraciones), " "saludá por el nombre del cliente si está disponible, mencioná que sos de " "Qualia seguros y preguntá en qué podés ayudar. No inventes datos que no " "estén en la póliza." ) class BotConversationSession: """Una conversación activa por llamada (turno a turno con VAD simple).""" def __init__( self, call: CallState, rtp: RtpSession, *, stt: SpeechToText, tts: TextToSpeech, llm: LanguageModel, settings: Settings, on_update: NotifyFn | None = None, on_action: ActionFn | None = None, action_triggers: list[BotActionTrigger] | None = None, metrics: MetricsTracker | None = None, poliza_task: asyncio.Task | None = None, ) -> None: self.call = call self.rtp = rtp self._poliza_task = poliza_task self.stt = stt self.tts = tts self.llm = llm self.settings = settings self.on_update = on_update self.on_action = on_action self.action_triggers = action_triggers or [] self.metrics = metrics or NullCallMetricsTracker() self._task: asyncio.Task | None = None self._stopped = asyncio.Event() self._history: list[ChatMessage] = [] self._speaking = False self._llm_turns = 0 self._stt_errors = 0 self._llm_errors = 0 self._llm_disabled = False self._warmup_task: asyncio.Task | None = None self._action_fired = False self._ambience: AmbienceLoop | None = None self._ambience_task: asyncio.Task | None = None def _metrics_session(self): return self.metrics.get(self.call.call_id) @property def running(self) -> bool: return self._task is not None and not self._task.done() def start(self) -> None: if self.running: return self._stopped.clear() self._task = asyncio.create_task( self._run(), name=f"bot-{self.call.call_id}" ) async def stop(self) -> None: self._stopped.set() await self._stop_ambience() warmup = self._warmup_task self._warmup_task = None if warmup and not warmup.done(): warmup.cancel() try: await warmup except asyncio.CancelledError: pass task = self._task self._task = None if task and not task.done(): task.cancel() try: await task except asyncio.CancelledError: pass async def finalize_resultado(self) -> dict | None: """Extrae resultado estructurado (json_schema) antes de liberar la cadena LLM.""" if self._llm_disabled: return None if not self._history and not self.call.transcript: return None try: result = await self.llm.extract_call_result( list(self._history), context=self._conversation_context(), transcript=list(self.call.transcript), ) except LlmQuotaExceeded as exc: logger.warning( "Bot call=%s: extract resultado omitido (cuota): %s", self.call.call_id, exc, ) return None except LlmFatalError as exc: logger.warning( "Bot call=%s: extract resultado falló: %s", self.call.call_id, exc, ) return None except Exception: logger.exception( "Bot call=%s: extract resultado error inesperado", self.call.call_id, ) return None if result: self.call.call_result = result logger.info( "Bot call=%s: resultado=%s", self.call.call_id, result, ) await self._notify() return result def _call_alive(self) -> bool: return ( not self._stopped.is_set() and self.call.status not in ("ended", "failed") and not self.rtp.is_closed ) async def _notify(self) -> None: if self.on_update: await self.on_update(self.call) def _set_agent_state(self, state: str) -> None: self.call.agent_state = state if self.call.status in ("ringing", "answered"): self.call.status = "talking" async def _run(self) -> None: call_id = self.call.call_id logger.info( "Bot conversación iniciada call=%s stt=%s tts=%s llm=%s doc=%s", call_id, self.stt.provider_name, self.tts.provider_name, self.llm.provider_name, self.call.document_id, ) try: # Softphone UDP a menudo no manda RTP en silencio; WebRTC sí. # Si ya hay destino (UNICASTRTP_LOCAL_*), seguimos y hablamos igual. rtp_ok = await self._wait_for_rtp(timeout=8.0) if not self._call_alive(): return if not rtp_ok: advertise = self.settings.external_media_advertise_host logger.error( "Bot call=%s: sin destino RTP hacia Asterisk. " "EXTERNAL_MEDIA_ADVERTISE_HOST=%s puerto=%s — " "debe ser la IP de ESTE servidor alcanzable desde Asterisk " "(UDP abierto). Revisá también UNICASTRTP_LOCAL_* del canal " "externalMedia.", call_id, advertise, self.rtp.local_port, ) self.call.agent_state = "error" await self._notify() return await self._await_poliza_if_pending() self._start_ambience() await self._speak_opening() if self._llm_disabled or not self._call_alive(): return while self._call_alive(): if self._llm_disabled: break if self._llm_turns >= MAX_LLM_TURNS: logger.warning( "Bot call=%s: tope de %d turnos LLM; deteniendo", call_id, MAX_LLM_TURNS, ) break self._set_agent_state("listening") await self._notify() utterance = await self._capture_utterance() if not self._call_alive() or self._llm_disabled: break if not utterance: continue self._set_agent_state("thinking") await self._notify() user_text = await self._recognize_user(utterance) if not self._call_alive() or self._llm_disabled: break if user_text is None: # Error de STT: no llamar a OpenAI. self._stt_errors += 1 if self._stt_errors >= MAX_STT_ERRORS: logger.error( "Bot call=%s: %d errores STT seguidos; abortando", call_id, self._stt_errors, ) self.call.agent_state = "error" await self._notify() break await self._speak(STT_ERROR_PROMPT) await self._post_speak_cooldown() continue if not user_text: logger.debug("STT vacío call=%s", call_id) continue # Turno de conversación válido (hubo texto) user_audio_ms = int(len(utterance) / 8.0) msess = self._metrics_session() if msess is not None: msess.record_user_turn(user_audio_ms) self._stt_errors = 0 logger.info("Bot STT call=%s: %r", call_id, user_text) self.call.transcript.append(f"user: {user_text}") self._history.append(ChatMessage(role="user", content=user_text)) await self._notify() bot_text = await self._reply_llm(user_text) if not self._call_alive() or self._llm_disabled: break if not bot_text: continue self.call.transcript.append(f"bot: {bot_text}") self._history.append( ChatMessage(role="assistant", content=bot_text) ) await self._speak_and_maybe_act(bot_text) if self._action_fired or not self._call_alive(): break await self._post_speak_cooldown() except asyncio.CancelledError: raise except Exception: logger.exception("Bot conversación error call=%s", call_id) self.call.agent_state = "error" await self._notify() finally: await self._stop_ambience() logger.info( "Bot conversación finalizada call=%s turns=%d", call_id, self._llm_turns, ) async def _speak_and_maybe_act(self, bot_text: str) -> None: """TTS del turno; si hay frase trigger, agenda la acción tras hablar.""" trigger = find_trigger(bot_text, self.action_triggers) speech = bot_text if trigger and not trigger.speak_phrase: speech = strip_phrase_from_speech(bot_text, trigger.phrase) or bot_text if speech.strip(): await self._speak(speech) if not trigger or self._action_fired or not self.on_action: return self._action_fired = True logger.info( "Bot call=%s: trigger %r → action=%s target=%s", self.call.call_id, trigger.phrase, trigger.action, trigger.target or "-", ) # No await: transfer hace stop_for_call y no debe cancelar esta misma task. asyncio.create_task( self._dispatch_action(trigger), name=f"bot-action-{self.call.call_id[:8]}", ) async def _dispatch_action(self, trigger: BotActionTrigger) -> None: if not self.on_action: return try: await self.on_action(self.call, trigger) except Exception: logger.exception( "Bot call=%s: falló acción %s", self.call.call_id, trigger.action, ) self._action_fired = False def _conversation_context(self) -> ConversationContext: return ConversationContext( call_id=self.call.call_id, document_id=self.call.document_id, caller_number=self.call.number, poliza=self.call.poliza_data, ) def _build_immediate_greeting(self) -> str: """Saludo TTS sin LLM: usa nombre CRM si hay póliza, si no BOT_GREETING.""" data = self.call.poliza_data or {} ''' nombre = " ".join( p for p in (str(data.get("nombre") or ""), str(data.get("apellido") or "")) if p ).strip() or str(data.get("nombre_completo") or "").strip() ''' nombre = data.get("nombre") or "" if nombre: return ( f"Hola {nombre}, soy Mariana de Qualia seguros. " "¿En qué puedo ayudarte?" ) return (self.settings.bot_greeting or "").strip() def _start_llm_warmup(self) -> None: if self._llm_disabled or self._warmup_task is not None: return if (self.settings.llm_provider or "").strip().lower() != "ollama": return async def _run() -> None: try: await self.llm.warmup() except Exception: logger.exception( "Bot call=%s: warmup LLM falló", self.call.call_id, ) self._warmup_task = asyncio.create_task( _run(), name=f"llm-warmup-{self.call.call_id[:8]}", ) logger.info( "Bot call=%s: warmup Ollama en paralelo al saludo", self.call.call_id, ) async def _await_warmup(self) -> None: task = self._warmup_task if task is None or task.done(): return try: await task except Exception: pass async def _await_poliza_if_pending(self) -> None: """Espera breve a CRM antes del saludo (carga en paralelo al RTP wait).""" task = self._poliza_task if task is None or task.done(): return try: await asyncio.wait_for(asyncio.shield(task), timeout=3.0) except asyncio.TimeoutError: logger.info( "Bot call=%s: CRM aún pendiente; saludo con fallback", self.call.call_id, ) except Exception: pass async def _speak_opening(self) -> None: """Saludo inicial. Por defecto (BOT_OPENING_MODE=immediate): TTS ya con nombre CRM / BOT_GREETING, y warmup de Ollama en paralelo. Con mode=llm: espera el saludo generado por el modelo (comportamiento viejo). """ mode = (self.settings.bot_opening_mode or "immediate").strip().lower() if mode == "llm" and self.call.poliza_data and not self._llm_disabled: logger.info( "Bot call=%s: saludo inicial vía LLM con póliza %s", self.call.call_id, (self.call.poliza_data or {}).get("poliza"), ) self._set_agent_state("thinking") await self._notify() bot_text = await self._reply_llm( OPENING_PROMPT, record_user=False, ) if self._llm_disabled or not self._call_alive(): return if bot_text: self.call.transcript.append(f"bot: {bot_text}") self._history.append( ChatMessage(role="assistant", content=bot_text) ) await self._speak(bot_text) await self._post_speak_cooldown() return logger.warning( "Bot call=%s: LLM no devolvió saludo; uso saludo inmediato", self.call.call_id, ) if self._llm_disabled and mode == "llm": return self._start_llm_warmup() greeting = self._build_immediate_greeting() if greeting: logger.info( "Bot call=%s: saludo inmediato %r", self.call.call_id, greeting[:120], ) self.call.transcript.append(f"bot: {greeting}") self._history.append(ChatMessage(role="assistant", content=greeting)) await self._speak(greeting) await self._post_speak_cooldown() async def _disable_llm(self, reason: str, *, speak: bool = True) -> None: if self._llm_disabled: return self._llm_disabled = True self.call.agent_state = "error" logger.error( "Bot call=%s: LLM deshabilitado — %s", self.call.call_id, reason, ) await self._notify() if speak and self._call_alive(): await self._speak(QUOTA_ERROR_PROMPT) await self._post_speak_cooldown() async def _recognize_user(self, utterance: bytes) -> str | None: """STT del turno. ``None`` = error del proveedor; ``\"\"`` = sin texto.""" t0 = asyncio.get_running_loop().time() try: result = await self.stt.recognize( utterance, audio_format=TELEPHONY ) except Exception: logger.exception( "Bot STT falló call=%s — se omite LLM/OpenAI este turno", self.call.call_id, ) msess = self._metrics_session() if msess is not None: latency_ms = (asyncio.get_running_loop().time() - t0) * 1000.0 msess.record_stt( latency_ms, audio_seconds=len(utterance) / 8000.0, ) return None latency_ms = (asyncio.get_running_loop().time() - t0) * 1000.0 msess = self._metrics_session() if msess is not None: msess.record_stt( latency_ms, audio_seconds=len(utterance) / 8000.0, ) return (result.text or "").strip() async def _reply_llm( self, user_text: str, *, record_user: bool = True ) -> str: """Solo se invoca con transcripción válida.""" if self._llm_disabled: return "" if record_user: pass history_for_llm = list(self._history[-MAX_HISTORY_MESSAGES:]) if not any( m.role == "user" and m.content == user_text for m in history_for_llm ): history_for_llm = history_for_llm + [ ChatMessage(role="user", content=user_text) ] self._llm_turns += 1 logger.info( "Bot LLM request call=%s turn=%d history=%d poliza=%s", self.call.call_id, self._llm_turns, len(history_for_llm), bool(self.call.poliza_data), ) # Si el warmup aún corre, esperar antes del primer turno real await self._await_warmup() t0 = asyncio.get_running_loop().time() try: reply = await self.llm.reply( history_for_llm, context=self._conversation_context(), ) except LlmQuotaExceeded as exc: await self._disable_llm(f"cuota OpenAI: {exc}") return "" except LlmFatalError as exc: await self._disable_llm(f"error fatal LLM: {exc}") return "" except Exception: self._llm_errors += 1 logger.exception("Bot LLM falló call=%s", self.call.call_id) if self._llm_errors >= MAX_LLM_ERRORS: await self._disable_llm( f"{self._llm_errors} fallos LLM consecutivos" ) return "" latency_ms = (asyncio.get_running_loop().time() - t0) * 1000.0 msess = self._metrics_session() if msess is not None: msess.record_llm( latency_ms, prompt_tokens=getattr(reply, "prompt_tokens", 0) or 0, completion_tokens=getattr(reply, "completion_tokens", 0) or 0, ) text = (reply.text or "").strip() if text: self._llm_errors = 0 return text async def _wait_for_rtp(self, timeout: float) -> bool: # Destino ya fijado vía ARI: no exigir paquetes entrantes (softphone UDP). if self.rtp.has_remote: frame = await self.rtp.recv_ulaw_frame(timeout=min(1.0, timeout)) if frame: logger.info( "Bot call=%s: RTP recibido (listo para audio)", self.call.call_id, ) else: logger.info( "Bot call=%s: sin RTP entrante aún (habitual en softphone " "con silence suppression); se usa UNICASTRTP_LOCAL_*", self.call.call_id, ) return True deadline = asyncio.get_running_loop().time() + timeout while self._call_alive(): remaining = deadline - asyncio.get_running_loop().time() if remaining <= 0: logger.warning( "Bot call=%s: timeout esperando RTP de Asterisk", self.call.call_id, ) return False frame = await self.rtp.recv_ulaw_frame(timeout=min(0.5, remaining)) if frame: logger.info( "Bot call=%s: RTP recibido (listo para audio)", self.call.call_id, ) return True return False async def _post_speak_cooldown(self) -> None: """Descarta frames entrantes para no transcribir eco del propio TTS.""" deadline = ( asyncio.get_running_loop().time() + POST_SPEAK_COOLDOWN_MS / 1000.0 ) while self._call_alive(): remaining = deadline - asyncio.get_running_loop().time() if remaining <= 0: break await self.rtp.recv_ulaw_frame(timeout=min(0.1, remaining)) async def _capture_utterance(self) -> bytes: """VAD por energía: silencia → fin de turno. Mantiene un pre-roll de frames previos al umbral para no cortar el ataque de la primera palabra (común en STT telefónico). """ silence_frames = max( 1, int(self.settings.bot_silence_ms / 20) ) min_speech_frames = max( 1, int(self.settings.bot_min_speech_ms / 20) ) max_frames = max( min_speech_frames, int(self.settings.bot_max_utterance_ms / 20), ) energy_threshold = self.settings.bot_vad_energy preroll_frames = max(0, int(self.settings.bot_preroll_ms / 20)) preroll: deque[bytes] = deque( maxlen=preroll_frames if preroll_frames > 0 else 1 ) buf = bytearray() speech_seen = 0 silence_seen = 0 started = False frames_seen = 0 no_rtp_warned = False listen_started = asyncio.get_running_loop().time() # Frames de silencio previos al speech (espera del usuario) pre_speech_silence_frames = 0 while self._call_alive(): frame = await self.rtp.recv_ulaw_frame(timeout=0.5) if frame is None: if ( not no_rtp_warned and not started and frames_seen == 0 and asyncio.get_running_loop().time() - listen_started >= 5.0 ): no_rtp_warned = True logger.warning( "Bot call=%s: 5s escuchando sin RTP del puente — " "Asterisk no está recibiendo audio del caller " "(NAT típico en datos móviles: falta " "external_media_address/local_net en el transport PJSIP, " "rtp_symmetric=yes, o UDP RTP abierto en firewall)", self.call.call_id, ) if started and silence_seen >= silence_frames: break continue frames_seen += 1 # Mientras el bot habla, descartamos audio entrante (sin barge-in aún) if self._speaking: preroll.clear() continue pcm = RtpSession.ulaw_to_pcm(frame) rms = audioop.rms(pcm, 2) is_speech = rms >= energy_threshold if not started: if preroll_frames > 0: preroll.append(frame) if is_speech: started = True speech_seen = 1 silence_seen = 0 if preroll: buf.extend(b"".join(preroll)) else: buf.extend(frame) preroll.clear() else: pre_speech_silence_frames += 1 continue if is_speech: speech_seen += 1 silence_seen = 0 buf.extend(frame) else: silence_seen += 1 buf.extend(frame) if silence_seen >= silence_frames and speech_seen >= min_speech_frames: break if speech_seen + silence_seen >= max_frames: break msess = self._metrics_session() if msess is not None and pre_speech_silence_frames > 0: # 20 ms por frame RTP msess.add_silence_seconds(pre_speech_silence_frames * 0.02) return bytes(buf) if speech_seen >= min_speech_frames else b"" async def _speak(self, text: str) -> None: if not self._call_alive(): return self._set_agent_state("speaking") await self._notify() # Ambience sigue durante synthesize; se pausa solo al enviar TTS mezclado t0 = asyncio.get_running_loop().time() audio = await self.tts.synthesize(text, audio_format=TELEPHONY) synth_ms = (asyncio.get_running_loop().time() - t0) * 1000.0 msess = self._metrics_session() if msess is not None: msess.record_tts( synth_ms, characters=len(text or ""), audio_ready_at=utc_now(), ) if not audio or not self._call_alive(): return play_started = asyncio.get_running_loop().time() self._speaking = True try: await self._play_ulaw(audio) finally: self._speaking = False play_ms = int((asyncio.get_running_loop().time() - play_started) * 1000) if msess is not None: msess.record_bot_turn(play_ms) def _start_ambience(self) -> None: if not self.settings.bot_ambience_enabled: return if self._ambience_task is not None: return path = self.settings.resolve_prompt_path(self.settings.bot_ambience_file) if not path: logger.warning( "Bot call=%s: ambience no encontrado (%s)", self.call.call_id, self.settings.bot_ambience_file, ) return try: self._ambience = get_ambience_loop( path, gain=self.settings.bot_ambience_gain ) except Exception: logger.exception( "Bot call=%s: no se pudo cargar ambience %s", self.call.call_id, path, ) return self._ambience_task = asyncio.create_task( self._ambience_loop(), name=f"ambience-{self.call.call_id[:8]}", ) logger.info( "Bot call=%s: ambience ON file=%s gain=%.2f (%.1fs loop)", self.call.call_id, path.name, self.settings.bot_ambience_gain, self._ambience.duration_s, ) async def _stop_ambience(self) -> None: task = self._ambience_task self._ambience_task = None self._ambience = None if task and not task.done(): task.cancel() try: await task except asyncio.CancelledError: pass async def _ambience_loop(self) -> None: """Envía ambiente en loop mientras no hay TTS (el TTS mezcla encima).""" while self._call_alive(): if self._speaking or self._ambience is None: await asyncio.sleep(0.02) continue try: self.rtp.send_ulaw(self._ambience.next_frame()) except Exception: logger.exception( "Bot call=%s: error enviando ambience", self.call.call_id, ) break await asyncio.sleep(0.02) async def _play_ulaw(self, ulaw: bytes) -> None: for offset in range(0, len(ulaw), SAMPLES_PER_PACKET): if not self._call_alive(): break chunk = ulaw[offset : offset + SAMPLES_PER_PACKET] if self._ambience is not None: chunk = self._ambience.mix_with_tts(chunk) self.rtp.send_ulaw(chunk) await asyncio.sleep(0.02) class BotSessionManager: """Administra sesiones de bot por call_id.""" def __init__( self, media_manager: MediaManager, settings: Settings | None = None, on_update: NotifyFn | None = None, on_action: ActionFn | None = None, metrics: MetricsTracker | None = None, ) -> None: self.media_manager = media_manager self.settings = settings or get_settings() self.on_update = on_update self.on_action = on_action self.metrics = metrics or NullCallMetricsTracker() self._sessions: dict[str, BotConversationSession] = {} self._stt: SpeechToText | None = None self._tts: TextToSpeech | None = None self._llm: LanguageModel | None = None self._action_triggers = load_bot_action_triggers(self.settings) def set_on_update(self, fn: NotifyFn | None) -> None: self.on_update = fn def set_on_action(self, fn: ActionFn | None) -> None: self.on_action = fn def set_metrics(self, metrics: MetricsTracker) -> None: self.metrics = metrics def _ensure_providers(self) -> tuple[SpeechToText, TextToSpeech, LanguageModel]: if self._stt is None: self._stt = create_speech_to_text(self.settings) if self._tts is None: self._tts = create_text_to_speech(self.settings) if self._llm is None: self._llm = create_language_model(self.settings) return self._stt, self._tts, self._llm async def start_for_call( self, call: CallState, *, poliza_task: asyncio.Task | None = None, ) -> None: if not self.settings.bot_enabled: logger.info("Bot deshabilitado (BOT_ENABLED=false) call=%s", call.call_id) return if call.call_id in self._sessions and self._sessions[call.call_id].running: return rtp = self.media_manager.get_rtp_session(call.call_id) if rtp is None: raise RuntimeError( f"Sin sesión RTP para bot call={call.call_id}; " "attach_bot_external_media debe ejecutarse antes" ) stt, tts, llm = self._ensure_providers() self.metrics.set_providers( call.call_id, stt=stt.provider_name, tts=tts.provider_name, llm=llm.provider_name, llm_model=getattr(llm, "model", None), ) session = BotConversationSession( call, rtp, stt=stt, tts=tts, llm=llm, settings=self.settings, on_update=self.on_update, on_action=self.on_action, action_triggers=self._action_triggers, metrics=self.metrics, poliza_task=poliza_task, ) self._sessions[call.call_id] = session session.start() async def warmup_llm(self) -> None: """Precarga el LLM (Ollama) al arrancar el backend.""" if not self.settings.bot_enabled: return if (self.settings.llm_provider or "").strip().lower() != "ollama": return _, _, llm = self._ensure_providers() logger.info("Warmup LLM al startup provider=ollama") await llm.warmup() async def stop_for_call(self, call_id: str) -> None: session = self._sessions.pop(call_id, None) if session: # Extraer resultado estructurado ANTES de cortar la cadena Responses. try: await session.finalize_resultado() except Exception: logger.exception( "Bot call=%s: fallo al finalizar resultado", call_id ) await session.stop() # Liberar cadena previous_response_id (Responses) de esta llamada if self._llm is not None and hasattr(self._llm, "drop_thread"): try: self._llm.drop_thread(call_id) # type: ignore[attr-defined] except Exception: pass def is_running(self, call_id: str) -> bool: session = self._sessions.get(call_id) return session is not None and session.running async def close(self) -> None: for call_id in list(self._sessions): await self.stop_for_call(call_id) for provider in (self._stt, self._tts, self._llm): if provider is not None: try: await provider.close() except Exception as exc: logger.debug("Cierre provider: %s", exc) self._stt = self._tts = self._llm = None def status(self) -> dict[str, Any]: return { "enabled": self.settings.bot_enabled, "stt_provider": self.settings.stt_provider, "tts_provider": self.settings.tts_provider, "llm_provider": self.settings.llm_provider, "active_sessions": [ cid for cid, s in self._sessions.items() if s.running ], }