Creación de un ciclo de conversación por voz
Ciclo grabar → transcribir → razonar → hablar con gestión de interrupciones.
Creación de un ciclo de conversación por voz es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 3 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.
El ciclo de conversación de voz
Un ciclo completo de conversación de voz conecta la entrada del micrófono con la salida del agente mediante un ciclo continuo: grabar → transcribir → agente → TTS → reproducir → repetir.
En esta lección se aborda cada componente y los desafíos que diferencian los ciclos de voz de los agentes basados en texto: detección de silencios, gestión de interrupciones y detección del fin del habla.
Grabación desde el micrófono
Utilice sounddevice para capturar audio del micrófono predeterminado. Grabe durante un tiempo fijo o hasta que se detecte silencio. Grabe siempre en mono a 16 kHz, que es la frecuencia de muestreo que espera Whisper.
Instálelo con pip install sounddevice soundfile numpy.
import sounddevice as sd
import numpy as np
import tempfile
import soundfile as sf
SAMPLE_RATE = 16000 # 16kHz mono — optimal for Whisper
DURATION = 5 # seconds
def record_audio(duration=DURATION, sample_rate=SAMPLE_RATE):
print(f'Recording for {duration} seconds...')
audio_data = sd.rec(
int(duration * sample_rate),
samplerate=sample_rate,
channels=1,
dtype='float32'
)
sd.wait() # block until recording finishes
print('Recording complete')
return audio_data, sample_rate
def audio_to_file(audio_data, sample_rate, filepath='/tmp/recording.wav'):
sf.write(filepath, audio_data, sample_rate)
return filepathDetección de silencios para identificar el fin del habla
Grabar durante un tiempo fijo resulta poco práctico: los usuarios deben esperar incluso si terminaron de hablar en 2 segundos. La detección de silencios detiene automáticamente la grabación cuando el usuario ha permanecido en silencio durante un periodo determinado.
import sounddevice as sd
import numpy as np
from collections import deque
SAMPLE_RATE = 16000
CHUNK_SIZE = 1024 # samples per chunk
SILENCE_THRESHOLD = 0.02 # RMS volume below this = silence
SILENCE_DURATION = 1.5 # seconds of silence to end recording
MAX_DURATION = 30 # max recording length in seconds
def record_until_silence():
chunks = []
silent_chunks = 0
max_silent = int(SILENCE_DURATION * SAMPLE_RATE / CHUNK_SIZE)
max_chunks = int(MAX_DURATION * SAMPLE_RATE / CHUNK_SIZE)
print('Listening... (speak now)')
with sd.InputStream(samplerate=SAMPLE_RATE, channels=1,
blocksize=CHUNK_SIZE, dtype='float32') as stream:
while True:
chunk, _ = stream.read(CHUNK_SIZE)
chunks.append(chunk.copy())
rms = np.sqrt(np.mean(chunk**2))
if rms < SILENCE_THRESHOLD:
silent_chunks += 1
else:
silent_chunks = 0 # speech detected — reset counter
if silent_chunks >= max_silent and len(chunks) > max_silent:
break
if len(chunks) >= max_chunks:
break
audio = np.concatenate(chunks, axis=0)
return audio, SAMPLE_RATEEl ciclo de voz completo
Conecte todos los componentes en un ciclo continuo. Después de que el agente responda, empiece a escuchar de nuevo inmediatamente para crear una conversación natural de ida y vuelta.
import openai
import os
client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
conversation_history = []
def voice_conversation_loop():
print('Voice agent started. Say something (Ctrl+C to exit).')
while True:
# 1. Record user
audio_data, sample_rate = record_until_silence()
audio_path = audio_to_file(audio_data, sample_rate)
# 2. Transcribe
user_text = transcribe_file(audio_path)
print(f'You: {user_text}')
if not user_text.strip():
continue # empty — listen again
# 3. Run agent
conversation_history.append({'role': 'user', 'content': user_text})
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'system', 'content': 'You are a helpful voice assistant. Keep responses concise.'}] + conversation_history[-10:]
)
agent_text = response.choices[0].message.content
conversation_history.append({'role': 'assistant', 'content': agent_text})
print(f'Agent: {agent_text}')
# 4. TTS and play
say(agent_text) # speaks sentence by sentenceConcepto de detección de la palabra de activación
La escucha permanente es costosa (por las llamadas continuas a Whisper) y afecta a la privacidad. La detección de la palabra de activación ejecuta un modelo local ligero que solo activa el procesamiento completo cuando se escucha una frase específica (por ejemplo, «Hey Agent»).
# Wake word detection with pvporcupine (Picovoice)
# pip install pvporcupine sounddevice
import pvporcupine
import sounddevice as sd
import numpy as np
import os
def listen_for_wake_word(wake_word='computer'):
porcupine = pvporcupine.create(
access_key=os.getenv('PICOVOICE_KEY'),
keywords=[wake_word] # built-in: computer, hey barista, hey google, jarvis...
)
print(f'Listening for wake word: "{wake_word}"...')
with sd.InputStream(
samplerate=porcupine.sample_rate,
channels=1,
dtype='int16',
blocksize=porcupine.frame_length
) as stream:
while True:
frame, _ = stream.read(porcupine.frame_length)
pcm = frame.flatten().astype('int16')
result = porcupine.process(pcm)
if result >= 0:
print(f'Wake word detected!')
porcupine.delete()
return TrueGestión de interrupciones
Los usuarios deben poder interrumpir al agente mientras habla. Implemente la gestión de interrupciones ejecutando la reproducción en un hilo independiente y supervisando el volumen del micrófono: si el usuario empieza a hablar, detenga la reproducción inmediatamente.
import threading
import sounddevice as sd
import numpy as np
interrupt_event = threading.Event()
def monitor_for_interrupt(threshold=0.03):
def audio_callback(indata, frames, time_info, status):
rms = np.sqrt(np.mean(indata**2))
if rms > threshold:
print('Interrupt detected!')
interrupt_event.set()
with sd.InputStream(callback=audio_callback, channels=1,
samplerate=16000, blocksize=1024):
while not interrupt_event.is_set():
sd.sleep(50)
def speak_with_interrupt(text, voice='nova'):
interrupt_event.clear()
# Start interrupt monitor in background
monitor_thread = threading.Thread(target=monitor_for_interrupt, daemon=True)
monitor_thread.start()
# Play audio sentence by sentence
for sentence in split_into_sentences(text):
if interrupt_event.is_set():
print('Playback interrupted')
break
audio_path = cached_tts(sentence, voice)
play_audio_file(audio_path)Filtrado de ruido
El audio del micrófono suele contener ruido de fondo: ventiladores, clics del teclado y eco de la habitación. Aplique una puerta de ruido sencilla para suprimir el audio por debajo de un umbral y, opcionalmente, utilice un detector de actividad de voz (VAD) para mejorar la precisión.
import numpy as np
NOISE_GATE_THRESHOLD = 0.015 # RMS threshold
def apply_noise_gate(audio_data, threshold=NOISE_GATE_THRESHOLD):
rms = np.sqrt(np.mean(audio_data**2))
if rms < threshold:
return np.zeros_like(audio_data) # silence below threshold
return audio_data
def has_speech_content(audio_data, threshold=0.02):
rms_values = [
np.sqrt(np.mean(audio_data[i:i+1600]**2))
for i in range(0, len(audio_data), 1600)
]
speech_frames = sum(1 for r in rms_values if r > threshold)
speech_ratio = speech_frames / max(len(rms_values), 1)
return speech_ratio > 0.1 # at least 10% of frames have speech
# Use in loop before transcribing
audio, sr = record_until_silence()
if has_speech_content(audio):
text = transcribe_file(audio_to_file(audio, sr))
else:
print('No speech detected — listening again')Máquina de estados para el ciclo de conversación
Un ciclo de voz sólido utiliza una máquina de estados para realizar un seguimiento de lo que hace el agente en cada momento y evitar condiciones de carrera entre la reproducción y la escucha.
from enum import Enum
class AgentState(Enum):
IDLE = 'idle'
LISTENING = 'listening'
TRANSCRIBING = 'transcribing'
THINKING = 'thinking'
SPEAKING = 'speaking'
current_state = AgentState.IDLE
def set_state(new_state):
global current_state
print(f'State: {current_state.value} -> {new_state.value}')
current_state = new_state
def voice_loop_with_state():
while True:
set_state(AgentState.LISTENING)
audio, sr = record_until_silence()
if not has_speech_content(audio):
continue
set_state(AgentState.TRANSCRIBING)
text = transcribe_file(audio_to_file(audio, sr))
if not text.strip():
continue
print(f'You: {text}')
set_state(AgentState.THINKING)
agent_response = run_agent(text)
print(f'Agent: {agent_response}')
set_state(AgentState.SPEAKING)
speak_with_interrupt(agent_response)Gestión del contexto de la conversación
El agente de voz mantiene el historial de la conversación para poder responder preguntas de seguimiento como «Cuénteme más sobre eso» o «¿Cuál dijo que era el precio?». Mantenga el historial en memoria y redúzcalo a los últimos N turnos para evitar que se desborde el contexto.
import openai
import os
client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
MAX_HISTORY_TURNS = 10
SYSTEM_PROMPT = 'You are a concise voice assistant. Keep responses under 3 sentences unless asked for detail.'
class VoiceAgent:
def __init__(self):
self.history = []
def respond(self, user_text):
self.history.append({'role': 'user', 'content': user_text})
# Trim history to last N turns
recent = self.history[-(MAX_HISTORY_TURNS * 2):]
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'system', 'content': SYSTEM_PROMPT}] + recent
)
agent_text = response.choices[0].message.content
self.history.append({'role': 'assistant', 'content': agent_text})
return agent_text
def reset(self):
self.history = []
print('Conversation history cleared')Cierre y limpieza ordenados
Gestione Ctrl+C y otras señales de salida correctamente: detenga el flujo de audio, elimine los archivos temporales y despídase antes de apagar el agente.
import signal
import sys
import glob
import os
TMP_DIR = '/tmp'
agent = None
def cleanup_temp_files():
for f in glob.glob(os.path.join(TMP_DIR, 'recording_*.wav')):
os.unlink(f)
print('Temp files cleaned up')
def graceful_shutdown(signum, frame):
print('\nShutting down voice agent...')
if agent:
say('Goodbye! Have a great day.')
cleanup_temp_files()
sys.exit(0)
# Register signal handler
signal.signal(signal.SIGINT, graceful_shutdown)
signal.signal(signal.SIGTERM, graceful_shutdown)
# Main entry point
agent = VoiceAgent()
print('Voice agent ready. Press Ctrl+C to exit.')
voice_loop_with_state()Registro de la conversación
Mantenga un registro persistente de la conversación con marcas de tiempo, texto transcrito, respuestas del agente y cualquier error. Esto es esencial para depurar problemas de los agentes de voz y crear funciones de revisión de conversaciones.
import json
import os
from datetime import datetime
LOG_DIR = '/tmp/voice_agent_logs'
os.makedirs(LOG_DIR, exist_ok=True)
conversation_log = []
def log_turn(speaker, text, latency_ms=None, error=None):
entry = {
'timestamp': datetime.now().isoformat(),
'speaker': speaker, # 'user' or 'agent'
'text': text,
'latency_ms': latency_ms
}
if error:
entry['error'] = error
conversation_log.append(entry)
def save_conversation_log(session_id=None):
if not session_id:
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
log_path = os.path.join(LOG_DIR, f'session_{session_id}.json')
with open(log_path, 'w') as f:
json.dump({
'session_id': session_id,
'turns': len(conversation_log),
'log': conversation_log
}, f, indent=2)
print(f'Log saved: {log_path}')
return log_path
# Usage in voice loop
log_turn('user', transcribed_text, latency_ms=420)
log_turn('agent', agent_response, latency_ms=850)Comprobación de conocimientos
¿Cuál es el objetivo de la detección de silencios en un ciclo de conversación de voz?
Resumen: construcción de un ciclo de conversación de voz
Un ciclo de conversación de voz: grabar desde el micrófono con detección de silencios → transcribir con Whisper → ejecutar el agente con el historial de la conversación → responder con TTS oración por oración → reproducir con supervisión de interrupciones → repetir.
Componentes clave: una máquina de estados para evitar condiciones de carrera, detección de la palabra de activación para la escucha permanente, una puerta de ruido para la calidad del audio y el recorte del historial de la conversación para gestionar el contexto. Gestione Ctrl+C correctamente mediante una limpieza adecuada y un mensaje de despedida.
Preguntas frecuentes
¿La lección «Creación de un ciclo de conversación por voz» es gratis?
Sí — el texto completo de «Creación de un ciclo de conversación por voz» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.
¿Qué aprenderé en «Creación de un ciclo de conversación por voz»?
Ciclo grabar → transcribir → razonar → hablar con gestión de interrupciones. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Agents?
No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 3 de 4.
¿Cuánto tiempo toma la lección «Creación de un ciclo de conversación por voz»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Agents?
Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Conversión de voz a texto con Whisper y Deepgram
- Conversión de texto a voz en las respuestas de agentes
- Creación de un ciclo de conversación por voz
- Optimización de la latencia para agentes de voz