Cómo iniciar y administrar sesiones en vivo

La API de Gemini Live permite interacciones de voz y texto con baja latencia, ya que procesa flujos continuos de audio o texto llamados sesiones para brindar respuestas habladas inmediatas y similares a las humanas. El desarrollador controla la administración del ciclo de vida de la sesión, desde el handshake inicial hasta la finalización correcta.

En esta página, se muestra cómo iniciar una sesión de conversación con los modelos de Gemini usando la API de Gemini Live. Puedes iniciar una sesión con Vertex AI Studio, el SDK de IA generativa de Google o WebSockets.

En esta página, también se muestra cómo hacer lo siguiente:

• Cómo extender una sesión más allá del límite de tiempo predeterminado
• Cómo reanudar una sesión anterior
• Actualiza las instrucciones del sistema durante una sesión
• Configura la ventana de contexto de una sesión
• Cómo habilitar la transcripción para una sesión
• Encripta los datos de la sesión con una clave de encriptación administrada por el cliente (CMEK)

Ciclo de vida de la sesión

Sin la compresión de la ventana de contexto, las sesiones solo de audio se limitan a 15 minutos y las sesiones de audio y video, a 2 minutos debido a los límites de tokens. Si superas estos límites, se finalizará la sesión, pero puedes usar la compresión de la ventana de contexto para extender las sesiones por un tiempo ilimitado.

La vida útil de una conexión se limita a unos 10 minutos debido a las restricciones de conexión de WebSocket. Cuando se termina la conexión, también se termina la sesión. En este caso, puedes configurar una sola sesión para que permanezca activa en varias conexiones con la reanudación de sesión. También recibirás un mensaje de GoAway antes de que finalice la conexión, lo que te permitirá tomar más medidas.

Cantidad máxima de sesiones simultáneas

Puedes tener hasta 1,000 sesiones simultáneas por proyecto en un plan de pago por uso (PayGo). Este límite no se aplica a los clientes que usan la Capacidad de procesamiento aprovisionada.

Inicia una sesión

En las siguientes pestañas, se muestra cómo iniciar una sesión de conversación en vivo con Vertex AI Studio, el SDK de IA generativa de Google o WebSockets:

Console

1. Abre Vertex AI Studio > Stream realtime.
2. Haz clic en Iniciar sesión para iniciar la conversación.

Para finalizar la sesión, haz clic en Detener sesión.

Python

Antes de comenzar, debes autenticarte en Gemini Enterprise Agent Platform con una clave de API o credenciales predeterminadas de la aplicación (ADC):

gcloud auth application-default login
      

Para obtener más información sobre cómo configurar la autenticación, consulta nuestra guía de inicio rápido.

import asyncio
from google import genai

# Replace the PROJECT_ID and LOCATION with your Project ID and location.
client = genai.Client(vertexai=True, project="PROJECT_ID", location="LOCATION")

# Configuration
MODEL = "gemini-3.8-live"
config = {
   "response_modalities": ["audio"],
}

async def main():
   # Establish WebSocket session
   async with client.aio.live.connect(model=MODEL, config=config) as session:
       print("Session established. Sending audio...")

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

Python

Cuando se usan WebSockets, la conexión se establece con un protocolo de enlace de WebSocket estándar. El extremo es regional y usa tokens de portador de OAuth 2.0 para la autenticación. En este caso, el token de autenticación suele pasarse en los encabezados de WebSocket (como Authorization: Bearer [TOKEN]).

import asyncio
import websockets

# Replace the PROJECT_ID and LOCATION with your Project ID and location.
PROJECT_ID = "PROJECT_ID"
LOCATION = "LOCATION"

# Authentication
token_list = !gcloud auth application-default print-access-token
ACCESS_TOKEN = token_list[0]

# Configuration
MODEL_ID = "gemini-3.8-live"
MODEL = f"projects/{PROJECT_ID}/locations/{LOCATION}/publishers/google/models/{MODEL_ID}"
config = {
   "response_modalities": ["audio"],
}

# Construct the WSS URL
HOST = f"{LOCATION}-aiplatform.googleapis.com"
URI = f"wss://{HOST}/ws/google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent"

async def main():
   headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}

   async with websockets.connect(URI, additional_headers=headers) as ws:
       print("Session established.")

       # Send Setup (Handshake)
       await ws.send(json.dumps({
           "setup": {
               "model": MODEL,
               "generation_config": config
           }
       }))
    # Send audio/video ...

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

Nota: Si bien algunos parámetros, incluidas las instrucciones del sistema, se pueden ajustar durante la sesión, el modelo y otros parámetros suelen ser inmutables una vez que se procesa el mensaje de configuración.

Cómo extender una sesión

Nota: La extensión de sesión solo está disponible cuando se usa el SDK de IA generativa de Google, no Vertex AI Studio.

La duración máxima predeterminada de una sesión de conversación es de 10 minutos, que puedes extender con notificaciones de goAway. Se envía una notificación de goAway (BidiGenerateContentServerMessage.goAway) al cliente 60 segundos antes de que finalice la sesión.

Para extender una sesión más allá del límite de conexión de 10 minutos, debes volver a conectarte con la reanudación de la sesión. Cuando recibes una notificación de goAway o cuando se finaliza la conexión por otros motivos, puedes iniciar una nueva conexión con un identificador de sesión que se obtuvo durante la sesión. De esta manera, se reanuda tu sesión con el contexto intacto en la nueva conexión. No hay límite para la cantidad de veces que puedes hacerlo. Para ver un ejemplo de cómo reanudar una sesión, consulta Cómo reanudar una sesión anterior.

En el siguiente ejemplo, se muestra cómo detectar la finalización inminente de una sesión escuchando una notificación de goAway:

Python

async for response in session.receive():
    if response.go_away is not None:
        # The connection will soon be terminated
        print(response.go_away.time_left)
      

Cómo reanudar una sesión anterior

Importante: Si necesitas garantizar que no se retengan datos en tu aplicación, no habilites la reanudación de la sesión.

La API de Gemini Live admite la reanudación de sesiones para evitar que el usuario pierda el contexto de la conversación durante una desconexión breve (por ejemplo, cuando cambia de Wi-Fi a 5G). Puedes reanudar una sesión anterior en un plazo de 24 horas. La reanudación de la sesión se logra almacenando datos en caché, incluidas las instrucciones de texto, audio y video, y los resultados del modelo. La privacidad a nivel del proyecto se aplica a estos datos almacenados en caché.

De forma predeterminada, la reanudación de sesión está inhabilitada. Para habilitar la reanudación de la sesión, establece el campo sessionResumption del mensaje BidiGenerateContentSetup. Si está habilitado, el servidor envía periódicamente mensajes SessionResumptionUpdate que contienen un session_id y un token de reanudación. Si se desconecta el WebSocket, el cliente puede volver a conectarse y, luego, incluir estas credenciales en el nuevo mensaje de configuración. Luego, el servidor restablece el contexto anterior, lo que permite que la conversación continúe sin problemas.

El período de reanudación es finito (por lo general, alrededor de 10 minutos). Si el cliente no se vuelve a conectar dentro de este período, se descarta el estado de la sesión para liberar recursos del servidor.

En el siguiente ejemplo, se muestra cómo conectarse al servicio, obtener un identificador de reanudación de sesión, simular una desconexión y, luego, volver a conectarse con el identificador para reanudar la sesión:

Python

import asyncio
from google import genai
from google.genai import types
import websockets

# Replace the PROJECT_ID and LOCATION with your Project ID and location.
client = genai.Client(vertexai=True, project="PROJECT_ID", location="LOCATION")

# Configuration
MODEL = "gemini-3.8-live"

async def resumable_session_example():
    """Demonstrates session resumption by connecting, disconnecting, and reconnecting."""
    session_handle = None

    print("Starting a new session...")
    try:
        async with client.aio.live.connect(
            model=MODEL,
            config=types.LiveConnectConfig(
                response_modalities=["audio"],
                session_resumption=types.SessionResumptionConfig(handle=None),
            ),
        ) as session:
            await session.send_content(
                content=types.Content(role="user", parts=[types.Part(text="Hello!")])
            )
            async for message in session.receive():
                if message.session_resumption_update:
                    update = message.session_resumption_update
                    if update.resumable and update.new_handle:
                        session_handle = update.new_handle
                        print(f"Received session handle: {session_handle}")
                        # For demonstration, we break to simulate a disconnect
                        # after receiving a handle.
                        break
                if message.server_content and message.server_content.turn_complete:
                    break
    except websockets.exceptions.WebSocketException as e:
        print(f"Initial connection failed: {e}")
        return

    if not session_handle:
        print("Did not receive a session handle. Cannot demonstrate resumption.")
        return

    print(f"\nSimulating disconnect and reconnecting with handle {session_handle}...")

    try:
        async with client.aio.live.connect(
            model=MODEL,
            config=types.LiveConnectConfig(
                response_modalities=["audio"],
                session_resumption=types.SessionResumptionConfig(handle=session_handle),
            ),
        ) as session:
            print("Successfully resumed session.")
            await session.send_content(
                content=types.Content(role="user", parts=[types.Part(text="I am back!")])
            )
            async for message in session.receive():
                if message.session_resumption_update:
                    update = message.session_resumption_update
                    if update.resumable and update.new_handle:
                        session_handle = update.new_handle
                        print(f"Received updated session handle: {session_handle}")
                if message.server_content:
                    print(f"Received server content: {message.server_content}")
                    if message.server_content.turn_complete:
                        break
            print("Resumed session finished.")
    except websockets.exceptions.WebSocketException as e:
        print(f"Failed to resume session: {e}")

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

Habilita la reanudación de sesión sin interrupciones con el modo sonido ambiente

Cuando habilitas la reanudación de la sesión, también puedes habilitar el modo transparente para que el proceso de reanudación sea más fluido para el usuario. Cuando se habilita el modo transparente, se devuelve explícitamente el índice del mensaje del cliente que corresponde a la instantánea del contexto. Esto ayuda a identificar qué mensaje del cliente debes volver a enviar cuando reanudas la sesión desde el identificador de reanudación.

Para habilitar el Modo sonido ambiente:

Python

config = {
   "response_modalities": ["audio"],
   "session_resumption_config": {
    "transparent": True,
   }
}
      

Encripta los datos de la sesión con CMEK

La reanudación de la sesión almacena los datos de la sesión en caché en reposo durante un máximo de 24 horas. Para proteger esos datos con tu propia clave en lugar de con unaGoogle-owned and Google-managed encryption key, usa un perfil de publicación. Un perfil de entrega vincula una clave de Cloud KMS a un proyecto, una ubicación y un alcance de API. Si existe un perfil de entrega, Agent Platform encripta los datos automáticamente. La configuración de tu sesión y el formato de la solicitud no cambian.

La CMEK del perfil de entrega de la API de Gemini Live es compatible con las multirregiones us y eu. No se admite la región global.

Para configurar la CMEK para la API de Gemini Live, haz lo siguiente:

1. Habilita la API de Cloud KMS y la API de Agent Platform en tu proyecto. Consulta Antes de comenzar.
2. Crea una clave de Cloud KMS en la multirregión us o eu. Consulta Crea un llavero de claves y una clave.
3. Otorga al agente de servicio de Gemini Enterprise Agent Platform el rol roles/cloudkms.cryptoKeyEncrypterDecrypter en la clave. Consulta Cómo otorgar permisos de Agent Platform.
4. Crea un perfil de entrega que establezca scope en GEMINI_LIVE y cmekConfig.encryptionSpec.kmsKeyName en tu clave. Consulta Cómo crear un perfil de entrega.

Nota: Los datos almacenados antes de que crees el perfil de publicación no se vuelven a encriptar de forma retroactiva con tu clave.

Actualiza las instrucciones del sistema durante una sesión

La API de Gemini Live te permite actualizar las instrucciones del sistema durante una sesión activa. Úsalo para adaptar las respuestas del modelo, por ejemplo, cambiar el idioma de la respuesta o modificar el tono.

Para actualizar las instrucciones del sistema durante la sesión, puedes enviar contenido de texto con el rol system. La instrucción del sistema actualizada seguirá vigente durante el resto de la sesión.

Python

session.send_client_content(
      content=types.Content(
          role="system", parts=[types.Part(text="new system instruction")]
      ),
      turn_complete=False
  )
      

Configura la ventana de contexto de la sesión

La ventana de contexto de la API de Gemini Live se usa para almacenar datos transmitidos en tiempo real (25 tokens por segundo [TPS] para audio y 258 TPS para video) y otro contenido, incluidas las entradas de texto y los resultados del modelo. Todos los modelos de la API de Gemini Live tienen un límite de ventana de contexto de 128,000 tokens.

En las sesiones de larga duración, a medida que avanza la conversación, se acumula el historial de tokens de audio y texto. Si este historial supera el límite del modelo, es posible que el modelo alucine, se ralentice o que la sesión se finalice de forma forzosa. Para habilitar sesiones más largas, puedes habilitar la compresión de la ventana de contexto configurando el campo contextWindowCompression como parte de la configuración de la sesión.

La compresión de la ventana de contexto usa una ventana deslizante del servidor para truncar los turnos más antiguos cuando está habilitada. Cuando los tokens acumulados superan una longitud máxima definida (establecida con el control deslizante Tamaño máximo del contenido en Vertex AI Studio o trigger_tokens en la API), el servidor poda automáticamente los turnos más antiguos o los resume para mantener el contexto dentro del límite. En ContextWindowCompressionConfig, puedes configurar un mecanismo de ventana deslizante y la cantidad de tokens definidos en el parámetro target_tokens que activa la compresión.

Esto permite duraciones de sesión teóricamente infinitas desde la perspectiva del usuario, ya que la "memoria" se administra constantemente. Sin compresión, las sesiones solo de audio podrían limitarse a aproximadamente 15 minutos antes de alcanzar los límites estrictos.

Las longitudes mínima y máxima para la longitud del contexto y el tamaño objetivo son las siguientes:

Parámetro de configuración (marca de API)	Valor mínimo	Valor máximo
Longitud máxima del contexto (trigger_tokens)	5,000	128,000
Tamaño del contexto objetivo (target_tokens)	0	128,000

Para establecer la ventana de contexto, haz lo siguiente:

Console

1. Abre Vertex AI Studio > Stream realtime.
2. Haz clic para abrir el menú Avanzado.
3. En la sección Contexto de la sesión, usa el control deslizante Tamaño máximo del contexto para establecer el tamaño del contexto en un valor entre 5,000 y 128,000.
4. (Opcional) En la misma sección, usa el control deslizante Tamaño del contexto objetivo para establecer el tamaño objetivo en un valor entre 0 y 128,000.

Python

Establece los campos context_window_compression.trigger_tokens y context_window_compression.sliding_window.target_tokens en el mensaje de configuración:

config = {
   "response_modalities": ["audio"],
   # Configures compression
   "context_window_compression" : {
    "trigger_tokens": 10000,
    "sliding_window": {"target_tokens" : 512}
   }
}
      

Habilita la transcripción de audio para la sesión

Puedes habilitar las transcripciones para el audio de entrada y salida.

Para recibir transcripciones, debes actualizar la configuración de la sesión. Debes agregar los objetos input_audio_transcription y output_audio_transcription, y asegurarte de que text se incluya en response_modalities.

Para mejorar la calidad de la transcripción del reconocimiento de voz automático (ASR) multilingüe, puedes proporcionar sugerencias de idioma con el campo language_codes dentro de input_audio_transcription o output_audio_transcription. Se recomienda proporcionar sugerencias para mejorar la calidad de la transcripción, ya que se reduce el riesgo de detección incorrecta del idioma, especialmente en el caso de las instrucciones breves. El campo language_codes acepta una lista de códigos de idioma BCP-47 (por ejemplo, "en-US", "es-US").

config = {
    "response_modalities": ["audio", "text"],
    "input_audio_transcription": {
        "language_codes": ["en-US"]
    },
    "output_audio_transcription": {},
}

Procesamiento de la respuesta

En la siguiente muestra de código, se muestra cómo conectarse con la sesión configurada y extraer las partes de texto (transcripciones) junto con los datos de audio.

# Receive Output Loop
async for message in session.receive():
    server_content = message.server_content
    if server_content:
        # Handle Model Turns (Audio + Text)
        model_turn = server_content.model_turn
        if model_turn and model_turn.parts:
            for part in model_turn.parts:
                # Handle Text (Transcriptions)
                if part.text:
                    print(f"Transcription: {part.text}")
                # Handle Audio
                if part.inline_data:
                    audio_data = part.inline_data.data
                    # Process audio bytes...
                    pass

        # Check for turn completion
        if server_content.turn_complete:
            print("Turn complete.")

¿Qué sigue?

• Envía transmisiones de audio y video
• Prácticas recomendadas para la API de Gemini Live
• Diseña instrucciones multimodales
• Introducción a las llamadas a funciones