Conversaciones interactivas con la API Live

La API Live permite interacciones bidireccionales de voz y vídeo con Gemini con baja latencia.

En esta guía se explica cómo configurar una conversación interactiva bidireccional, ajustar la configuración de audio, gestionar sesiones y más.

Modelos admitidos

Puedes usar la API Live con los siguientes modelos:

Versión del modelo Nivel de disponibilidad
gemini-live-2.5-flash GA privada*
gemini-live-2.5-flash-preview-native-audio Vista previa pública

* Ponte en contacto con el representante del equipo de tu cuenta de Google para solicitar acceso.

Iniciar una conversación

Consola

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

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

Python

Para habilitar conversaciones interactivas en tiempo real, ejecuta el siguiente ejemplo en un ordenador local con acceso al micrófono y al altavoz (no en un cuaderno de Colab). En este ejemplo se configura una conversación con la API, lo que te permite enviar peticiones de audio y recibir respuestas de audio: 

"""
# Installation
# on linux
sudo apt-get install portaudio19-dev

# on mac
brew install portaudio

python3 -m venv env
source env/bin/activate
pip install google-genai
"""

import asyncio
import pyaudio
from google import genai
from google.genai import types

CHUNK=4200
FORMAT=pyaudio.paInt16
CHANNELS=1
RECORD_SECONDS=5
MODEL = 'gemini-live-2.5-flash'
INPUT_RATE=16000
OUTPUT_RATE=24000

client = genai.Client(
    vertexai=True,
    project=GOOGLE_CLOUD_PROJECT,
    location=GOOGLE_CLOUD_LOCATION,
)
config = {
    "response_modalities": ["AUDIO"],
    "input_audio_transcription": {},  # Configure input transcription
    "output_audio_transcription": {},  # Configure output transcription
}

async def main():
    print(MODEL)
    p = pyaudio.PyAudio()
    async with client.aio.live.connect(model=MODEL, config=config) as session:
        #exit()
        async def send():
            stream = p.open(
                format=FORMAT, channels=CHANNELS, rate=INPUT_RATE, input=True, frames_per_buffer=CHUNK)
            while True:
                frame = stream.read(CHUNK)
                await session.send(input={"data": frame, "mime_type": "audio/pcm"})
                await asyncio.sleep(10**-12)
        async def receive():
            output_stream = p.open(
                format=FORMAT, channels=CHANNELS, rate=OUTPUT_RATE, output=True, frames_per_buffer=CHUNK)
            async for message in session.receive():
                if message.server_content.input_transcription:
                  print(message.server_content.model_dump(mode="json", exclude_none=True))
                if message.server_content.output_transcription:
                  print(message.server_content.model_dump(mode="json", exclude_none=True))
                if message.server_content.model_turn:
                    for part in message.server_content.model_turn.parts:
                        if part.inline_data.data:
                            audio_data=part.inline_data.data
                            output_stream.write(audio_data)
                            await asyncio.sleep(10**-12)
        send_task = asyncio.create_task(send())
        receive_task = asyncio.create_task(receive())
        await asyncio.gather(send_task, receive_task)

asyncio.run(main())

Python

Configura una conversación con la API que te permita enviar peticiones de texto y recibir respuestas de audio:

# Set model generation_config
CONFIG = {"response_modalities": ["AUDIO"]}

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {bearer_token[0]}",
}

async def main() -> None:
    # Connect to the server
    async with connect(SERVICE_URL, additional_headers=headers) as ws:

        # Setup the session
        async def setup() -> None:
            await ws.send(
                json.dumps(
                    {
                        "setup": {
                            "model": "gemini-live-2.5-flash",
                            "generation_config": CONFIG,
                        }
                    }
                )
            )

            # Receive setup response
            raw_response = await ws.recv(decode=False)
            setup_response = json.loads(raw_response.decode("ascii"))
            print(f"Connected: {setup_response}")
            return

        # Send text message
        async def send() -> bool:
            text_input = input("Input > ")
            if text_input.lower() in ("q", "quit", "exit"):
                return False

            msg = {
                "client_content": {
                    "turns": [{"role": "user", "parts": [{"text": text_input}]}],
                    "turn_complete": True,
                }
            }

            await ws.send(json.dumps(msg))
            return True

        # Receive server response
        async def receive() -> None:
            responses = []

            # Receive chucks of server response
            async for raw_response in ws:
                response = json.loads(raw_response.decode())
                server_content = response.pop("serverContent", None)
                if server_content is None:
                    break

                model_turn = server_content.pop("modelTurn", None)
                if model_turn is not None:
                    parts = model_turn.pop("parts", None)
                    if parts is not None:
                        for part in parts:
                            pcm_data = base64.b64decode(part["inlineData"]["data"])
                            responses.append(np.frombuffer(pcm_data, dtype=np.int16))

                # End of turn
                turn_complete = server_content.pop("turnComplete", None)
                if turn_complete:
                    break

            # Play the returned audio message
            display(Markdown("**Response >**"))
            display(Audio(np.concatenate(responses), rate=24000, autoplay=True))
            return

        await setup()

        while True:
            if not await send():
                break
            await receive()

Inicia la conversación, introduce tus peticiones o escribe q, quit o exit para salir.

await main()

Cambiar los ajustes de idioma y voz

La API Live usa Chirp 3 para admitir respuestas de voz sintetizadas en una variedad de voces e idiomas en HD. Para ver una lista completa y demostraciones de cada voz, consulta Chirp 3: voces en HD.

Para definir la voz y el idioma de la respuesta, haz lo siguiente:

Consola

  1. Abre Vertex AI Studio > Stream realtime (Vertex AI Studio > Transcribir en tiempo real).
  2. En el desplegable Salidas, selecciona una voz en el menú desplegable Voz.
  3. En el mismo desplegable, selecciona un idioma en el menú desplegable Idioma.
  4. Haz clic en Iniciar sesión para iniciarla.

Python

config = LiveConnectConfig(
    response_modalities=["AUDIO"],
    speech_config=SpeechConfig(
        voice_config=VoiceConfig(
            prebuilt_voice_config=PrebuiltVoiceConfig(
                voice_name=voice_name,
            )
        ),
        language_code="en-US",
    ),
)

Cambiar los ajustes de detección de actividad de voz

La detección de actividad de voz (VAD) permite que el modelo reconozca cuándo está hablando una persona. Esto es esencial para crear conversaciones naturales, ya que permite que el usuario interrumpa el modelo en cualquier momento.

El modelo realiza automáticamente la detección de actividad de voz (VAD) en un flujo de entrada de audio continuo. Puedes configurar los ajustes de VAD mediante el campo realtimeInputConfig.automaticActivityDetection del mensaje de configuración. Cuando la detección de actividad de voz detecta una interrupción, la generación en curso se cancela y se descarta. En el historial de la sesión solo se conserva la información que ya se ha enviado al cliente. A continuación, el servidor envía un mensaje para informar de la interrupción.

Si el flujo de audio se pausa durante más de un segundo (por ejemplo, si el usuario desactiva el micrófono), envía un evento audioStreamEnd para vaciar el audio almacenado en caché. El cliente puede reanudar el envío de datos de audio en cualquier momento.

También puedes inhabilitar la detección automática de voz configurando realtimeInputConfig.automaticActivityDetection.disabled en true en el mensaje de configuración. Con esta configuración, el cliente detecta la voz del usuario y envía mensajes activityStart y activityEnd en los momentos oportunos. No se envía un audioStreamEnd. Las interrupciones se marcan con activityEnd.

Python

config = LiveConnectConfig(
    response_modalities=["TEXT"],
    realtime_input_config=RealtimeInputConfig(
        automatic_activity_detection=AutomaticActivityDetection(
            disabled=False,  # default
            start_of_speech_sensitivity=StartSensitivity.START_SENSITIVITY_LOW, # Either START_SENSITIVITY_LOW or START_SENSITIVITY_HIGH
            end_of_speech_sensitivity=EndSensitivity.END_SENSITIVITY_LOW, # Either END_SENSITIVITY_LOW or END_SENSITIVITY_HIGH
            prefix_padding_ms=20,
            silence_duration_ms=100,
        )
    ),
)

async with client.aio.live.connect(
    model=MODEL_ID,
    config=config,
) as session:
    audio_bytes = Path("sample.pcm").read_bytes()

    await session.send_realtime_input(
        media=Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000")
    )

    # if stream gets paused, send:
    # await session.send_realtime_input(audio_stream_end=True)

    response = []
    async for message in session.receive():
        if message.server_content.interrupted is True:
            # The model generation was interrupted
            response.append("The session was interrupted")

        if message.text:
            response.append(message.text)

    display(Markdown(f"**Response >** {''.join(response)}"))

Prolongar una sesión

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

Puedes ampliar la duración de la sesión en incrementos de 10 minutos con el SDK de IA generativa. Puedes ampliar una sesión tantas veces como quieras. Para ver un ejemplo, consulta Habilitar e inhabilitar la reanudación de la sesión.

Ventana de contexto

La ventana de contexto de la API Live se usa para almacenar datos transmitidos en tiempo real (25 tokens por segundo [TPS] para audio y 258 TPS para vídeo) y otro contenido, incluidas las entradas de texto y las salidas del modelo.

Si la ventana de contexto supera la longitud máxima (que se define con el control deslizante Tamaño máximo del contenido en Vertex AI Studio o con trigger_tokens en la API), se truncan las conversaciones más antiguas mediante la compresión de la ventana de contexto para evitar que la sesión finalice de forma abrupta. La compresión de la ventana de contexto se activa cuando la ventana de contexto alcanza su longitud máxima (que se puede definir en Vertex AI Studio con el control deslizante Tamaño de contexto objetivo o con target_tokens en la API) y elimina las partes más antiguas de la conversación hasta que el número total de tokens vuelva a ser igual al tamaño objetivo.

Por ejemplo, si la longitud máxima del contexto es de 32000 tokens y el tamaño objetivo es de 16000 tokens:

  1. Turno 1: empieza la conversación. En este ejemplo, la solicitud usa 12.000 tokens.
    • Tamaño total del contexto: 12.000 tokens
  2. Turno 2: haces otra solicitud. Esta solicitud usa otros 12.000 tokens.
    • Tamaño total del contexto: 24.000 tokens
  3. Turno 3: haces otra solicitud. Esta solicitud usa 14.000 tokens.
    • Tamaño total del contexto: 38.000 tokens

Como el tamaño total del contexto ahora es superior al máximo de 32.000 tokens, se activa la compresión de la ventana de contexto. El sistema vuelve al principio de la conversación y empieza a eliminar turnos antiguos hasta que el tamaño total de los tokens sea inferior al objetivo de 16.000 tokens:

  • Se elimina Turn 1 (12.000 tokens). Ahora el total es de 26.000 tokens, que sigue siendo superior al objetivo de 16.000 tokens.
  • Elimina Turn 2 (12.000 tokens). Ahora el total es de 14.000 tokens.

El resultado final es que solo Turn 3 permanece en la memoria activa y la conversación continúa a partir de ese punto.

Las longitudes mínimas y máximas de la longitud del contexto y del tamaño de destino son las siguientes:

Ajuste (marca de API) Valor mínimo Valor máximo
Longitud máxima del contexto (trigger_tokens) 5000 128.000
Tamaño del contexto objetivo (target_tokens) 0 128.000

Para definir la ventana de contexto, sigue estos pasos:

Consola

  1. Abre Vertex AI Studio > Stream realtime (Vertex AI Studio > Transcribir en tiempo real).
  2. Haz clic para abrir el menú Avanzado.
  3. En la sección Contexto de la sesión, use el control deslizante Tamaño máximo del contexto para definir un valor entre 5000 y 128.000.
  4. (Opcional) En la misma sección, use el control deslizante Tamaño del contexto de destino para definir el tamaño de destino en un valor entre 0 y 128.000.

Python

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

config = types.LiveConnectConfig(
      temperature=0.7,
      response_modalities=['TEXT'],
      system_instruction=types.Content(
          parts=[types.Part(text='test instruction')], role='user'
      ),
      context_window_compression=types.ContextWindowCompressionConfig(
          trigger_tokens=1000,
          sliding_window=types.SlidingWindow(target_tokens=10),
      ),
  )

Sesiones simultáneas

Puedes tener hasta 5000 sesiones simultáneas por proyecto.

Actualizar las instrucciones del sistema durante una sesión

La API Live te permite actualizar las instrucciones del sistema durante una sesión activa. Úsalo para adaptar las respuestas del modelo, como cambiar el idioma 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á en vigor durante el resto de la sesión.

Python

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

Habilitar e inhabilitar la reanudación de sesiones

La reanudación de la sesión te permite volver a conectarte a una sesión anterior en un plazo de 24 horas. Para ello, se almacenan datos en caché, como texto, vídeo, peticiones de audio y resultados de modelos. La privacidad a nivel de proyecto se aplica a estos datos almacenados en caché.

De forma predeterminada, la reanudación de la sesión está inhabilitada.

Para habilitar la función de reanudación de sesiones, define el campo sessionResumption del mensaje BidiGenerateContentSetup. Si se habilita, el servidor hará periódicamente una instantánea de los contextos de sesión almacenados en caché y la guardará en el almacenamiento interno.

Cuando se crea una captura correctamente, se devuelve un resumptionUpdate con el ID de controlador que puedes registrar y usar más adelante para reanudar la sesión a partir de la captura.

A continuación, se muestra un ejemplo de cómo habilitar la reanudación de la sesión y obtener el ID de controlador:

Python

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

client = genai.Client(
    vertexai=True,
    project=GOOGLE_CLOUD_PROJECT,
    location=GOOGLE_CLOUD_LOCATION,
)
model = "gemini-live-2.5-flash"

async def main():
    print(f"Connecting to the service with handle {previous_session_handle}...")
    async with client.aio.live.connect(
        model=model,
        config=types.LiveConnectConfig(
            response_modalities=["AUDIO"],
            session_resumption=types.SessionResumptionConfig(
                # The handle of the session to resume is passed here,
                # or else None to start a new session.
                handle=previous_session_handle
            ),
        ),
    ) as session:
        while True:
            await session.send_client_content(
                turns=types.Content(
                    role="user", parts=[types.Part(text="Hello world!")]
                )
            )
            async for message in session.receive():
                # Periodically, the server will send update messages that may
                # contain a handle for the current state of the session.
                if message.session_resumption_update:
                    update = message.session_resumption_update
                    if update.resumable and update.new_handle:
                        # The handle should be retained and linked to the session.
                        return update.new_handle

                # For the purposes of this example, placeholder input is continually fed
                # to the model. In non-sample code, the model inputs would come from
                # the user.
                if message.server_content and message.server_content.turn_complete:
                    break

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

Para que la sesión se reanude sin problemas, habilita el modo Transparencia:

Python

types.LiveConnectConfig(
            response_modalities=["AUDIO"],
            session_resumption=types.SessionResumptionConfig(
                transparent=True,
    ),
)

Una vez que se ha habilitado el modo transparente, se devuelve explícitamente el índice del mensaje del cliente que corresponde a la instantánea del contexto. Esto te ayuda a identificar qué mensaje del cliente debes volver a enviar cuando reanudes la sesión desde el identificador de reanudación.

Más información

Para obtener más información sobre cómo usar la API Live, consulta los siguientes artículos: