Tutorial paso a paso: creación de un asistente de IA de audio y vídeo de baja latencia basado en Gemini Multimodal Live API

Gemini Live API admite de forma nativa entrada y salida de audio, y la arquitectura de extremo a extremo reduce la latencia a menos de 500 ms sin necesidad de transferencia ASR/TTS. Este artículo comparte cómo utilizar este conjunto de API para crear un asistente de IA que realmente pueda tener conversaciones en tiempo real.

¿Qué es la API en vivo multimodal de Gemini?

Primero aclaremos las cosas. ¿Cómo funciona la API tradicional de Gemini? Le envías un texto y te responde con un texto, sencillo y directo. Pero si desea realizar interacción de voz, debe conectar usted mismo ASR (reconocimiento de voz) y TTS (síntesis de voz). Va y viene en el medio y el retraso aumenta repentinamente.

Lo que es diferente de Gemini Multimodal Live API es que admite de forma nativa entrada y salida de audio. En otras palabras, el sonido recopilado por su micrófono puede enviarse directamente a él y devolverá una secuencia de audio puro sin necesidad de realizar ninguna conversión de formato por su parte. Este diseño de arquitectura de extremo a extremo reduce directamente el retraso a menos de 500 ms.

Probé esta función mientras trabajaba en un proyecto de hogar inteligente. El usuario dijo “apaga las luces de la sala de estar” y la IA respondió casi tan pronto como se pronunciaron las palabras. La suavidad realmente hará que la gente olvide que la persona de enfrente es un programa.

El modelo actualmente admitido es gemini-2.0-flash-native-audio-preview. Preste atención a este número de versión. Google todavía está iterando rápidamente. Se recomienda prestar atención a las actualizaciones periódicamente.

Diseño de arquitectura y selección de tecnología.

Bien, ahora hablemos de cómo configurar este sistema. Mi sugerencia es una arquitectura con front-end y back-end separados. La razón es muy simple: La clave API no se puede exponer en el front-end.

El flujo de datos general es así:

[Navegador] --WebSocket--> [Proxy Python backend] --WebSocket--> [Gemini Live API]
   |                           |                           |
Captura micrófono           Relay + lógica de negocio   Procesamiento IA
Reproducción altavoz        VAD / control de interrupción  Generación de audio

Quizás esté pensando, ¿por qué el navegador no puede conectarse directamente a Gemini? Técnicamente es posible, pero significa escribir la clave API en JavaScript, y cualquiera que abra las herramientas de desarrollador podrá obtener su clave. Hice este tipo de cosas una vez y la factura se disparó al día siguiente. Aprendí una lección profunda.

Entonces nuestra pila de tecnología está configurada así:

El AudioWorklet en Web Audio API es algo bueno. Puede procesar audio en un hilo separado sin bloquear el hilo principal. Daré el código de implementación específico más adelante.

Establecimiento de conexión WebSocket y gestión de sesiones

Bien, ahora empieza a escribir código. Lo primero que hay que resolver es cómo conectarse al servicio de Gemini.

Railway - Plataforma de despliegue moderna, sin ops complejas, en producción en minutos

Empezar ahora →Anuncio

El punto final WebSocket de Live API tiene este aspecto:

wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1alpha.GenerativeService.BidiGenerateContent?key=YOUR_API_KEY

Preste atención a v1alpha, que indica que todavía es una versión preliminar, la interfaz puede cambiar, así que tenga cuidado al usarla en un entorno de producción.

Una vez establecida la conexión, lo primero es enviar un mensaje de Configuración para indicarle a Gemini cómo quieres chatear:

import asyncio
import json
import websockets

GEMINI_API_KEY = "your-api-key-here"
GEMINI_WS_URL = (
    f"wss://generativelanguage.googleapis.com/ws/"
    f"google.ai.generativelanguage.v1alpha.GenerativeService.BidiGenerateContent"
    f"?key={GEMINI_API_KEY}"
)

CONFIG = {
    "setup": {
        "model": "models/gemini-2.0-flash-native-audio-preview",
        "generation_config": {
            "response_modalities": ["AUDIO"],
            "speech_config": {
                "voice_config": {
                    "prebuilt_voice_config": {
                        "voice_name": "Charon"  # Opcional: Charon, Aoede, etc.
                    }
                }
            }
        },
        "system_instruction": {
            "parts": [{"text": "Eres un asistente de IA útil; responde de forma concisa y natural."}]
        }
    }
}

async def connect():
    async with websockets.connect(GEMINI_WS_URL) as ws:
        # Enviar configuración setup
        await ws.send(json.dumps(CONFIG))

        # Esperar respuesta setupComplete
        response = await ws.recv()
        data = json.loads(response)

        if "setupComplete" in data:
            print("✅ Conexión establecida; puedes empezar a conversar")
            return ws
        else:
            raise Exception(f"Setup fallido: {data}")

Hay varios parámetros de los que vale la pena hablar aquí:

• response_modalities: Establecido en ["AUDIO"] para indicar que solo queremos respuestas de voz. Si también desea texto, puede cambiarlo a ["AUDIO", "TEXT"]
• voice_name: Gemini proporciona varios sonidos preestablecidos. Prefiero “Charon”, que suena más tranquilo.

Cuando se trata de desconexión y reconexión, recomiendo utilizar una estrategia de retroceso exponencial. No vuelvas a intentarlo tan pronto como inicies, lo que puede bloquear fácilmente el servicio:

async def connect_with_retry(max_retries=5):
    for attempt in range(max_retries):
        try:
            return await connect()
        except Exception as e:
            wait_time = min(2 ** attempt, 30)  # Esperar como máximo 30 segundos
            print(f"Conexión fallida ({e}); reintentando en {wait_time}s...")
            await asyncio.sleep(wait_time)
    raise Exception("No se pudo conectar tras varios reintentos")

Recopilación y transmisión de flujo de audio PCM de 16 kHz

Bien, ahora que se establece la conexión, el siguiente paso es solucionar el problema de dónde viene el audio y cómo enviarlo allí.

Primero hablemos de por qué elegimos 16 kHz. El rango de frecuencia de las voces humanas generalmente está entre 85 Hz y 255 Hz (las voces masculinas son más bajas y las femeninas más altas). Según el teorema de muestreo de Nyquist, 8 kHz es teóricamente suficiente. Pero, de hecho, es necesario conservar algunos detalles. 16 kHz es un punto óptimo, que garantiza la calidad del sonido sin aumentar demasiado la cantidad de datos. Gemini recomienda oficialmente esta frecuencia de muestreo.

El código de la colección front-end se ve así:

class AudioRecorder {
  constructor() {
    this.sampleRate = 16000;
    this.bufferSize = 1024;
    this.audioContext = null;
    this.workletNode = null;
    this.stream = null;
    this.onAudioData = null; // Función callback
  }

  async start() {
    // Solicitar permiso de micrófono
    this.stream = await navigator.mediaDevices.getUserMedia({
      audio: {
        sampleRate: 16000,
        channelCount: 1,
        echoCancellation: true,
        noiseSuppression: true
      }
    });

    // Crear AudioContext con frecuencia de muestreo fija
    this.audioContext = new AudioContext({
      sampleRate: 16000
    });

    // Cargar procesador AudioWorklet
    await this.audioContext.audioWorklet.addModule('pcm-processor.js');

    const source = this.audioContext.createMediaStreamSource(this.stream);
    this.workletNode = new AudioWorkletNode(this.audioContext, 'pcm-processor');

    // Procesar datos de audio
    this.workletNode.port.onmessage = (event) => {
      const float32Data = event.data;

      // Convertir a PCM Int16
      const int16Data = this.float32ToInt16(float32Data);

      // Codificar en Base64 y enviar
      const base64Data = btoa(String.fromCharCode(...new Uint8Array(int16Data.buffer)));

      if (this.onAudioData) {
        this.onAudioData(base64Data);
      }
    };

    source.connect(this.workletNode);
    console.log('🎤 Captura de audio iniciada');
  }

  float32ToInt16(float32Array) {
    const int16Array = new Int16Array(float32Array.length);
    for (let i = 0; i < float32Array.length; i++) {
      // Float32 (-1.0 ~ 1.0) -> Int16 (-32768 ~ 32767)
      const s = Math.max(-1, Math.min(1, float32Array[i]));
      int16Array[i] = s < 0 ? s * 0x8000 : s * 0x7FFF;
    }
    return int16Array;
  }

  stop() {
    if (this.workletNode) {
      this.workletNode.disconnect();
    }
    if (this.audioContext) {
      this.audioContext.close();
    }
    if (this.stream) {
      this.stream.getTracks().forEach(track => track.stop());
    }
    console.log('🛑 Captura de audio detenida');
  }
}

AudioWorklet requiere un archivo separado pcm-processor.js:

// pcm-processor.js
class PCMProcessor extends AudioWorkletProcessor {
  process(inputs, outputs, parameters) {
    const input = inputs[0];
    if (input && input[0]) {
      // Enviar al hilo principal
      this.port.postMessage(input[0].slice());
    }
    return true; // Mantener el procesador activo
  }
}

registerProcessor('pcm-processor', PCMProcessor);

Después de que el backend recibe los datos, los reenvía a Gemini:

async def send_audio(ws, base64_pcm_data):
    """Enviar datos de audio a Gemini"""
    message = {
        "realtime_input": {
            "media_chunks": [{
                "mime_type": "audio/pcm;rate=16000",
                "data": base64_pcm_data
            }]
        }
    }
    await ws.send(json.dumps(message))

Aquí hay un problema que me gustaría recordarle: el getUserMedia de algunos navegadores ignorará el sampleRate que especifique y el valor real devuelto puede ser 44,1 kHz o 48 kHz. Para estar seguro, es mejor volver a muestrearlo en AudioContext, o simplemente usar una biblioteca de terceros como audiobuffer-to-wav para manejarlo.

Implementación de detección de actividad de voz VAD

Ahora nos enfrentamos a un problema: si todo el audio se envía a Gemini independientemente de la situación, los datos también se transmitirán durante el silencio, lo que desperdicia ancho de banda y dinero. En este momento, es necesario que aparezca VAD (Detección de actividad de voz, Detección de actividad de voz).

La función de VAD es simple: determinar si alguien está hablando en este audio. Habla sólo cuando alguien esté hablando y descansa cuando nadie esté hablando.

Recomiendo utilizar WebRTC VAD de código abierto de Google, que es liviano, rápido y tiene buenos resultados. Python tiene una biblioteca empaquetada webrtcvad:

import webrtcvad
import collections
import numpy as np

class VADProcessor:
    def __init__(self, aggressiveness=2, frame_duration_ms=20):
        """
        aggressiveness: 0-3; cuanto más alto, más estricto (más fácil clasificar voz como silencio)
        frame_duration_ms: 10, 20, or 30
        """
        self.vad = webrtcvad.Vad(aggressiveness)
        self.frame_duration_ms = frame_duration_ms
        self.sample_rate = 16000

        # Búfer circular para suavizado
        self.ring_buffer = collections.deque(maxlen=30)  # 600ms
        self.triggered = False

    def process_frame(self, pcm_bytes):
        """
        Procesar un frame de audio; devuelve si debe enviarse
        """
        is_speech = self.vad.is_speech(pcm_bytes, self.sample_rate)

        if not self.triggered:
            # Estado no activado: acumular frames de voz
            self.ring_buffer.append((pcm_bytes, is_speech))
            num_voiced = sum(1 for _, speech in self.ring_buffer if speech)

            # Si el 90% de los frames son voz, activar
            if num_voiced > 0.9 * self.ring_buffer.maxlen:
                self.triggered = True
                # Enviar también los datos del búfer
                return b''.join([f for f, _ in self.ring_buffer])
            return None
        else:
            # Estado activado
            if is_speech:
                self.ring_buffer.append((pcm_bytes, True))
                return pcm_bytes
            else:
                self.ring_buffer.append((pcm_bytes, False))
                num_unvoiced = sum(1 for _, speech in self.ring_buffer if not speech)

                # Si el 90% es silencio, finalizar activación
                if num_unvoiced > 0.9 * self.ring_buffer.maxlen:
                    self.triggered = False
                    self.ring_buffer.clear()
                return pcm_bytes

Probablemente se vea así cuando se usa:

Alibaba Cloud - Opción preferida para desarrolladores, 15 % de descuento exclusivo en Lighthouse

Obtener 15% dto. →Anuncio

vad = VADProcessor(aggressiveness=2)

async def handle_client_audio(websocket, gemini_ws):
    async for message in websocket:
        data = json.loads(message)

        if 'audio' in data:
            pcm_bytes = base64.b64decode(data['audio'])

            # Detección VAD
            result = vad.process_frame(pcm_bytes)

            if result:
                # Hay voz; reenviar a Gemini
                await send_audio(gemini_ws, base64.b64encode(result).decode())

agresividad Este parámetro es bastante sutil. Si lo pones demasiado bajo, un poco de ruido de fondo se considerará habla; Si lo configura demasiado alto, es posible que se pierda el habla suave. Mi experiencia es comenzar con 2 y ajustar según el escenario real.

Si su entorno de implementación no puede instalar webrtcvad, también puede utilizar la detección de umbral de energía simple como alternativa:

// Alternativa frontend: detección simple por energía RMS
function detectVoiceActivity(audioData, threshold = 0.015) {
    const sum = audioData.reduce((acc, val) => acc + val * val, 0);
    const rms = Math.sqrt(sum / audioData.length);
    return rms > threshold;
}

Implementación de la función de irrupción natural.

No sé si tienes esta sensación: cuando chateas con algunos asistentes de voz, una vez que comienza a hablar durante mucho tiempo, solo puedes esperar y no puedes interrumpir aunque quieras, lo cual es muy frustrante.

La función de irrupción resuelve este problema. Cuando la IA está hablando, el usuario puede interrumpir directamente y la IA detendrá inmediatamente la salida actual y escuchará lo que el usuario tiene que decir.

La buena noticia es que Gemini Live API admite esta función de forma nativa y lo hace de manera bastante inteligente. Sólo necesitas habilitar la detección automática de actividad en la configuración:

CONFIG = {
    "setup": {
        "model": "models/gemini-2.0-flash-native-audio-preview",
        "generation_config": {
            "response_modalities": ["AUDIO"],
        },
        "realtime_input_config": {
            "automatic_activity_detection": {
                "disabled": False,
                "start_of_speech_sensitivity": "START_SENSITIVITY_HIGH",
                "end_of_speech_sensitivity": "END_SENSITIVITY_LOW"
            }
        }
    }
}

La configuración de “sensibilidad” es un poco complicada:

• start_of_speech_sensitivity establecido en HIGH significa que la IA es más sensible al usuario que comienza a hablar y es más probable que provoque interrupciones
• end_of_speech_sensitivity está configurado en LOW, lo que significa que la IA esperará un momento para confirmar que el usuario realmente ha terminado de hablar antes de responder, para evitar errores de juicio.

Lo que el cliente tiene que hacer aquí es escuchar el evento “interrumpido” y luego dejar de reproducir inmediatamente:

class GeminiClient {
  constructor() {
    this.audioQueue = [];
    this.isPlaying = false;
    this.currentSource = null;
  }

  async handleMessage(event) {
    const message = JSON.parse(event.data);

    // Procesar señal de interrupción
    if (message.server_content?.interrupted) {
      console.log('⚡ El usuario interrumpió; deteniendo reproducción');
      this.stopPlayback();
      return;
    }

    // Procesar audio devuelto por la IA
    if (message.server_content?.model_turn) {
      const parts = message.server_content.model_turn.parts;

      for (const part of parts) {
        if (part.inline_data?.mime_type.startsWith('audio/')) {
          const audioData = base64ToArrayBuffer(part.inline_data.data);
          this.queueAudio(audioData);
        }
      }
    }
  }

  stopPlayback() {
    // Vaciar cola de reproducción
    this.audioQueue = [];
    this.isPlaying = false;

    // Detener el audio que se está reproduciendo
    if (this.currentSource) {
      try {
        this.currentSource.stop();
      } catch (e) {
        // Puede que ya se haya detenido
      }
      this.currentSource = null;
    }
  }

  async queueAudio(audioData) {
    this.audioQueue.push(audioData);
    if (!this.isPlaying) {
      this.playNext();
    }
  }

  async playNext() {
    if (this.audioQueue.length === 0) {
      this.isPlaying = false;
      return;
    }

    this.isPlaying = true;
    const audioData = this.audioQueue.shift();

    // Decodificar y reproducir
    const audioBuffer = await this.audioContext.decodeAudioData(audioData.slice());
    this.currentSource = this.audioContext.createBufferSource();
    this.currentSource.buffer = audioBuffer;
    this.currentSource.connect(this.audioContext.destination);

    this.currentSource.onended = () => {
      this.playNext();
    };

    this.currentSource.start();
  }
}

Hay un detalle a tener en cuenta aquí: el método stop() puede generar una excepción si el audio ha terminado de reproducirse de forma natural. Así que agregué un try-catch para evitar que la consola se volviera roja.

Optimización del rendimiento y control de retrasos.

Finalmente, hablemos de cómo minimizar la latencia de este sistema.

Primero necesitamos saber de dónde viene el retraso:

1. Transmisión de red: el tiempo de ida y vuelta de un paquete de datos desde el navegador al servidor hasta Gemini.
2. Códec de audio: tiempo de compresión/descompresión PCM (pero el PCM en sí no tiene pérdidas, por lo que esta sobrecarga es muy pequeña)
3. Acumulación de búfer: profundidad del búfer configurada para una reproducción fluida

Respecto a estos puntos, mi experiencia en optimización es:

Reducir la profundidad del buffer

No configure el búfer de reproducción demasiado grande, solo lo suficiente. Normalmente uso 100-200 ms:

// Configurar un búfer pequeño
const audioContext = new AudioContext({
  sampleRate: 16000,
  latencyHint: 'interactive'  // Modo de baja latencia
});

Velocidad de bits adaptativa (en realidad, lo principal aquí es el almacenamiento en búfer adaptativo)

KiloClaw - OpenClaw gestionado para empresas: orquestación de agentes IA y enrutamiento inteligente

Empezar →Anuncio

Si se detecta que la fluctuación de la red es relativamente grande, puede aumentar el búfer de manera adecuada; cuando la red sea estable, redúzcala.

Cancelación de eco local

Si el usuario enciende el altavoz en lugar de usar auriculares, el micrófono captará el sonido de la IA, formando un bucle. Afortunadamente, getUserMedia viene con cancelación de eco:

navigator.mediaDevices.getUserMedia({
  audio: {
    echoCancellation: true,
    noiseSuppression: true,
    autoGainControl: true
  }
})

Indicadores de seguimiento

¿Cómo saber si su optimización es efectiva? Puede utilizar la API de rendimiento para administrar:

// Registrar métricas de latencia
class LatencyMonitor {
  constructor() {
    this.metrics = [];
  }

  recordSendTime() {
    this.lastSendTime = performance.now();
  }

  recordReceiveTime() {
    const latency = performance.now() - this.lastSendTime;
    this.metrics.push(latency);

    // Mantener las últimas 100 entradas
    if (this.metrics.length > 100) {
      this.metrics.shift();
    }

    // Calcular latencia media
    const avg = this.metrics.reduce((a, b) => a + b, 0) / this.metrics.length;
    console.log(`📊 Latencia media: ${avg.toFixed(2)}ms`);
  }
}

Los datos que medí en el entorno de prueba son aproximadamente:

• Latencia de extremo a extremo: 300-500 ms (depende de las condiciones de la red)
• Tiempo de respuesta del primer paquete: 200-400 ms
• Retraso del diálogo continuo: 150-300 ms

Si su latencia es significativamente mayor que este valor, puede solucionar el problema de acuerdo con esta lista:

• [] ¿La conexión WebSocket se realiza a través de HTTPS/WSS? HTTP tiene una sobrecarga adicional
• [] ¿Dónde está implementado el servidor? Cuanto más cerca estés de los centros de datos de Google, mejor
• ¿La detección de VAD introduce demasiado retraso? Intente reducir la longitud del marco.
• [] ¿El búfer de reproducción frontal está configurado demasiado grande?

Otro problema tiene que ver con el contexto de audio: Chrome requiere la interacción del usuario antes de reproducir sonido, así que recuerde agregar un botón “Iniciar conversación” a la página y no lo reproduzca automáticamente tan pronto como aparezca.

Resumen

En este punto, hemos completado un proceso completo de desarrollo de la aplicación Gemini Live API. Desde la introducción inicial del concepto hasta el diseño arquitectónico, la conexión WebSocket, la recopilación de audio, la detección de VAD, la función de interrupción y, finalmente, la optimización del rendimiento, hice todo lo posible para compartir los obstáculos que encontré en cada paso.

Para ser honesto, el campo de la interacción de voz en tiempo real todavía se está desarrollando rápidamente y la API de Gemini Live se actualiza constantemente. Pero creo que esta infraestructura puede resistir la prueba; al menos mi propio proyecto ha estado funcionando durante varios meses y la estabilidad es bastante buena.

Si encuentra algún problema durante el desarrollo real, no dude en comunicarse. Después de todo, cuando se trata de tecnología, siempre es más lento para una sola persona explorarla. Sólo discutiéndolo juntos podremos avanzar más rápido.