Conversas interativas com a API Live

A API Live permite interações bidirecionais de voz e vídeo com baixa latência com o Gemini.

Este guia aborda como configurar uma conversa interativa bidirecional, ajustar as definições de áudio, gerir sessões e muito mais.

Modelos suportados

Pode usar a API Live com os seguintes modelos:

Versão do modelo Nível de disponibilidade
gemini-live-2.5-flash GA privado*
gemini-live-2.5-flash-preview-native-audio Pré-visualização pública
gemini-live-2.5-flash-preview-native-audio-09-2025 Pré-visualização pública

* Contacte o representante da equipa da sua Conta Google para pedir acesso.

Iniciar uma conversa

Consola

  1. Abra o Vertex AI Studio > Transmitir em tempo real.
  2. Clique em Iniciar sessão para iniciar a conversa.

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

Python

Para ativar conversas interativas em tempo real, execute o seguinte exemplo num computador local com acesso ao microfone e ao altifalante (não num bloco de notas do Colab). Este exemplo configura uma conversa com a API, o que lhe permite enviar comandos de áudio e receber respostas de áudio: 

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

Configure uma conversa com a API que lhe permite enviar comandos de texto e receber respostas de áudio:

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

Inicie a conversa, introduza os seus comandos ou escreva q, quit ou exit para sair.

await main()

Altere as definições de idioma e voz

A API Live usa o Chirp 3 para suportar respostas de voz sintetizada numa variedade de vozes e idiomas em HD. Para ver uma lista completa e demonstrações de cada voz, consulte o artigo Chirp 3: vozes em HD.

Para definir a voz e o idioma de resposta:

Consola

  1. Abra o Vertex AI Studio > Transmitir em tempo real.
  2. No expansor Saídas, selecione uma voz no menu pendente Voz.
  3. No mesmo expansor, selecione um idioma no menu pendente Idioma.
  4. Clique em Iniciar sessão para iniciar a sessão.

Python

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

Altere as definições de deteção de atividade de voz

A deteção de atividade de voz (VAD) permite que o modelo reconheça quando uma pessoa está a falar. Isto é essencial para criar conversas naturais, uma vez que permite a um utilizador interromper o modelo em qualquer altura.

O modelo realiza automaticamente a deteção de atividade de voz (VAD) numa stream de entrada de áudio contínua. Pode configurar as definições de VAD através do campo realtimeInputConfig.automaticActivityDetection da mensagem de configuração. Quando a DVA deteta uma interrupção, a geração em curso é cancelada e rejeitada. Apenas as informações já enviadas ao cliente são retidas no histórico de sessões. Em seguida, o servidor envia uma mensagem para comunicar a interrupção.

Se a stream de áudio for pausada durante mais de um segundo (por exemplo, se o utilizador desativar o microfone), envie um evento audioStreamEnd para limpar o áudio em cache. O cliente pode retomar o envio de dados de áudio em qualquer altura.

Em alternativa, desative a DVA automática definindo realtimeInputConfig.automaticActivityDetection.disabled como true na mensagem de configuração. Com esta configuração, o cliente deteta a voz do utilizador e envia mensagens activityStart e activityEnd nos momentos adequados. Não é enviado um audioStreamEnd. As interrupções são marcadas com 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)}"))

Prolongue uma sessão

O comprimento máximo predefinido de uma sessão de conversa é de 10 minutos. É enviada uma goAway notificação (BidiGenerateContentServerMessage.goAway) ao cliente 60 segundos antes do fim da sessão.

Pode prolongar a duração da sessão em incrementos de 10 minutos através do SDK de IA gen. Não existe um limite para o número de vezes que pode prolongar uma sessão. Para ver um exemplo, consulte Ative e desative a retoma da sessão.

Capacidade de resposta

A janela de contexto da API Live é usada para armazenar dados transmitidos em tempo real (25 tokens por segundo (TPS) para áudio e 258 TPS para vídeo) e outro conteúdo, incluindo entradas de texto e resultados do modelo.

Se a janela de contexto exceder o comprimento máximo (definido através do controlo de deslize Tamanho máximo do conteúdo no Vertex AI Studio ou trigger_tokens na API), as interações mais antigas são truncadas através da compressão da janela de contexto para evitar a terminação abrupta da sessão. A compressão da janela de contexto é acionada quando a janela de contexto atinge o comprimento máximo (definido no Vertex AI Studio através do controlo de deslize Tamanho do contexto alvo ou através de target_tokens na API) e elimina as partes mais antigas da conversa até que a contagem total de tokens volte a este tamanho alvo.

Por exemplo, se o comprimento máximo do contexto estiver definido como 32000 tokens e o tamanho de destino estiver definido como 16000 tokens:

  1. Turno 1: a conversa começa. Neste exemplo, o pedido usa 12 000 tokens.
    • Tamanho total do contexto: 12 000 tokens
  2. Turno 2: faz outro pedido. Este pedido usa mais 12 000 tokens.
    • Tamanho total do contexto: 24 000 tokens
  3. Turno 3: faz outro pedido. Este pedido usa 14 000 tokens.
    • Tamanho total do contexto: 38 000 tokens

Uma vez que o tamanho total do contexto é agora superior ao máximo de 32 000 tokens, a compressão da capacidade de resposta é acionada. O sistema volta ao início da conversa e começa a eliminar turnos antigos até que o tamanho total dos tokens seja inferior ao objetivo de 16 000 tokens:

  • Elimina a volta 1 (12 000 tokens). O total é agora de 26 000 tokens, o que continua a ser superior ao objetivo de 16 000 tokens.
  • Elimina Turn 2 (12 000 tokens). O total é agora de 14 000 tokens.

O resultado final é que apenas a Turn 3 permanece na memória ativa e a conversa continua a partir desse ponto.

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

Definição (sinalização da API) Valor mínimo Valor máximo
Comprimento máximo do contexto (trigger_tokens) 5000 128 000
Tamanho do contexto de destino (target_tokens) 0 128 000

Para definir a capacidade de resposta:

Consola

  1. Abra o Vertex AI Studio > Transmitir em tempo real.
  2. Clique para abrir o menu Avançadas.
  3. Na secção Contexto da sessão, use o controlo de deslize Tamanho máximo do contexto para definir o tamanho do contexto para um valor entre 5000 e 128 000.
  4. (Opcional) Na mesma secção, use o controlo de deslize Tamanho do contexto de destino para definir o tamanho de destino para 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 = 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),
      ),
  )

Sessões em simultâneo

Pode ter até 1000 sessões simultâneas por projeto.

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

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

Para atualizar as instruções do sistema durante a sessão, pode enviar conteúdo de texto com a função system. A instrução do sistema atualizada vai permanecer em vigor durante a sessão restante.

Python

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

Ative e desative a retoma da sessão

A retoma da sessão permite-lhe voltar a ligar-se a uma sessão anterior no prazo de 24 horas. Isto é conseguido através do armazenamento de dados em cache, incluindo texto, comandos de vídeo, áudio e resultados do modelo. A privacidade ao nível do projeto é aplicada a estes dados em cache.

Por predefinição, o recomeço da sessão está desativado.

Para ativar a funcionalidade de retoma da sessão, defina o campo sessionResumption da mensagem BidiGenerateContentSetup. Se estiver ativado, o servidor tira periodicamente um instantâneo dos contextos da sessão em cache atuais e armazena-o no armazenamento interno.

Quando um instantâneo é tirado com êxito, é devolvido um resumptionUpdate com o ID do identificador que pode registar e usar mais tarde para retomar a sessão a partir do instantâneo.

Segue-se um exemplo de como ativar a retoma da sessão e obter o ID do identificador:

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 conseguir um reinício da sessão sem problemas, ative o modo transparente:

Python

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

Depois de ativar o modo transparente, o índice da mensagem do cliente que corresponde ao instantâneo do contexto é devolvido explicitamente. Isto ajuda a identificar que mensagem do cliente tem de enviar novamente quando retoma a sessão a partir do identificador de retoma.

Mais informações

Para mais informações sobre a utilização da API Live, consulte: