Iniciar e gerenciar sessões ao vivo

A API Gemini Live permite interações de voz e texto de baixa latência processando fluxos contínuos de áudio ou texto chamados de sessões para fornecer 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.

Nesta página, mostramos 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 instruções do sistema durante uma sessão
• Configurar a janela de contexto de uma sessão
• Ativar a transcrição para uma sessão
• Criptografar dados de sessão com uma chave de criptografia gerenciada pelo cliente (CMEK)

Ciclo de vida da sessão

Sem a compressã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. Exceder esses limites encerra a sessão, mas você pode usar a compressão da janela de contexto para estender as sessões por um período ilimitado.

A vida útil de uma conexão é limitada a cerca de 10 minutos devido a restrições de conexão do WebSocket. Quando a conexão termina, a sessão também é encerrada. Nesse caso, é possível configurar uma única sessão para permanecer ativa em várias conexões usando a retomada de sessão. Você também vai receber uma mensagem no GoAway antes do fim da conexão para tomar outras medidas.

Máximo de sessões simultâneas

Você pode 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 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 > Transmitir em tempo real.
2. Clique em Iniciar sessão para começar a conversa.

Para encerrar a sessão, clique em Parar sessão.

Python

Antes de começar, autentique a Gemini Enterprise Agent Platform usando uma chave de API ou credenciais padrão do aplicativo (ADC, na sigla em inglês):

gcloud auth application-default login
      

Para mais informações sobre como configurar a autenticação, consulte nosso guia de 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-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

Ao usar WebSockets, a conexão é estabelecida com um handshake padrão do WebSocket. O endpoint é regional e usa tokens de portador do 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-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())
      

Observação: embora alguns parâmetros, incluindo instruções do sistema, possam ser ajustados no meio da sessão, o modelo e outros parâmetros geralmente são imutáveis depois que a mensagem de configuração é processada.

Ampliar uma sessão

Observação: a extensão de sessão só está disponível ao usar o SDK de IA Generativa do Google, não o Vertex AI Studio.

A duração máxima padrão de uma sessão de conversa é de 10 minutos, que pode ser estendida usando notificações goAway. Uma notificação goAway (BidiGenerateContentServerMessage.goAway) é enviada ao cliente 60 segundos antes do fim 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 sua 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 ver um exemplo de como retomar uma sessão, consulte Retomar uma sessão anterior.

O exemplo a seguir mostra como detectar um encerramento de sessão iminente ao detectar 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

Importante: se você precisar garantir retenção zero de dados no seu aplicativo, não ative a retomada da sessão.

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

Por padrão, a retomada de sessão está desativada. Para ativar a retomada da sessão, defina o campo sessionResumption da mensagem BidiGenerateContentSetup. Se ativado, o servidor envia periodicamente mensagens SessionResumptionUpdate que contêm 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-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())
      

Ativar a retomada de sessão sem problemas com o Modo som ambiente

Ao ativar a retomada da sessão, você também pode ativar o modo transparente para ajudar a tornar o processo de retomada mais fácil 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 som ambiente:

Python

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

Criptografar dados de sessão com CMEK

A retomada da sessão armazena dados de sessão em cache em repouso por até 24 horas. Para proteger esses dados com sua própria chave em vez de umGoogle-owned and Google-managed encryption key, use um perfil de veiculação. Um perfil de exibição vincula uma chave do Cloud KMS a um projeto, local e escopo da API. Enquanto um perfil de serviço existir, a plataforma do agente criptografa os dados automaticamente. A configuração da sessão e o formato da solicitação não mudam.

A CMEK de perfil de veiculação para a API Gemini Live é compatível com as multirregiões us e eu. A região global não é compatível.

Para configurar a CMEK para a API Gemini Live, faça o seguinte:

1. Ative as APIs Cloud KMS e Agent Platform no seu projeto. Consulte Antes de começar.
2. Crie uma chave do Cloud KMS na multirregião us ou eu. Consulte Criar um keyring e uma chave.
3. Conceda ao agente de serviço da Gemini Enterprise Agent Platform a função roles/cloudkms.cryptoKeyEncrypterDecrypter na chave. Consulte Conceder permissões do Agent Platform.
4. Crie um perfil de exibição que defina scope como GEMINI_LIVE e cmekConfig.encryptionSpec.kmsKeyName como sua chave. Consulte Criar um perfil de veiculação.

Observação: os dados armazenados antes da criação do perfil de veiculação não são criptografados novamente de forma retroativa com sua chave.

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

Com a API Gemini Live, é possível atualizar as instruções do sistema durante uma sessão ativa. Use isso para adaptar as respostas do modelo, como mudar o idioma 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 atualizada vai continuar em vigor durante o restante da sessão.

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, habilite a compressã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 (use o controle deslizante Tamanho máximo do conteúdo no Vertex AI Studio ou trigger_tokens na API), o servidor corta automaticamente as conversas mais antigas ou as resume para manter o contexto dentro do limite. No 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 compressã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 da meta 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 pretendido (target_tokens)	0	128.000

Para definir a janela de contexto:

Console

1. Abra Vertex AI Studio > Transmitir em tempo real.
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 para a sessão

É possível ativar as 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 de fala automático (ASR) multilíngue, forneça dicas de idioma usando o campo language_codes em input_audio_transcription ou output_audio_transcription. É recomendável 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": {},
}

Processando 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

• Enviar streams de áudio e vídeo
• Práticas recomendadas com a API Gemini Live
• Criar comandos multimodais
• Introdução às chamadas de função