import os import sys import asyncio from dotenv import load_dotenv from loguru import logger # Importes de Pipecat from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import LLMRunFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.worker import PipelineParams, PipelineWorker from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( LLMContextAggregatorPair, ) from pipecat.services.google.llm import GoogleLLMService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.elevenlabs.tts import ElevenLabsTTSService from pipecat.services.llm_service import FunctionCallParams from pipecat.transports.livekit.transport import LiveKitParams, LiveKitTransport from pipecat.runner.livekit import configure, generate_token from pipecat.workers.runner import WorkerRunner # Configuración de Logger logger.remove() logger.add(sys.stderr, level="INFO") # Cargar variables de entorno del archivo .env si existe load_dotenv(override=True) # Establecer credenciales de LiveKit por defecto si no están definidas en el entorno os.environ.setdefault("LIVEKIT_URL", "wss://livekit.comunik.com.de") os.environ.setdefault("LIVEKIT_API_KEY", "APIdxklr3mif8bs") os.environ.setdefault("LIVEKIT_API_SECRET", "yhvyct65ogyaxotwqs3ww2xbp2xcmm1jdpz4csebdbmj") os.environ.setdefault("LIVEKIT_ROOM_NAME", "sala-demo-medico") LIVEKIT_URL = os.environ["LIVEKIT_URL"] LIVEKIT_API_KEY = os.environ["LIVEKIT_API_KEY"] LIVEKIT_API_SECRET = os.environ["LIVEKIT_API_SECRET"] LIVEKIT_ROOM_NAME = os.environ["LIVEKIT_ROOM_NAME"] # Instancias globales para comunicación y control desde herramientas transport = None worker = None dialogue_started = False # Validar otras claves necesarias if not os.getenv("DEEPGRAM_API_KEY"): logger.warning("ALERTA: La variable de entorno DEEPGRAM_API_KEY no está configurada. El servicio STT fallará.") if not os.getenv("GEMINI_API_KEY") and not os.getenv("GOOGLE_API_KEY"): logger.warning("ALERTA: La variable de entorno GEMINI_API_KEY o GOOGLE_API_KEY no está configurada. El LLM fallará.") # Base de datos médica simulada (MOCK) MOCK_MEDICO_DB = { "Medicina General": { "doctores": { "Dr. Juan Pérez": ["9:00 am", "10:30 am", "2:00 pm"], "Dra. María Gómez": ["11:00 am", "3:30 pm", "4:30 pm"] } }, "Cardiología": { "doctores": { "Dr. Carlos Mendoza": ["8:30 am", "11:30 am"], "Dra. Laura Torres": ["2:30 pm", "5:00 pm"] } }, "Pediatría": { "doctores": { "Dr. Luis Delgado": ["10:00 am", "12:00 pm", "3:00 pm"], "Dra. Elena Ruiz": ["9:30 am", "1:30 pm", "4:00 pm"] } } } # Prompt del sistema para el personaje de Sofía SYSTEM_PROMPT = """ Eres Sofía, una asistente médica virtual amable y muy eficiente para el agendamiento de citas médicas virtuales. Tu objetivo es ayudar al usuario a agendar una cita médica de forma rápida y por voz en un entorno de telemedicina. Sigue estas reglas estrictas para la conversación: 1. Saluda de forma muy corta en tu primer mensaje (ej. "Hola, soy Sofía, tu asistente médica virtual. ¿Qué especialidad necesitas hoy?"). 2. Las especialidades disponibles son únicamente: Medicina General, Cardiología y Pediatría. Si el usuario pide otra especialidad, infórmale amablemente cuáles están disponibles. 3. Para obtener los doctores y horarios disponibles de una especialidad médica, DEBES llamar a la función `consultar_disponibilidad`. 4. Al presentar las opciones de doctores y horarios, ofrece como MÁXIMO dos opciones en total para no abrumar al usuario por voz (ej. "Tengo al Dr. Pérez a las 9:00 o a la Dra. Gómez a las 11:00. ¿Cuál de estas opciones prefieres?"). 5. Tan pronto como el usuario te diga qué doctor y horario prefiere, DEBES llamar de inmediato a la herramienta `agendar_cita` con la especialidad, doctor y horario seleccionados. Esto es OBLIGATORIO para guardar la cita y actualizar la pantalla del usuario en tiempo real. 6. Luego de recibir la confirmación de la cita de la herramienta, infórmale al usuario que la cita ha sido confirmada con éxito (menciona doctor, hora y especialidad) y pregúntale de inmediato si le puedes ayudar con algo adicional (ej. "¿Te puedo ayudar con alguna otra consulta o agendamiento?"). 7. Si el usuario te responde que no necesita nada más, o que desea finalizar la llamada, despídete con calidez (ej. "¡Perfecto, que tengas un excelente día, hasta luego!") y en esa misma respuesta llama obligatoriamente a la herramienta `finalizar_llamada` con el parámetro motivo='despedida' para colgar la llamada de forma automática. No llames a `finalizar_llamada` hasta que el usuario confirme que no requiere más ayuda. Una vez invocada la herramienta finalizar_llamada, no debes generar ningún mensaje adicional de texto ni continuar hablando. 8. Al ser un demo de prueba interactivo, NO solicites datos personales reales (como correo electrónico, dirección, teléfono o cédula de identidad/DNI). 9. Mantén tus respuestas breves y conversacionales para que el usuario pueda interrumpirte de manera fluida si lo necesita. 10. Resolución de Contexto Implícito (¡Muy Importante!): Si el usuario responde con referencias breves o parciales a las opciones que acabas de presentar (ej. "sí, esa", "con la doctora", "con María", "la de las 11", "la primera", "la segunda"), DEBES usar el historial para deducir la opción completa. Completa los argumentos de la herramienta `agendar_cita` (especialidad, doctor, horario) de forma automática y llámala de inmediato. No le repitas la pregunta ni le pidas detalles adicionales si ya los tienes en el historial de la conversación. """ # --- IMPLEMENTACIÓN DE HERRAMIENTAS (Function Calling) --- async def consultar_disponibilidad(params: FunctionCallParams, especialidad: str = None): """ Consulta los doctores y horarios disponibles para una especialidad médica determinada. Args: especialidad: Nombre de la especialidad requerida. Debe ser 'Medicina General', 'Cardiología' o 'Pediatría'. """ if not especialidad: especialidad = params.arguments.get("especialidad") if not especialidad: logger.warning("[TOOL] consultar_disponibilidad llamada sin el argumento 'especialidad'") await params.result_callback({ "error": "Por favor especifica la especialidad médica ('Medicina General', 'Cardiología' o 'Pediatría') para poder consultar la disponibilidad." }) return logger.info(f"[TOOL] consultar_disponibilidad llamada para la especialidad: {especialidad}") # Normalización simple esp_lower = especialidad.lower() esp_key = None if "general" in esp_lower: esp_key = "Medicina General" elif "cardio" in esp_lower: esp_key = "Cardiología" elif "pediat" in esp_lower: esp_key = "Pediatría" if esp_key: result = MOCK_MEDICO_DB[esp_key] logger.info(f"[TOOL] Resultado encontrado para {esp_key}: {result}") await params.result_callback(result) else: logger.warning(f"[TOOL] Especialidad no encontrada para la consulta: {especialidad}") await params.result_callback({ "error": f"La especialidad '{especialidad}' no está disponible. Solo contamos con Medicina General, Cardiología y Pediatría." }) async def agendar_cita(params: FunctionCallParams, doctor: str = None, horario: str = None, especialidad: str = None): """ Agenda una cita médica virtual con un doctor específico en el horario y especialidad seleccionada. Args: doctor: El nombre del doctor seleccionado. horario: El horario elegido para la cita (ej. '09:00', '14:30'). especialidad: La especialidad médica correspondiente. """ if not doctor: doctor = params.arguments.get("doctor") if not horario: horario = params.arguments.get("horario") if not especialidad: especialidad = params.arguments.get("especialidad") if not doctor or not horario or not especialidad: logger.warning(f"[TOOL] agendar_cita llamada con parámetros incompletos: doctor={doctor}, horario={horario}, especialidad={especialidad}") await params.result_callback({ "error": "Faltan datos para agendar la cita. Se requiere especificar el nombre del doctor, el horario y la especialidad." }) return logger.info(f"[TOOL] agendar_cita llamada para: {doctor} a las {horario} ({especialidad})") # Simulamos el agendamiento guardándolo y confirmando response = { "success": True, "doctor": doctor, "horario": horario, "especialidad": especialidad, "mensaje": f"Cita agendada con éxito para las {horario} con el especialista {doctor}." } # Enviar datos en tiempo real al frontend a través del canal de datos de LiveKit import json global transport if transport and hasattr(transport, "_client") and transport._client._room: try: room = transport._client.room payload = { "type": "appointment_info", "doctor": doctor, "horario": horario, "especialidad": especialidad } await room.local_participant.publish_data( payload=json.dumps(payload).encode("utf-8"), reliable=True ) logger.info("[TOOL] Datos de cita enviados al frontend exitosamente.") except Exception as e: logger.error(f"[TOOL] Error al publicar datos en LiveKit: {e}") await params.result_callback(response) async def finalizar_llamada(params: FunctionCallParams, motivo: str = "despedida"): """ Finaliza la llamada de voz actual and cuelga la conexión WebRTC. Args: motivo: El motivo por el cual se finaliza la llamada (ej. 'despedida', 'cancelacion'). """ if not motivo: motivo = params.arguments.get("motivo", "despedida") logger.info(f"[TOOL] finalizar_llamada invocada (motivo: {motivo}). Sofía colgará la llamada en 3 segundos...") # Enviar señal de colgado en tiempo real al frontend a través de LiveKit import json global transport if transport and hasattr(transport, "_client") and transport._client._room: try: room = transport._client.room payload = { "type": "hangup_signal" } await room.local_participant.publish_data( payload=json.dumps(payload).encode("utf-8"), reliable=True ) logger.info("[TOOL] Señal de colgado (hangup_signal) enviada al frontend exitosamente.") except Exception as e: logger.error(f"[TOOL] Error al enviar señal de colgado: {e}") # Espera para que el audio del TTS termine de reproducirse en el cliente antes de cortar la sala await asyncio.sleep(6.0) await params.result_callback({"status": "desconectando"}) global worker if worker: try: logger.info("[TOOL] Deteniendo el worker...") await worker.stop() except Exception as e: logger.error(f"[TOOL] Error al detener el worker: {e}") # --- CONFIGURACIÓN E INICIALIZACIÓN DE TTS --- def setup_tts_service(): """ Configura y retorna el servicio de síntesis de voz (TTS) de acuerdo a las variables de entorno. Soporta Cartesia (recomendado por baja latencia) y ElevenLabs. """ provider = os.getenv("TTS_PROVIDER", "").lower() cartesia_key = os.getenv("CARTESIA_API_KEY") elevenlabs_key = os.getenv("ELEVENLABS_API_KEY") # Determinar proveedor basándose en keys si no está explícito if provider == "cartesia" or (not provider and cartesia_key): logger.info("Configurando TTS de Cartesia (Multilingual Sonic)...") voice_id = os.getenv("CARTESIA_VOICE_ID", "d4db5fb9-f44b-4bd1-85fa-192e0f0d75f9") # Por defecto, Spanish Lady (España) language = os.getenv("CARTESIA_LANGUAGE", "es") return CartesiaTTSService( api_key=cartesia_key or "", settings=CartesiaTTSService.Settings( voice=voice_id, model=os.getenv("CARTESIA_MODEL", "sonic-3.5"), language=language, ) ) elif provider == "elevenlabs" or (not provider and elevenlabs_key): logger.info("Configurando TTS de ElevenLabs (Multilingual v2)...") voice_id = os.getenv("ELEVENLABS_VOICE_ID", "EXAVITQu4vr4xnSDxMaL") # Bella return ElevenLabsTTSService( api_key=elevenlabs_key or "", settings=ElevenLabsTTSService.Settings( voice=voice_id, model="eleven_multilingual_v2", ) ) else: # Fallback local o simulado en caso de no tener llaves logger.warning("Claves de TTS no encontradas en el entorno. Usando Cartesia con clave simulada.") return CartesiaTTSService( api_key="", settings=CartesiaTTSService.Settings( voice="d4db5fb9-f44b-4bd1-85fa-192e0f0d75f9", model=os.getenv("CARTESIA_MODEL", "sonic-3.5"), language="es", ) ) # --- FUNCIÓN PRINCIPAL DEL AGENTE --- async def main(): import random import string # Configurar credenciales y URL de LiveKit directamente url = LIVEKIT_URL room_name = LIVEKIT_ROOM_NAME # Generar identidad única para el bot en esta ejecución para evitar conflictos por sesiones previas bot_id = "".join(random.choices(string.ascii_letters + string.digits, k=6)) bot_identity = f"sofia-agent-{bot_id}" try: # Generar token seguro para el bot token = generate_token( room_name=room_name, participant_name=bot_identity, api_key=LIVEKIT_API_KEY, api_secret=LIVEKIT_API_SECRET ) logger.info(f"Conexión LiveKit configurada exitosamente para la sala: {room_name} (bot identity: {bot_identity})") except Exception as e: logger.error(f"Error al generar token de LiveKit para el bot: {e}") return # Generar e imprimir el Token de pruebas para el Frontend (index.html) try: client_token = generate_token( room_name=room_name, participant_name="usuario-pruebas", api_key=LIVEKIT_API_KEY, api_secret=LIVEKIT_API_SECRET ) print("\n" + "="*80) print("TOKEN DE PRUEBA GENERADO PARA EL FRONTEND (index.html):") print(client_token) print("Copie el token de arriba y péguelo en index.html para simular la llamada.") print("="*80 + "\n") except Exception as token_err: logger.error(f"No se pudo generar el token temporal para el frontend: {token_err}") # 1. Configurar Transporte global transport transport = LiveKitTransport( url=url, token=token, room_name=room_name, params=LiveKitParams( audio_in_enabled=True, audio_out_enabled=True, vad=SileroVADAnalyzer(), # Detección nativa para interrupciones ), ) # 2. Configurar STT (Escuchar en español) stt = DeepgramSTTService( api_key=os.getenv("DEEPGRAM_API_KEY", ""), settings=DeepgramSTTService.Settings( model="nova-2-general", language="es", ) ) # 3. Configurar LLM (Pensar con Gemini 1.5 Flash) llm = GoogleLLMService( api_key=os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY", ""), settings=GoogleLLMService.Settings( model=os.getenv("GEMINI_MODEL", "gemini-3.5-flash"), temperature=0.3, # Mayor coherencia y apego al prompt ) ) # Registrar herramientas con el LLM llm.register_function("consultar_disponibilidad", consultar_disponibilidad) llm.register_function("agendar_cita", agendar_cita) llm.register_function("finalizar_llamada", finalizar_llamada) # 4. Configurar TTS (Hablar en español de baja latencia) tts = setup_tts_service() # 5. Configurar el contexto de mensajes de la conversación y las herramientas disponibles context = LLMContext( messages=[ { "role": "system", "content": SYSTEM_PROMPT, } ], tools=[consultar_disponibilidad, agendar_cita, finalizar_llamada] ) user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context) # 6. Definir Pipeline pipeline = Pipeline( [ transport.input(), # Audio entrante del usuario stt, # Transcripción del audio a texto user_aggregator, # Acumular contexto del usuario llm, # Procesar lógica y generar texto / tools tts, # Convertir respuesta del bot a audio transport.output(), # Emitir audio hacia el canal WebRTC assistant_aggregator, # Acumular contexto de la respuesta de Sofía ] ) # Inicializar Worker global worker worker = PipelineWorker(pipeline) global dialogue_started dialogue_started = False # Evento: Cuando el bot se suscribe al canal de audio del usuario @transport.event_handler("on_audio_track_subscribed") async def on_audio_track_subscribed(transport, participant_id): global dialogue_started if not dialogue_started: dialogue_started = True logger.info(f"Canal de audio suscrito para {participant_id}. Iniciando saludo de Sofía...") # Pausa breve de 1.5s para asegurar que el canal de salida en el cliente esté listo y no se corte el saludo await asyncio.sleep(1.5) await worker.queue_frame(LLMRunFrame()) # Evento: Cuando el primer participante se conecta a la sala de LiveKit @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): logger.info(f"Usuario de prueba conectado: {participant}. Esperando suscripción de audio...") # Evento: Cuando el participante se desconecta @transport.event_handler("on_participant_disconnected") async def on_participant_disconnected(transport, participant): logger.info(f"El usuario {participant} se desconectó de la sala.") global dialogue_started dialogue_started = False # Iniciar servidor web interno (para servir el Frontend y proveer Tokens sin copiar y pegar) from aiohttp import web app = web.Application() # Middleware para habilitar CORS async def cors_middleware(app, handler): async def middleware(request): if request.method == "OPTIONS": response = web.Response() else: response = await handler(request) response.headers["Access-Control-Allow-Origin"] = "*" response.headers["Access-Control-Allow-Headers"] = "Content-Type" response.headers["Access-Control-Allow-Methods"] = "POST, GET, OPTIONS" return response return middleware app.middlewares.append(cors_middleware) # Endpoint para generar token automáticamente async def handle_token(request): try: client_token = generate_token( room_name=room_name, participant_name="usuario-pruebas", api_key=LIVEKIT_API_KEY, api_secret=LIVEKIT_API_SECRET ) return web.json_response({ "token": client_token, "url": url, "room": room_name }) except Exception as e: return web.json_response({"error": str(e)}, status=500) app.router.add_get("/api/token", handle_token) # Servir archivos estáticos del directorio actual app.router.add_static("/", path=".", show_index=True) app_runner = web.AppRunner(app) await app_runner.setup() site = web.TCPSite(app_runner, "0.0.0.0", 8000) await site.start() logger.info("================================================================================") logger.info("SERVIDOR WEB INTERNO CORRIENDO EN: http://localhost:8000") logger.info(" - Panel de Pruebas: http://localhost:8000/index.html") logger.info(" - Demo de Widget Flotante (Popup Esquina): http://localhost:8000/widget_demo.html") logger.info(" - Demo de Modal Centrado (Popup Centro): http://localhost:8000/modal_demo.html") logger.info(" - Endpoint de Token: http://localhost:8000/api/token") logger.info("================================================================================") try: # Iniciar la ejecución de la tarea del bot mediante WorkerRunner runner = WorkerRunner() await runner.add_workers(worker) logger.info("Iniciando Agente de Voz de Sofía. Esperando conexiones de LiveKit...") await runner.run() finally: # Apagar servidor web de forma limpia al terminar await app_runner.cleanup() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("Agente detenido manualmente por el usuario.")