Gemini en vivo en Vertex AI: Arquitectura de transmisión multimodal en tiempo real

Para los Directores de Tecnología y Arquitectos Líderes, el cambio de las interacciones LLM "solicitud-respuesta" a un flujo de datosestatístico y bidireccional 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 video en tiempo real con baja latencia que resultan genuinamente conversacionales.

A diferencia de las plataformas tradicionales que enlazan la conversión de voz a texto (STT), la inferencia con 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 entre modalidades de forma nativa. Esto unifica el contexto, permitiendo que el modelo "vea" una transmisión de vídeo en vivo e "escuche" interrupciones al instante.

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

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

La Arquitectura: Transmisión bidireccional mediante WebSockets

El núcleo de la API en vivo es el método <s1>BidiGenerateContentGeneración de contenido automatizadamétodo. A diferencia de los puntos finales estándar REST, esto establece una conexión WebSocket persistente. Este canal con 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 en directo: Si el usuario habla mientras el modelo está emitiendo audio, el servidor detecta esto (Detección de Actividad Vocal) y envía una señal interrumpida, lo que permite al cliente detener la reproducción inmediatamente.

Flujo de Protocolo:

1. Acuerdo: 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íncronarealtime_input (fragmentos multimedia) y recibeserver_content.

Requisitos

Para implementar esto, necesitará un proyecto de Google Cloud con la API de Vertex AI habilitada.

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

Implementación Técnica: Cliente Asíncrono en Python

Si bien puedes gestionar las conexiones WebSocket directamente, el SDK <s1>google-genaigoogle-genaiEl SDK simplifica la complejidad del enmarcado al tiempo que expone los controles necesarios.

A continuación, se presenta un patrón listo para su uso en la creación de agentes de voz basados en consola. Esta implementación gestiona la naturaleza asíncrona del envío de datos de 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 ser manejados independientemente. Bloquear la entrada del micrófono impedirá que el cliente procese la respuesta de audio del modelo, destruyendo así el efecto en tiempo real.<s4>Configuración de Modalidad
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 texto que el cliente debe sintetizar.
3. Manejo de "Barge-in": La bandera "interrumpida" en server_content es crucial. Cuando el modelo detecta el habla del usuario durante su propia salida, deja de generar. Su cliente <s15>debe escuchar esta bandera para limpiar inmediatamente su búfer de audio local, o el usuario escuchará audio "fantasma" que completa la 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 crear MVPs, escalar productos e innovar con tecnologías modernas como React, Node.js, y más.

Try 4Geeks Teams

Optimizando para contextos empresariales

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

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

Gemini Live soporta la llamada de herramientas en tiempo real. Puedes definir las herramientas (por ejemplo, <s1>check_inventoryver inventarioquery_crmsetup) en el configuraciónconfiguración.

• Flujo: El usuario habla el comando -> El modelo pausa la audio -> Envía evento tool_call -> El cliente ejecuta el código -> El cliente envía respuesta tool_response -> El modelo reanuda la audio con la respuesta.
• Consejo sobre latencia: Asegúrese de que las ejecuciones de herramientas en su 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 "Resumen de Sesión" es vital. Aunque la API actual no admite la recuperación exacta de un contexto de sesión desde donde se interrumpió (el estado es temporal), su cliente debe almacenar el historial de la conversación (los intercambios) y reintroducirlos como client_contentcontext al reconectar.

3. Integración de video

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

• Límite de velocidad: No envíe videos 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 aleja de cadenas inestables y de alta latencia de modelos aislados hacia motores unificados de razonamiento multimodales. Para los equipos de ingeniería, el desafío cambia de administrar la latencia del flujo a administrar el estado y la concurrencia de WebSocket.

Si su organización está buscando crear agentes personalizados que aprovechen estas capacidades en tiempo real para flujos de trabajo complejos, asociarse con un experto en servicios de ingeniería de inteligencia artificial para empresas puede acelerar su proceso de implementació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