Desarrollo de apps multimodales con Gemini Live en Vertex AI

Para los directores de tecnología y arquitectos principales, el cambio de las interacciones LLM de "solicitud-respuesta" ainteracciones de flujo bidireccional y estado representa el siguiente avance en la ingeniería de IA. La API Gemini Multimodal Live (potenciada por Gemini 2.0 Flash) permite interacciones de voz y vídeo en tiempo real y de baja latencia que se sienten genuinamente conversacionales.

A diferencia de las soluciones tradicionales que encadenan la conversión de voz a texto (STT), la inferencia de modelos de lenguaje grandes (LLM) y la conversión de texto a voz (TTS)—generando retrasos en cada paso—Gemini Live gestiona la integración de modalidades de forma nativa. Esto unifica el contexto, permitiendo que el modelo "vea" una transmisión de video en vivo e "escuche" interrupciones de forma instantánea.

Este artículo detalla los patrones arquitectónicos para implementar Gemini Live en Vertex AI, centrándose en el protocolo WebSocket, las estrategias de segmentación de audio y la gestión de sesiones necesarias para aplicaciones de nivel empresarial.

Equipo de ingeniería de software compartido bajo demanda, mediante suscripción.

Acceda a un equipo flexible y compartido de ingeniería de software bajo demanda a través de una suscripción mensual predecible. Desarrolladores, diseñadores, ingenieros de control de calidad y un gerente de proyecto gratuito le ayudan a crear MVPs, escalar productos e innovar con tecnologías modernas como React, Node.js y más.

Try 4Geeks Teams

La Arquitectura: Flujo Bidireccional a través de WebSockets

El núcleo de la API Live es el método BidiGenerateContent . A diferencia de los endpoints REST estándar, esto establece una conexión persistente de WebSocket. Este canal de estado permite:

1. Entrada en tiempo real: El cliente transmite audio (PCM) y video (fotogramas JPEG) de forma continua.
2. Eventos del servidor: El servidor envía fragmentos de audio (respuesta), transcripciones de texto y llamadas a herramientas de forma asíncrona.
3. Entrada de voz: Si el usuario habla mientras el modelo está generando audio, el servidor detecta esto (Detección de actividad de voz) y envía una señal de interrupción, permitiendo que el cliente detenga la reproducción inmediatamente.

Flujo de Protocolo:

1. Acuerdo de colaboración: Autentícate a través de OAuth 2.0.
2. Configuración: Envía la configuración inicial (versión del modelo, instrucciones del sistema, ajustes de voz).
3. Bucle de sesión: Envía de forma asíncrona realtime_input (fragmentos de medios) y recibe server_content.

Requisitos

Para implementar esto, necesitas un Proyecto de Google Cloud con la API de Vertex AI habilitada.

• Modelo: gemini-2.0-flash-exp (o la versión más reciente para uso empresarial).
• Región: us-central1 (La disponibilidad de la API suele estar limitada a una región durante la fase de prueba).
• IAM: Asegúrese de que su cuenta de servicio tenga losroles de Usuario de Vertex AI.roles.

Implementación técnica: Cliente Python Asíncrono

Mientras que puedes gestionar las WebSockets de forma directa, el SDK google-genai simplifica la complejidad de la configuración al tiempo que expone los controles necesarios.

A continuación, se presenta un patrón listo para su uso, para un agente de voz basado en la consola. Esta implementación maneja la naturaleza asíncrona del envío de la entrada del micrófono, al tiempo que procesa la salida de audio del modelo.

Dependencias:

pip install google-genai pyaudio

La lógica principal:

import asyncio
import pyaudio
from google import genai

# Configuration
PROJECT_ID = "your-project-id"
LOCATION = "us-central1"
MODEL_ID = "gemini-2.0-flash-exp"

# Audio Settings (Gemini expects 16kHz, 1 channel, 16-bit PCM)
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
CHUNK_SIZE = 1024

client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)

async def audio_stream_generator(input_stream):
    """Yields audio chunks from the microphone."""
    while True:
        data = await asyncio.to_thread(input_stream.read, CHUNK_SIZE, exception_on_overflow=False)
        yield data

async def run_live_session():
    # Initialize PyAudio
    p = pyaudio.PyAudio()
    mic_stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE, input=True, frames_per_buffer=CHUNK_SIZE)
    speaker_stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE, output=True)

    print("--- Connecting to Gemini Live ---")
    
    config = {
        "response_modalities": ["AUDIO"],
        "system_instruction": "You are a senior technical assistant. Be concise and precise."
    }

    try:
        async with client.aio.live.connect(model=MODEL_ID, config=config) as session:
            print("--- Session Active. Speak now. ---")

            # Task 1: Send Audio
            async def send_audio():
                async for chunk in audio_stream_generator(mic_stream):
                    await session.send(input=chunk, end_of_turn=False)

            # Task 2: Receive Audio
            async def receive_audio():
                async for response in session.receive():
                    # Handle Text/Audio parts
                    if response.server_content:
                        model_turn = response.server_content.model_turn
                        if model_turn:
                            for part in model_turn.parts:
                                if part.inline_data: # Audio data
                                    await asyncio.to_thread(speaker_stream.write, part.inline_data.data)
                                    
                    # Handle Interruption (Barge-in)
                    if response.server_content and response.server_content.interrupted:
                         print("\n[Interrupted] Stopping playback...")
                         # In a real GUI app, you would clear the audio buffer here.

            # Run both tasks concurrently
            await asyncio.gather(send_audio(), receive_audio())

    except Exception as e:
        print(f"Session Error: {e}")
    finally:
        mic_stream.stop_stream()
        mic_stream.close()
        speaker_stream.stop_stream()
        speaker_stream.close()
        p.terminate()

if __name__ == "__main__":
    asyncio.run(run_live_session())

Decisiones Arquitectónicas Clave en el Código

1. Concurrencia (asyncio.gather): Los flujos de entrada y salida deben manejarse de forma independiente. Bloquear la entrada del micrófono impedirá que el cliente procese la respuesta de audio del modelo, eliminando el efecto en tiempo real.<s4>Configuración de Modos
2. Configuración de la modalidad: Establecemos explícitamente response_modalities=["AUDIO"]. Esto indica a Gemini que genere audio PCM sin procesar directamente, en lugar de generar texto que el cliente debe sintetizar.
3. Manejo de "Barge-in": La bandera en server_content es crucial. Cuando el modelo detecta el habla del usuario durante su propia salida, se detiene la generación. Su cliente debe escuchar esta bandera para limpiar inmediatamente su búfer de audio local, de lo contrario, el usuario escuchará "audio fantasma" que finaliza una frase después de haberla interrumpido.

Equipo de Ingeniería de Software Compartido bajo Demanda, mediante Suscripción.

Acceda a un equipo flexible y compartido de ingeniería de software bajo demanda a través de una suscripción mensual predecible. Desarrolladores, diseñadores, ingenieros de control de calidad y un gerente de proyecto gratuito le ayudan a construir MVPs, escalar productos e innovar con tecnologías modernas como React, Node.js y más.

Try 4Geeks Teams

Optimización para contextos empresariales

Cuando se trata de escalar los servicios de ingeniería de IA para empresas utilizando esta API, considere los siguientes tres factores:Utilizando esta API, considere los siguientes tres factores:

1. Llamada a la función (invocación de función)

Gemini Live admite la llamada a herramientas en tiempo real. Puede definir herramientas (p. ej., <s1>check_inventoryver inventarioquery_crmconsulta CRM) en el configuraciónconfiguración.

• Flujo: El usuario habla el comando -> El modelo pausa la audio -> Envía el evento <s2>tool_callllamada a herramientaevento -> Cliente ejecuta código -> Cliente envíatool_response-> El modelo vuelve a reproducir el audio con la respuesta.
• Consejo de latencia: Asegúrese de que la ejecución de sus herramientas en el backend estén altamente optimizadas (por ejemplo, búsquedas en Redis frente a consultas SQL lentas) para mantener la ilusión de conversación.

2. Estabilidad de la red y reconexión

Las conexiones WebSockets son frágiles en entornos móviles. Implementar una estrategia de "Reanudación de Sesión" es vital. Aunque la API actual no admite la reanudación de un contexto de sesión exactamente donde se interrumpió (el estado es transitorio), su cliente debe almacenar el historial de la conversación (las interacciones) y reintroducirlos como client_contentcontext al restablecer la conexión.

3. Integración de vídeo

Para aplicaciones de servicio técnico o soporte técnico, puede enviar fotogramas de vídeo.

• Límite de velocidad: No enviar vídeos a 60 fps. Enviar 1-2 fotogramas por segundo (FPS) suele ser suficiente para que el modelo comprenda el contexto visual (por ejemplo, identificar un cable roto) sin incurrir en costes y ancho de banda excesivos.
• Formato: Convertir los fotogramas en fragmentos JPEG codificados en base64 antes de enviarlos a través de realtime_input.

Conclusión

La API Gemini Live nos permite alejarnos de cadenas frágiles y de alta latencia de modelos aislados, hacia motores de razonamiento unificados y multimodales. Para los equipos de ingeniería, el desafío cambia de gestionar la latencia de la tubería a gestionar el estado y la concurrencia de WebSocket.

Si su organización busca crear agentes personalizados que aprovechen estas capacidades en tiempo real para flujos de trabajo complejos, colaborar con un experto en servicios de ingeniería de inteligencia artificial para empresas puede acelerar su camino hacia la producción.ai engineering services for enterprises 

4Geeks ofrece experiencia especializada en ingeniería de productos, crecimiento y IA, ayudando a equipos globales a implementar soluciones de IA robustas y escalables.

Equipo de Ingeniería de Software Compartido Bajo Demanda, Por Suscripción.

Acceda a un equipo flexible y compartido de ingeniería de software bajo demanda a través de una suscripción mensual predecible. Desarrolladores, diseñadores, ingenieros de control de calidad y un gerente de proyecto gratuito le ayudan a crear MVPs, escalar productos e innovar con tecnologías modernas como React, Node.js y más.

Try 4Geeks Teams

Preguntas frecuentes