Iniciar e gerenciar sessões ao vivo

A API Gemini Live permite interações de voz e texto de baixa latência processando streams contínuos de áudio ou texto chamados sessões para oferecer respostas faladas imediatas e semelhantes às humanas. O gerenciamento do ciclo de vida da sessão, desde o handshake inicial até o encerramento normal, é controlado pelo desenvolvedor.

Esta página mostra como iniciar uma sessão de conversa com modelos do Gemini usando a API Gemini Live. É possível iniciar uma sessão usando o Vertex AI Studio, o SDK de IA Generativa do Google ou WebSockets.

Esta página também mostra como fazer o seguinte:

  • Estender uma sessão além do limite de tempo padrão
  • Retomar uma sessão anterior
  • Atualizar as instruções do sistema durante uma sessão
  • Configurar a janela de contexto de uma sessão
  • Ativar a transcrição de uma sessão

Tempo de vida da sessão

Sem a compactação da janela de contexto, as sessões somente de áudio são limitadas a 15 minutos, e as sessões de áudio e vídeo são limitadas a 2 minutos devido aos limites de token. Se esses limites forem excedidos, a sessão será encerrada, mas você poderá usar a compactação da janela de contexto para estender as sessões por um período ilimitado.

O tempo de vida de uma conexão é limitado a cerca de 10 minutos devido a restrições de conexão WebSocket. Quando a conexão é encerrada, a sessão também é. Nesse caso, é possível configurar uma única sessão para permanecer ativa em várias conexões usando a retomada da sessão. Você também vai receber uma mensagem GoAway antes do término da conexão, permitindo que você tome outras ações.

Número máximo de sessões simultâneas

É possível ter até 1.000 sessões simultâneas por projeto em um plano de pagamento por uso (PayGo). Esse limite não se aplica a clientes que usam a Capacidade de Processamento Provisionada.

Iniciar uma sessão

As guias a seguir mostram como iniciar uma sessão de conversa ao vivo usando o Vertex AI Studio, o SDK de IA Generativa do Google ou WebSockets:

Console

  1. Abra Vertex AI Studio > Stream realtime.
  2. Clique em Start session para iniciar a conversa.

Para encerrar a sessão, clique em Stop session.

Python

Antes de começar, é necessário fazer a autenticação na Gemini Enterprise Agent Platform usando uma chave de API ou credenciais padrão do aplicativo (ADC): 

gcloud auth application-default login

Para mais informações sobre como configurar a autenticação, consulte nosso início 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-live-2.5-flash-native-audio"
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

Ao usar WebSockets, a conexão é estabelecida com um handshake WebSocket padrão. O endpoint é regional e usa tokens do portador OAuth 2.0 para autenticação. Nesse cenário, o token de autenticação geralmente é transmitido nos cabeçalhos do 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-live-2.5-flash-native-audio"
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())

Estender uma sessão

A duração máxima padrão de uma sessão de conversa é de 10 minutos. Uma goAway notificação (BidiGenerateContentServerMessage.goAway) é enviada ao cliente 60 segundos antes do término da sessão.

Para estender uma sessão além do limite de conexão de 10 minutos, é necessário se reconectar usando a retomada da sessão. Quando você recebe uma notificação goAway ou quando a conexão é encerrada por outros motivos, é possível iniciar uma nova conexão usando um identificador de sessão obtido durante a sessão. Isso retoma a sessão com o contexto intacto na nova conexão. Não há limite para o número de vezes que você pode fazer isso. Para conferir um exemplo de retomada de uma sessão, consulte Retomar uma sessão anterior.

O exemplo a seguir mostra como detectar um encerramento de sessão iminente ouvindo uma notificação 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)

Retomar uma sessão anterior

A API Gemini Live oferece suporte à retomada da sessão para evitar que o usuário perca o contexto da conversa durante uma breve desconexão (por exemplo, ao alternar do Wi-Fi para o 5G). É possível retomar uma sessão anterior em até 24 horas. A retomada da sessão é feita armazenando dados em cache, incluindo texto, vídeo, comandos de áudio e saídas do modelo. A privacidade para envolvidos no projeto é aplicada a esses dados armazenados em cache.

Por padrão, a retomada da sessão está desativada. Para ativar a retomada da sessão, defina o campo sessionResumption da BidiGenerateContentSetup mensagem. Se ativado, o servidor envia periodicamente mensagens SessionResumptionUpdate contendo um session_id e um token de retomada. Se o WebSocket for desconectado, o cliente poderá se reconectar e incluir essas credenciais na nova mensagem de configuração. Em seguida, o servidor restaura o contexto anterior, permitindo que a conversa continue sem problemas.

A janela de retomada é finita (geralmente cerca de 10 minutos). Se o cliente não se reconectar dentro desse período, o estado da sessão será descartado para liberar recursos do servidor.

O exemplo a seguir se conecta ao serviço, recebe um identificador de retomada de sessão, simula uma desconexão e se reconecta usando o identificador para retomar a sessão:

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-live-2.5-flash-native-audio"

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())

Ativar a retomada de sessão perfeita com o modo transparente

Ao ativar a retomada da sessão, também é possível ativar o modo transparente para ajudar a tornar o processo de retomada mais simples para o usuário. Quando o modo transparente está ativado, o índice da mensagem do cliente que corresponde ao snapshot de contexto é retornado explicitamente. Isso ajuda a identificar qual mensagem do cliente você precisa enviar novamente ao retomar a sessão do identificador de retomada.

Para ativar o modo transparente:

Python

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

Atualizar as instruções do sistema durante uma sessão

A API Gemini Live permite atualizar as instruções do sistema durante uma sessão ativa. Use isso para adaptar as respostas do modelo, como mudar o idioma da resposta ou modificar o tom.

Para atualizar as instruções do sistema no meio da sessão, envie conteúdo de texto com a função system. A instrução do sistema atualizada permanecerá em vigor para a sessão restante.

Python

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

Configurar a janela de contexto da sessão

A janela de contexto da API Gemini Live é usada para armazenar dados transmitidos em tempo real (25 tokens por segundo (TPS) para áudio e 258 TPS para vídeo) e outros conteúdos, incluindo entradas de texto e saídas do modelo. Todos os modelos da API Gemini Live têm um limite de janela de contexto de 128 mil tokens.

Em sessões longas, à medida que a conversa avança, o histórico de tokens de áudio e texto se acumula. Se esse histórico exceder o limite do modelo, ele poderá alucinar, ficar mais lento ou a sessão poderá ser encerrada à força. Para ativar sessões mais longas, é possível ativar a compactação da janela de contexto definindo o campo contextWindowCompression como parte da configuração da sessão.

A compactação da janela de contexto usa uma janela deslizante do lado do servidor para truncar as conversas mais antigas quando ativada. Quando os tokens acumulados excedem um comprimento máximo definido (definido usando o controle deslizante Tamanho máximo do conteúdo no Vertex AI Studio ou trigger_tokens na API), o servidor poda automaticamente as conversas mais antigas ou as resume para manter o contexto dentro do limite. Em ContextWindowCompressionConfig, é possível configurar um mecanismo de janela deslizante e o número de tokens definidos no parâmetro target_tokens que aciona a compactação.

Isso permite durações de sessão teoricamente infinitas do ponto de vista do usuário, já que a "memória" é gerenciada constantemente. Sem compactação, as sessões somente de áudio podem ser limitadas a aproximadamente 15 minutos antes de atingir limites rígidos.

Os comprimentos mínimo e máximo para o tamanho do contexto e o tamanho de destino são:

Configuração (flag da API) Valor mínimo Valor máximo
Tamanho máximo de contexto (trigger_tokens) 5.000 128.000
Tamanho do contexto de destino (target_tokens) 0 128.000

Para definir a janela de contexto:

Console

  1. Abra Vertex AI Studio > Stream realtime.
  2. Clique para abrir o menu Avançado.
  3. Na seção Contexto da sessão , use o controle deslizante Tamanho máximo do contexto para definir o tamanho do contexto como um valor entre 5.000 e 128.000.
  4. (Opcional) Na mesma seção, use o controle deslizante Tamanho do contexto de destino para definir o tamanho de destino como um valor entre 0 e 128.000.

Python

Defina os campos context_window_compression.trigger_tokens e context_window_compression.sliding_window.target_tokens na mensagem de configuração: 

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

Ativar a transcrição de áudio da sessão

É possível ativar transcrições para o áudio de entrada e saída.

Para receber transcrições, atualize a configuração da sessão. É necessário adicionar os objetos input_audio_transcription e output_audio_transcription e garantir que text esteja incluído em response_modalities.

Para melhorar a qualidade da transcrição do reconhecimento automático de fala (ASR) multilíngue, é possível fornecer dicas de idioma usando o campo language_codes em input_audio_transcription ou output_audio_transcription. Recomendamos fornecer dicas para melhorar a qualidade da transcrição, já que isso reduz o risco de detecção incorreta de idioma, especialmente para comandos curtos. O campo language_codes aceita uma lista de códigos de idioma BCP-47 (por exemplo, "en-US", "es-US").

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

Processar a resposta

O exemplo de código a seguir demonstra como se conectar usando a sessão configurada e extrair as partes de texto (transcrições) junto com os dados de áudio.

# 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.")

A seguir