Diseño de agentes multimodales en tiempo real utilizando Gemini y WebSockets

La era de los chatbots "de entrada y salida de texto" está desapareciendo rápidamente. Las aplicaciones empresariales modernas exigen agentes "en tiempo real"—sistemas inteligentes capaces de percibir y responder a audio, vídeo y texto en tiempo real. Para un director de tecnología (CTO) o ingeniero de software senior, el desafío no es simplemente solicitar un LLM; sino diseñar una arquitectura de baja latencia y de estado que gestione eficazmente los flujos multimodales.

En este artículo, desglosaremos la arquitectura necesaria para construir un agente conversacional multimodal en tiempo real utilizando los modelos Gemini 1.5 Pro/Flash de Google y Python. Nos centraremos específicamente en las capacidades de "Bidirectional (Bidi) Streaming" a través de WebSockets, que permiten interacciones de voz humanas, interrumpibles.Bidirectional (Bidi) Streaming capacidades a través de WebSockets, que permiten interacciones de voz humanas, interrumpibles.

Como socio en servicios de ingeniería de IA para empresas, 4Geeks ayuda con frecuencia a las organizaciones a migrar de modelos estáticos de solicitud-respuesta a estas arquitecturas dinámicas basadas en sesiones.

El Cambio Arquitectónico: De REST a WebSockets

La integración tradicional de LLM se basa en solicitudes HTTP sin estado. Sin embargo, las conversaciones multimodales verdaderas (como asistentes de voz o agentes de análisis de video) requieren una conexión persistente para manejar flujos de datos continuos.

La arquitectura generalmente sigue este patrón:

1. Capa del Cliente: Captura audio (PCM) o fotogramas de video y los transmite a través de una WebSocket.
2. Capa de Orquestación (Backend): Un servicio de Python (por ejemplo, FastAPI) que valida la sesión, gestiona el estado y enruta el flujo a la API de Gemini.
3. Capa del Modelo: Procesamiento de Gemini 1.5 Flash (optimizado para baja latencia) del flujo multimodal y generación de tokens en tiempo real.

Implementación Técnica

Utilizaremos el SDK google-genai moderno (el SDK Python V2) y FastAPI para crear un proxy de backend seguro. Las conexiones directas entre cliente y API son posibles, pero se desaconsejan para casos de uso empresariales debido a los riesgos de seguridad (exposición de claves de API) y a la falta de control sobre el contexto de la sesión.

1. Requisitos y configuración

Asegúrese de tener la última versión del SDK instalada. En un entorno de producción, administrará las dependencias a través de poetry o requirements.txt.

pip install -q -U google-genai fastapi uvicorn websockets

2. Inicializando el cliente Gemini

Inicializamos el cliente con el modelo Gemini 1.5 Flash. Se recomienda utilizar el modelo Flash para agentes conversacionales debido a su tiempo de respuesta del primer token (TTFT) significativamente más bajo en comparación con Pro.

import os
from google import genai
from google.genai import types

# Initialize the client
# Ensure GEMINI_API_KEY is set in your environment variables
client = genai.Client(api_key=os.environ.get("GEMINI_API_KEY"))

# Configuration for the session
config = types.GenerateContentConfig(
    temperature=0.7,
    max_output_tokens=2048,
    response_modalities=["TEXT"] # or ["AUDIO"] for voice-to-voice
)

3. Construir el proxy WebSocket

Este es el núcleo del desafío de ingeniería. Debemos crear un punto final de WebSocket en FastAPI que acepte flujos de audio de clientes, los envíe a Gemini y transmita la respuesta de vuelta al cliente.

Utilizamos la biblioteca asyncio de Python para gestionar el flujo bidireccional sin bloquear.

import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from google.genai import types

app = FastAPI()

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    
    # Establish a live session with Gemini
    async with client.aio.live.connect(model="gemini-2.0-flash-exp", config=config) as session:
        try:
            # Create tasks to handle sending and receiving concurrently
            receive_task = asyncio.create_task(handle_client_input(websocket, session))
            send_task = asyncio.create_task(handle_model_output(websocket, session))
            
            # Wait for either to finish (or error)
            done, pending = await asyncio.wait(
                [receive_task, send_task],
                return_when=asyncio.FIRST_COMPLETED,
            )
            
            for task in pending:
                task.cancel()
                
        except Exception as e:
            print(f"Session error: {e}")
        finally:
            await websocket.close()

async def handle_client_input(websocket: WebSocket, session):
    """Receives audio/text from client and pushes to Gemini."""
    try:
        while True:
            # Expecting JSON with base64 encoded chunks or text
            data = await websocket.receive_json()
            
            if "text" in data:
                await session.send(data["text"], end_of_turn=True)
            elif "audio_chunk" in data:
                # data['audio_chunk'] should be base64 encoded PCM data
                # Gemini expects raw bytes or specific blob types
                await session.send(
                     {"mime_type": "audio/pcm", "data": data["audio_chunk"]}, 
                     end_of_turn=False
                )
    except WebSocketDisconnect:
        print("Client disconnected")

async def handle_model_output(websocket: WebSocket, session):
    """Receives stream from Gemini and pushes to client."""
    async for response in session.receive():
        # Gemini sends text chunks or audio bytes depending on config
        if response.text:
            await websocket.send_json({"text": response.text})
        elif response.data:
             # Handle audio bytes output if voice-to-voice is enabled
            await websocket.send_bytes(response.data)

4. Manejo de formatos de audio y latencia

Una común dificultad en la integración multimodales es el formato de audio. La API Gemini Live generalmente espera:

• Formato: PCM lineal (16 bits, little-endian)
• Tasa de muestreo: 16 kHz o 24 kHz (la consistencia es clave)
• Tamaño del fragmento: Enviar audio en fragmentos pequeños (por ejemplo, duración de 20 ms - 40 ms) para minimizar la latencia del búfer. Enviar fragmentos de 1 segundo resultará en una notable demora para el usuario.

Si su aplicación cliente (React Native, Flutter o Swift) captura audio en un formato diferente (como AAC o Opus), debe transcodificarlo a PCM antes de enviarlo al backend o utilizar una biblioteca como ffmpeg en el lado del servidor, aunque esto introduce latencia.

Patrón Avanzado: Uso de Herramientas y Llamadas a Funciones

Para que el agente sea verdaderamente útil para "flujos de trabajo empresariales", necesita interactuar con sus APIs internas (por ejemplo, verificar el inventario, programar citas). Gemini admite la llamada dinámica a funciones dentro de la sesión de transmisión.

Usted define las herramientas en la configuración inicial:

def get_inventory_status(item_id: str):
    # Mock database lookupPython
    return {"item_id": item_id, "status": "in_stock", "quantity": 150}

tools = [get_inventory_status]

# Update config to include tools
config = types.GenerateContentConfig(
    tools=tools,
    response_modalities=["TEXT"]
)

Cuando el modelo decide llamar a una función, pausa la generación y envía una señal function_call . Tu backend debe ejecutar la función de Python y devolver la function_response al contexto de la sesión, permitiendo que el modelo genere la respuesta final en lenguaje natural.

Consideraciones sobre el rendimiento para directores de tecnología (CTOs)

Al implementar estos [servicios de ingeniería de IA para empresas], considere las siguientes métricas:

1. Tiempo al Primer Token (TTFT): Para interacciones por voz, el TTFT debe ser inferior a 500 ms para que sea natural. Utilice variantes de Gemini Flash y asegúrese de que su infraestructura de WebSocket (por ejemplo, AWS API Gateway + Lambda o Kubernetes) esté optimizada para conexiones persistentes.
2. Gestión de la Ventana de Contexto: Si bien Gemini 1.5 admite hasta 1-2 millones de tokens, mantener la ventana de contexto llena de datos de audio sin procesar aumenta el coste y la latencia. Implemente una estrategia de ventana deslizante o resuma las interacciones anteriores en el historial de la conversación.
3. Medidas de Seguridad: Los agentes en tiempo real pueden generar contenido no deseado o inseguro. Siempre configure las safety_settings en el GenerateContentConfig para BLOCK_LOW_AND_ABOVE para aplicaciones orientadas a empresas.

Conclusión

Integrar Gemini para conversaciones multimodales en tiempo real requiere un cambio de los paradigmas REST sin estado a arquitecturas basadas en eventos y flujos de datos. Utilizando WebSockets y las capacidades de baja latencia de Gemini 1.5 Flash, los equipos de ingeniería pueden crear agentes que no solo "leen" texto, sino que también "escuchan" y "observan" junto con el usuario.

En 4Geeks, nos especializamos en la creación de estas arquitecturas de alto rendimiento. Ya sea que necesite implementar agentes de IA personalizados o escalar su infraestructura en la nube para manejar el streaming en tiempo real, somos su socio para la excelencia en la ingeniería.

Preguntas frecuentes