Agentes multimodales en tiempo real utilizando Gemini y WebSockets

La era de los chatbots "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 interactuar con un LLM; sino diseñar una arquitectura de baja latencia y 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 Streaming Bidireccional (Bidi) a través de WebSockets, que permiten interacciones por 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 y 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 vídeo) 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 vídeo y los transmite a través de WebSocket.
2. Capa de Orquestación (Backend): Un servicio de Python (por ejemplo, FastAPI) que valida la sesión, gestiona el estado y redirige el flujo a la API 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 cliente-API son posibles, pero se desaconsejan para casos de uso empresarial debido a los riesgos de seguridad (exposición de claves API) y 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. Inicializar el cliente de Gemini

Inicializamos el cliente con el modelo Gemini 1.5 Flash. Se recomienda utilizar el modelo Flash para agentes conversacionales debido a su significativamente menor tiempo de respuesta inicial (TTFT) en comparación con la versión 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 de 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 asyncio de Python para gestionar el flujo bidireccional sin bloquear.asíncronopara gestionar el flujo bidireccional sin generar bloqueos.

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. Gestión 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 pequeños fragmentos (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 llamada a funciones

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

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 de 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 de rendimiento para directores técnicos (CTOs)

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

1. Tiempo para el Primer Token (TTFT): Para las interacciones por voz, el TTFT debe ser inferior a 500 ms para que se sienta natural. Utilice variantes de Gemini Flash y asegúrese de que su infraestructura 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 el contexto lleno de datos de audio crudos aumenta los costos y la latencia. Implemente una estrategia de ventana deslizante o resuma las conversaciones anteriores.
3. Medidas de Seguridad: Los agentes en tiempo real pueden generar contenido incorrecto o inseguro. Configure siempre 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. Al aprovechar 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 flujos de datos en tiempo real, somos su socio para lograr la excelencia en ingeniería.

Preguntas frecuentes