# # python_robotica_interaccion_voz.py # # Interacción Humano Robot por voz # # REQUISITOS: # Descargar el modelo VOSK (STT) para español de: https://alphacephei.com/vosk/models # Descargar el modelo Claude (TTS) para español de: https://huggingface.co/rhasspy/piper-voices/tree/main/es/es_MX/claude/high # Descargar el modelo "qwen2.5:3b" para Ollama de: https://ollama.com/search # # Estados que se manejan durante la conversación: # "WAITING_FOR_WAKEWORD" -> Escuchar constantemente para detectar la palabra clave("pony") # "WAITING_FOR_COMMAND" -> Esperando un comando ó entrar en modo plática activa # "CONVERSATION" -> Modo de conversación contínua usando Ollama y un LLM # # Rogelio Ferreira Escutia - 11/septiembre/2026 # # Bibliotecas utilizadas import os import json import wave import time import sys # Biblitecas utilizadas y que se requieren instalar from vosk import Model, KaldiRecognizer import pyaudio import ollama import numpy as np from piper.voice import PiperVoice os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide" # Ocultar el mensaje de bienvenida de pygame import pygame from ollama import Client, ResponseError # Lista de constantes utilizadas VOSK_MODEL_PATH = "models_stt/vosk-model-es-0.42" # Ruta del modelo "Vosk" (STT) para procesamiento de Español WAKEWORD = "pony" # Palabra clave VOICE_MODEL_PATH = "models_tts/es_MX-claude-high.onnx" # Ruta del modelo "claude" (TTS) para procesamiento de Español VOICE_OUTPUT_WAV_FILE = "voice_output.wav" # archivo WAV de salida # LLMs a utilizar con Ollama (debe estar instalado Ollama y el modelo que se requiera) LLM_MODEL_EMBEDDINGS = "bge-m3" # Modelo local corriendo en Ollama para detección de similitudes LLM_MODEL_CONVERSATION = "qwen2.5:3b" # Modelo local corriendo en Ollama para conversación activa # Inicializar el sistema STT (Speach To Text) usando "Vosk" def init_stt(): print("\n\t-- Iniciando Vosk --\n") if not os.path.exists(VOSK_MODEL_PATH): print(f"Error: Directorio '{VOSK_MODEL_PATH}' no encontrado!") print("Descargar el modelo ´VOSK´ para Español antes de ejecutar el programa.") sys.exit(1) model = Model(VOSK_MODEL_PATH) recognizer = KaldiRecognizer(model, 16000) return recognizer # Reconocimiento de Voz (TTS - Text To Speech) def listen_phrase(recognizer, stream) -> str: # Vaciar el búfer del reconocedor de Vosk para que olvide fragmentos de palabras pasadas recognizer.Reset() # Leer y borrar todo el audio que se acumuló en PyAudio mientras Piper hablaba while stream.get_read_available() > 0: stream.read(stream.get_read_available(), exception_on_overflow=False) # Escucha hasta que es capturada una frase while True: data = stream.read(4000, exception_on_overflow=False) if len(data) == 0: continue # Escucha hasta que finaliza una frase if recognizer.AcceptWaveform(data): result = json.loads(recognizer.Result()) text = result.get("text", "").strip().lower() if text: return text # Calcular en porcentaje la similud de 2 frases def calculate_similarity(frase_1: str, frase_2: str): # Convertir las frases a minúsculas frase_1 = frase_1.lower() frase_2 = frase_2.lower() # Generando los Words Embeddings word_embedding_1 = ollama.embed(model=LLM_MODEL_EMBEDDINGS, input=frase_1) word_embedding_2 = ollama.embed(model=LLM_MODEL_EMBEDDINGS, input=frase_2) # Creando los vectores vector_1 = np.array(word_embedding_1["embeddings"][0]) vector_2 = np.array(word_embedding_2["embeddings"][0]) # Cálculo de la Similitud Coseno: (A · B) / (||A|| * ||B||) producto_punto = np.dot(vector_1, vector_2) normal_vector_1 = np.linalg.norm(vector_1) normal_vector_2 = np.linalg.norm(vector_2) if normal_vector_1 == 0 or normal_vector_2 == 0: return 0.0 similitud = producto_punto / (normal_vector_1 * normal_vector_2) # Convertir el resultado en una escala de porcentaje (0.0% a 100.0%) porcentaje_similitud = max(0.0, float(similitud)) * 100 return round(porcentaje_similitud, 2) # Convertir texto a voz y reproducirlo def text_to_voice(text, VOICE_MODEL_PATH, VOICE_OUTPUT_WAV_FILE): # Cargar la voz de Piper voice = PiperVoice.load(VOICE_MODEL_PATH) # Sintetizar el texto y guardarlo en un archivo WAV with wave.open(VOICE_OUTPUT_WAV_FILE, "wb") as wav_file: voice.synthesize_wav(text, wav_file) # Inicializar pygame mixer pygame.mixer.init() # Cargar el archivo WAV pygame.mixer.music.load(VOICE_OUTPUT_WAV_FILE) # Reproducir el archivo print(f"\t[Robot -> Voz]: ´{text}´") pygame.mixer.music.play() # Esperar a que termine la reproducción while pygame.mixer.music.get_busy(): time.sleep(1) # Verificar si existe esa orden y ejecutarla def execute_robot_action(command: str): print(f"\t[Verificar]: Comando a verificar -> '{command}'") # Lista de comandos a ejecutar posibles set_of_commands = [ "cierra la puerta", "dime la hora", "dime mis actividades para hoy", "cuál es la temperatura actual", "cómo te llamas" ] command_found = False for phrase in set_of_commands: similarity = calculate_similarity(command, phrase) print(f"\t\tSimilaridad: {similarity}% - Frase usuario: {command} - Comando posible: {phrase}", ) if similarity>90: command_found = True text = "Entendido!" text_to_voice(text, VOICE_MODEL_PATH, VOICE_OUTPUT_WAV_FILE) # --- Aquí va el código que se requiera para ejecución del comando --- conversation_status = "WAITING_FOR_WAKEWORD" break if not command_found: text = "No puedo ejecutar ese comando" text_to_voice(text, VOICE_MODEL_PATH, VOICE_OUTPUT_WAV_FILE) return # Recibe una pregunta y la contesta por medio de un modelo def llm_conversation(user_query: str, model_name: str) -> str: try: respuesta = cliente_ollama.chat( model=model_name, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_query}, ], options={ "temperature": temperatura, "num_predict": max_tokens, "keep_alive": '-1', }, ) contenido = respuesta["message"]["content"].strip() return contenido if contenido else "No tengo una respuesta para eso." except ResponseError as e: # El modelo no existe localmente, o Ollama devolvió un error de API print(f"No pude usar el modelo {model_name} en este momento") return except Exception: # Cualquier otro fallo (timeout, conexión caída, etc.) # Se captura de forma amplia a propósito: en un pipeline de voz # en vivo es preferible dar una respuesta hablada de fallback # que dejar que la excepción se propague y corte el flujo. print("Tuve un problema para procesar eso, ¿lo intentamos de nuevo?") return # Función principal def main(): print("\n *** Interacción Humano Robot - Ponytron ***") recognizer = init_stt() # Inicializa PyAudio para escuchar de manera contínua audio = pyaudio.PyAudio() stream = audio.open( format=pyaudio.paInt16, channels=1, rate=16000, input=True, frames_per_buffer=8000 ) stream.start_stream() # Definir el ESTADO INICIAL de la conversación conversation_status = "WAITING_FOR_WAKEWORD" # Historial del Chat chat_history = [] print("\n\t-> Sistema de Voz Iniciado: ESTADOS -> 1) WAITING_FOR_WAKEWORD 2) WAITING_FOR_COMMAND 3)CONVERSATION\n") try: while True: # ------------------------------------------------------------- # ESTADO 1: "WAITING_FOR_WAKEWORD" -> Escuchar constantemente para detectar la palabra clave("pony") # ------------------------------------------------------------- if conversation_status == "WAITING_FOR_WAKEWORD": print(f"[Estado 1: Modo -> Escuchando...] Esperando palabra clave ('{WAKEWORD}')...") text = listen_phrase(recognizer, stream) similarity = calculate_similarity(text, WAKEWORD) print(f"\t[Usuario -> Frase escuchada]: ´{text}´ : Similaridad con ´{WAKEWORD}´: {similarity}%", ) if similarity>55: text = "Te escucho." text_to_voice(text, VOICE_MODEL_PATH, VOICE_OUTPUT_WAV_FILE) conversation_status = "WAITING_FOR_COMMAND" # ------------------------------------------------------------- # ESTADO 2: "WAITING_FOR_COMMAND" -> Esperando un comando ó entrar en modo Conversación Activa # ------------------------------------------------------------- elif conversation_status == "WAITING_FOR_COMMAND": print("[Estado 2: Modo -> Escuchando comando...] Esperando un comando ó frase para iniciar Conversación Activa...") command = listen_phrase(recognizer, stream) print(f"\t[Usuario -> Frase escuchada]: ´{command}´") # Frases posibles para iniciar la "conversación activa" conversation_starters = [ "hay que platicar", "platicar contigo", "quiero platicar contigo", "me gustaría conversar contigo" ] # Detectar si el usuario quiere iniciar una "Conversación Activa" print("\t[Verificar]: Detectar si quiere entrar a Conversación Activa") for phrase in conversation_starters: similarity = calculate_similarity(command, phrase) print(f"\t\tSimilaridad: {similarity}% - Frase usuario: {command} - Frase posible: {phrase}", ) if similarity>80: text = "¡Si Claro! ¿De qué te gustaría hablar?" text_to_voice(text, VOICE_MODEL_PATH, VOICE_OUTPUT_WAV_FILE) conversation_status = "CONVERSATION" break else: # Ejecutando comando execute_robot_action(command) conversation_status = "WAITING_FOR_WAKEWORD" # ------------------------------------------------------------- # ESTADO 3: "CONVERSATION" -> Modo de conversación contínua usando Ollama y un LLM # ------------------------------------------------------------- elif conversation_status == "CONVERSATION": print("[Estado 3: Modo -> Conversación Activa] Esperando respuesta del usuario...") command = listen_phrase(recognizer, stream) print(f"\t[Usuario -> Voz]: {command}") # Frases posibles para terminar la "conversación activa" conversation_exit = [ "hemos terminado la plática", "terminar plática", "terminar conversación", "ya no quiero platicar contigo" ] # Detectar si el usuario quiere terminar una "conversación activa" for phrase in conversation_exit: similarity = calculate_similarity(command, phrase) print(f"\t\tSimilaridad: {similarity}% - Frase usuario: {command} - Frase posible: {phrase}", ) if similarity>80: text = "Entendido! Terminando la plática. Quedo atento." text_to_voice(text, VOICE_MODEL_PATH, VOICE_OUTPUT_WAV_FILE) conversation_status = "WAITING_FOR_WAKEWORD" break if conversation_status == "CONVERSATION": # Procesamiento de conversación usando un LLM local con Ollama robot_voice = llm_conversation(command, LLM_MODEL_CONVERSATION) text_to_voice(robot_voice, VOICE_MODEL_PATH, VOICE_OUTPUT_WAV_FILE) except KeyboardInterrupt: print("\nCerrando sistema de voz...") finally: stream.stop_stream() stream.close() audio.terminate() print("\nFIN OK\n") # Programa principal if __name__ == "__main__": # Definir el contexto de arranque del Modelo antes de las preguntas # Contexto LARGO system_prompt = ( "Eres un robot de servicio autónomo que conversa por voz con personas en casas y oficinas." "Responde siempre en español de forma directa, concisa y muy natural. " "Sé amable, usa oraciones cortas y evita explicaciones largas. " "Nunca uses listas, viñetas, markdown, emojis, tecnicismos ni texto entre paréntesis. " "No incluyas saludos largos, frases ceremoniales, ni listas estructuradas. " "No repitas la pregunta del usuario. " "Si no sabes algo, dilo en una frase corta. " "Limita tu respuesta a una o dos oraciones como máximo, " "salvo que te pidan explícitamente más detalle." ) # Contexto CORTO system_prompt = ( "Te llamas Ponytron y eres un robot de servicio autónomo." "Responde en español de forma concisa, clara y natural en un máximo de 2 oraciones. " ) # Configuraciones iniciales del modelo temperatura: float = 0.4 max_tokens: int = 120 timeout: float = 20.0 # Creación del cliente para Ollama cliente_ollama = Client(host="http://localhost:11434") main()