A API Gemini Live permite interações de voz e vídeo bidirecionais de baixa latência com o Gemini. Com a API Live, você pode oferecer aos usuários finais a experiência de conversas por voz naturais e semelhantes às humanas, além de interromper as respostas do modelo usando comandos de voz. A API Live pode processar entradas de texto, áudio e vídeo e fornecer saídas de texto e áudio.
Recursos
A API Live apresenta as seguintes especificações técnicas:
- Entradas:texto, áudio e vídeo
- Saídas:texto e áudio (fala sintetizada)
- Duração padrão da sessão:10 minutos
- A duração da sessão pode ser estendida em incrementos de 10 minutos, conforme necessário
- Janela de contexto:32 mil tokens
- Seleção entre 8 vozes para respostas
- Suporte para respostas em 31 idiomas
Usar a API Live
As seções a seguir fornecem exemplos de como usar os recursos da API Live.
Para mais informações, consulte o guia de referência da API Gemini Live.
Enviar texto e receber áudio
SDK da Gen AI para Python
voice_name = "Aoede" # @param ["Aoede", "Puck", "Charon", "Kore", "Fenrir", "Leda", "Orus", "Zephyr"] config = LiveConnectConfig( response_modalities=["AUDIO"], speech_config=SpeechConfig( voice_config=VoiceConfig( prebuilt_voice_config=PrebuiltVoiceConfig( voice_name=voice_name, ) ), ), ) async with client.aio.live.connect( model=MODEL_ID, config=config, ) as session: text_input = "Hello? Gemini are you there?" display(Markdown(f"**Input:** {text_input}")) await session.send_client_content( turns=Content(role="user", parts=[Part(text=text_input)])) audio_data = [] async for message in session.receive(): if ( message.server_content.model_turn and message.server_content.model_turn.parts ): for part in message.server_content.model_turn.parts: if part.inline_data: audio_data.append( np.frombuffer(part.inline_data.data, dtype=np.int16) ) if audio_data: display(Audio(np.concatenate(audio_data), rate=24000, autoplay=True))
Enviar e receber mensagens de texto
Gen AI SDK for Python
Instalar
pip install --upgrade google-genai
Defina variáveis de ambiente para usar o SDK da IA generativa com a Vertex AI:
# Replace the `GOOGLE_CLOUD_PROJECT` and `GOOGLE_CLOUD_LOCATION` values # with appropriate values for your project. export GOOGLE_CLOUD_PROJECT=GOOGLE_CLOUD_PROJECT export GOOGLE_CLOUD_LOCATION=us-central1 export GOOGLE_GENAI_USE_VERTEXAI=True
Enviar áudio
SDK da Gen AI para Python
import asyncio import wave from google import genai client = genai.Client(api_key="GEMINI_API_KEY", http_options={'api_version': 'v1alpha'}) model = "gemini-2.0-flash-live-preview-04-09" config = {"response_modalities": ["AUDIO"]} async def main(): async with client.aio.live.connect(model=model, config=config) as session: wf = wave.open("audio.wav", "wb") wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(24000) message = "Hello? Gemini are you there?" await session.send_client_content( turns=Content(role="user", parts=[Part(text=message)])) async for idx,response in async_enumerate(session.receive()): if response.data is not None: wf.writeframes(response.data) # Un-comment this code to print audio data info # if response.server_content.model_turn is not None: # print(response.server_content.model_turn.parts[0].inline_data.mime_type) wf.close() if __name__ == "__main__": asyncio.run(main())
A API Live é compatível com os seguintes formatos de áudio:
- Formato de áudio de entrada:áudio PCM bruto de 16 bits a 16 kHz little-endian
- Formato de áudio de saída:áudio PCM bruto de 16 bits em little-endian a 24 kHz
Transcrição de áudio
A API Live pode transcrever áudio de entrada e saída:
SDK da Gen AI para Python
# Set model generation_config CONFIG = { 'response_modalities': ['AUDIO'], } headers = { "Content-Type": "application/json", "Authorization": f"Bearer {bearer_token[0]}", } # Connect to the server async with connect(SERVICE_URL, additional_headers=headers) as ws: # Setup the session await ws.send( json.dumps( { "setup": { "model": "gemini-2.0-flash-live-preview-04-09", "generation_config": CONFIG, 'input_audio_transcription': {}, 'output_audio_transcription': {} } } ) ) # Receive setup response raw_response = await ws.recv(decode=False) setup_response = json.loads(raw_response.decode("ascii")) # Send text message text_input = "Hello? Gemini are you there?" display(Markdown(f"**Input:** {text_input}")) msg = { "client_content": { "turns": [{"role": "user", "parts": [{"text": text_input}]}], "turn_complete": True, } } await ws.send(json.dumps(msg)) responses = [] input_transcriptions = [] output_transcriptions = [] # 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 if (input_transcription := server_content.get("inputTranscription")) is not None: if (text := input_transcription.get("text")) is not None: input_transcriptions.append(text) if (output_transcription := server_content.get("outputTranscription")) is not None: if (text := output_transcription.get("text")) is not None: output_transcriptions.append(text) 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 if input_transcriptions: display(Markdown(f"**Input transcription >** {''.join(input_transcriptions)}")) if responses: # Play the returned audio message display(Audio(np.concatenate(responses), rate=24000, autoplay=True)) if output_transcriptions: display(Markdown(f"**Output transcription >** {''.join(output_transcriptions)}"))
Mudar as configurações de idioma e voz
A API Live usa o Chirp 3 para oferecer suporte a respostas de voz sintetizadas em 8 vozes HD e 31 idiomas.
Você pode escolher entre as seguintes vozes:
Aoede
(feminino)Charon
(masculino)Fenrir
(masculino)Kore
(feminino)Leda
(feminino)Orus
(masculino)Puck
(masculino)Zephyr
(feminino)
Para conferir demonstrações de como essas vozes soam e conferir a lista completa de idiomas disponíveis, consulte Chirp 3: vozes HD.
Para definir a voz e o idioma da resposta:
SDK da Gen AI para Python
config = LiveConnectConfig( response_modalities=["AUDIO"], speech_config=SpeechConfig( voice_config=VoiceConfig( prebuilt_voice_config=PrebuiltVoiceConfig( voice_name=voice_name, ) ), language_code="en-US", ), )
Console
- Abra o Vertex AI Studio > API em tempo real.
- No expansor Saídas, selecione uma voz no menu suspenso Voz.
- No mesmo menu, selecione um idioma no menu suspenso Language.
- Clique em Iniciar sessão para iniciar a sessão.
Para melhores resultados ao solicitar e exigir que o modelo responda em um idioma diferente do inglês, inclua o seguinte como parte das instruções do sistema:
RESPOND IN LANGUAGE. YOU MUST RESPOND UNMISTAKABLY IN LANGUAGE.
Ter uma conversa por streaming
SDK da Gen AI para Python
Configure uma conversa com a API que 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-2.0-flash-live-preview-04-09", "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, insira suas instruções ou digite q
, quit
ou exit
para sair.
await main()
Console
- Abra o Vertex AI Studio > API em tempo real.
- Clique em Iniciar sessão para iniciar a sessão de conversa.
Para encerrar a sessão, clique em
Parar sessão.Duração da sessão
A duração máxima padrão de uma sessão de conversa é de 10 minutos. Uma notificação go_away
(BidiGenerateContentServerMessage.go_away
) será enviada de volta ao
cliente 60 segundos antes do término da sessão.
Ao usar a API, você pode estender a duração da sessão em incrementos de 10 minutos. Não há limite para a quantidade de vezes que você pode estender uma sessão. Para um exemplo de como estender a duração da sessão, consulte Ativar e desativar a retomada da sessão. No momento, esse recurso está disponível apenas na API, não no Vertex AI Studio.
Janela de contexto
O comprimento máximo de contexto para uma sessão na API Live é de 32.768 tokens por padrão, que são alocados para armazenar dados em tempo real que são transmitidos em uma taxa de 25 tokens por segundo (TPS) para áudio e 258 TPS para vídeo e outros conteúdos, incluindo entradas baseadas em texto, saídas de modelo etc.
Se a janela de contexto exceder o comprimento máximo, os contextos das viradas mais antigas da janela de contexto serão truncados para que o tamanho geral da janela de contexto fique abaixo da limitação.
O comprimento de contexto padrão da sessão e o comprimento de contexto desejado após
o truncamento podem ser configurados usando
context_window_compression.trigger_tokens
e
context_window_compression.sliding_window.target_tokens
da mensagem
de configuração, respectivamente.
Sessões simultâneas
Por padrão, é possível ter até 10 sessões simultâneas por projeto.
Atualizar as instruções do sistema durante a sessão
Com a API Live, você pode atualizar as instruções do sistema no meio de uma sessão ativa. Você pode usar isso para adaptar as respostas do modelo durante a sessão, como mudar o idioma em que o modelo responde para outro ou modificar o tom em que o modelo responde.
Mudar as configurações de detecção de atividade de voz
Por padrão, o modelo realiza automaticamente a detecção de atividade de voz (VAD, na sigla em inglês) em um
stream de entrada de áudio contínuo. O VAD pode ser configurado com o campo
realtimeInputConfig.automaticActivityDetection
da mensagem de configuração.
Quando o stream de áudio é pausado por mais de um segundo (por exemplo, porque o
usuário desativou o microfone), um evento
audioStreamEnd
precisa ser enviado para limpar qualquer áudio em cache. O cliente pode retomar o envio
de dados de áudio a qualquer momento.
Como alternativa, o VAD automático pode ser desativado definindo
realtimeInputConfig.automaticActivityDetection.disabled
como true
na mensagem
de configuração. Nessa configuração, o cliente é responsável por detectar a fala do usuário
e enviar mensagens
activityStart
e
activityEnd
nos momentos adequados. Um audioStreamEnd
não é enviado nessa
configuração. Em vez disso, qualquer interrupção do stream é marcada por uma
mensagem activityEnd
.
Ativar e desativar a retomada da sessão
Esse recurso é desativado por padrão. Ela precisa ser ativada pelo usuário sempre que ele chamar a API especificando o campo na solicitação da API. Além disso, a privacidade no nível do projeto é aplicada aos dados armazenados em cache. A ativação da retomada de sessão permite que o usuário se conecte novamente a uma sessão anterior em até 24 horas armazenando dados em cache, incluindo dados de comando de texto, vídeo e áudio e saídas de modelo, por até 24 horas. Para não reter dados, não ative esse recurso.
Para ativar o recurso de retomada de sessão, defina o campo session_resumption
da
mensagem BidiGenerateContentSetup
. Se ativada, o servidor vai fazer periodicamente
um snapshot dos contextos de sessão em cache atuais e armazená-lo no
armazenamento interno. Quando um snapshot é feito, um resumption_update
é retornado com o ID do gerenciador, que pode ser gravado e usado mais tarde para retomar
a sessão do snapshot.
Confira um exemplo de como ativar o recurso de retomada de sessão e coletar as informações do ID do manipulador:
SDK da Gen AI para Python
# Set model generation_config CONFIG = {"response_modalities": ["TEXT"]} headers = { "Content-Type": "application/json", "Authorization": f"Bearer {bearer_token[0]}", } # Connect to the server async with connect(SERVICE_URL, additional_headers=headers) as ws: # Setup the session await ws.send( json.dumps( { "setup": { "model": "gemini-2.0-flash-live-preview-04-09", "generation_config": CONFIG, # Enable session resumption. "session_resumption": {}, } } ) ) # Receive setup response raw_response = await ws.recv(decode=False) setup_response = json.loads(raw_response.decode("ascii")) # Send text message text_input = "Hello? Gemini are you there?" display(Markdown(f"**Input:** {text_input}")) msg = { "client_content": { "turns": [{"role": "user", "parts": [{"text": text_input}]}], "turn_complete": True, } } await ws.send(json.dumps(msg)) responses = [] handle_id = "" turn_completed = False resumption_received = False # Receive chucks of server response, # wait for turn completion and resumption handle. async for raw_response in ws: response = json.loads(raw_response.decode()) server_content = response.pop("serverContent", None) resumption_update = response.pop("sessionResumptionUpdate", None) if server_content is not None: model_turn = server_content.pop("modelTurn", None) if model_turn is not None: parts = model_turn.pop("parts", None) if parts is not None: responses.append(parts[0]["text"]) # End of turn turn_complete = server_content.pop("turnComplete", None) if turn_complete: turn_completed = True elif resumption_update is not None: handle_id = resumption_update['newHandle'] resumption_received = True else: continue if turn_complete and resumption_received: break # Print the server response display(Markdown(f"**Response >** {''.join(responses)}")) display(Markdown(f"**Session Handle ID >** {handle_id}"))
Se você quiser retomar a sessão anterior, defina o campo handle
da
configuração setup.session_resumption
para o ID de identificador gravado
anteriormente:
SDK da Gen AI para Python
# Set model generation_config CONFIG = {"response_modalities": ["TEXT"]} headers = { "Content-Type": "application/json", "Authorization": f"Bearer {bearer_token[0]}", } # Connect to the server async with connect(SERVICE_URL, additional_headers=headers) as ws: # Setup the session await ws.send( json.dumps( { "setup": { "model": "gemini-2.0-flash-live-preview-04-09", "generation_config": CONFIG, # Enable session resumption. "session_resumption": { "handle": handle_id, }, } } ) ) # Receive setup response raw_response = await ws.recv(decode=False) setup_response = json.loads(raw_response.decode("ascii")) # Send text message text_input = "What was the last question I asked?" display(Markdown(f"**Input:** {text_input}")) msg = { "client_content": { "turns": [{"role": "user", "parts": [{"text": text_input}]}], "turn_complete": True, } } await ws.send(json.dumps(msg)) responses = [] handle_id = "" turn_completed = False resumption_received = False # Receive chucks of server response, # wait for turn completion and resumption handle. async for raw_response in ws: response = json.loads(raw_response.decode()) server_content = response.pop("serverContent", None) resumption_update = response.pop("sessionResumptionUpdate", None) if server_content is not None: model_turn = server_content.pop("modelTurn", None) if model_turn is not None: parts = model_turn.pop("parts", None) if parts is not None: responses.append(parts[0]["text"]) # End of turn turn_complete = server_content.pop("turnComplete", None) if turn_complete: turn_completed = True elif resumption_update is not None: handle_id = resumption_update['newHandle'] resumption_received = True else: continue if turn_complete and resumption_received: break # Print the server response # Expected answer: "You just asked if I was there." display(Markdown(f"**Response >** {''.join(responses)}")) display(Markdown(f"**Session Handle >** {resumption_update}"))
Se você quiser retomar a sessão sem problemas, ative o modo transparente:
SDK da Gen AI para Python
await ws.send( json.dumps( { "setup": { "model": "gemini-2.0-flash-live-preview-04-09", "generation_config": CONFIG, # Enable session resumption. "session_resumption": { "transparent": True, }, } } ) )
Usar chamadas de função
É possível usar a chamada de função para criar uma descrição de uma função e transmitir essa descrição ao modelo em uma solicitação. A resposta do modelo inclui o nome de uma função que corresponde à descrição e os argumentos para chamá-la.
Todas as funções precisam ser declaradas no início da sessão enviando definições
de ferramentas como parte da mensagem setup
.
SDK da Gen AI para Python
# Set model generation_config CONFIG = {"response_modalities": ["TEXT"]} # Define function declarations TOOLS = { "function_declarations": { "name": "get_current_weather", "description": "Get the current weather in the given location", "parameters": { "type": "OBJECT", "properties": {"location": {"type": "STRING"}}, }, } } headers = { "Content-Type": "application/json", "Authorization": f"Bearer {bearer_token[0]}", } # Connect to the server async with connect(SERVICE_URL, additional_headers=headers) as ws: # Setup the session await ws.send( json.dumps( { "setup": { "model": "gemini-2.0-flash-live-preview-04-09", "generation_config": CONFIG, "tools": TOOLS, } } ) ) # Receive setup response raw_response = await ws.recv(decode=False) setup_response = json.loads(raw_response.decode()) # Send text message text_input = "Get the current weather in Santa Clara, San Jose and Mountain View" display(Markdown(f"**Input:** {text_input}")) msg = { "client_content": { "turns": [{"role": "user", "parts": [{"text": text_input}]}], "turn_complete": True, } } await ws.send(json.dumps(msg)) responses = [] # Receive chucks of server response async for raw_response in ws: response = json.loads(raw_response.decode("UTF-8")) if (tool_call := response.get("toolCall")) is not None: for function_call in tool_call["functionCalls"]: responses.append(f"FunctionCall: {str(function_call)}\n") if (server_content := response.get("serverContent")) is not None: if server_content.get("turnComplete", True): break # Print the server response display(Markdown("**Response >** {}".format("\n".join(responses))))
Usar a execução do código
É possível usar a execução de código com a API Live para gerar e executar código Python diretamente.
SDK da Gen AI para Python
# Set model generation_config CONFIG = {"response_modalities": ["TEXT"]} # Set code execution TOOLS = {"code_execution": {}} headers = { "Content-Type": "application/json", "Authorization": f"Bearer {bearer_token[0]}", } # Connect to the server async with connect(SERVICE_URL, additional_headers=headers) as ws: # Setup the session await ws.send( json.dumps( { "setup": { "model": "gemini-2.0-flash-live-preview-04-09", "generation_config": CONFIG, "tools": TOOLS, } } ) ) # Receive setup response raw_response = await ws.recv(decode=False) setup_response = json.loads(raw_response.decode()) # Send text message text_input = "Write code to calculate the 15th fibonacci number then find the nearest palindrome to it" display(Markdown(f"**Input:** {text_input}")) msg = { "client_content": { "turns": [{"role": "user", "parts": [{"text": text_input}]}], "turn_complete": True, } } await ws.send(json.dumps(msg)) responses = [] # Receive chucks of server response async for raw_response in ws: response = json.loads(raw_response.decode("UTF-8")) if (server_content := response.get("serverContent")) is not None: if (model_turn:= server_content.get("modelTurn")) is not None: if (parts := model_turn.get("parts")) is not None: if parts[0].get("text"): responses.append(parts[0]["text"]) for part in parts: if (executable_code := part.get("executableCode")) is not None: display( Markdown( f"""**Executable code:** ```py {executable_code.get("code")} ``` """ ) ) if server_content.get("turnComplete", False): break # Print the server response display(Markdown(f"**Response >** {''.join(responses)}"))
Usar o Embasamento com a Pesquisa Google
É possível usar a orientação com a Pesquisa
Google com
a API Live usando google_search
:
SDK da Gen AI para Python
# Set model generation_config CONFIG = {"response_modalities": ["TEXT"]} # Set google search TOOLS = {"google_search": {}} headers = { "Content-Type": "application/json", "Authorization": f"Bearer {bearer_token[0]}", } # Connect to the server async with connect(SERVICE_URL, additional_headers=headers) as ws: # Setup the session await ws.send( json.dumps( { "setup": { "model": "gemini-2.0-flash-live-preview-04-09", "generation_config": CONFIG, "tools": TOOLS, } } ) ) # Receive setup response raw_response = await ws.recv(decode=False) setup_response = json.loads(raw_response.decode()) # Send text message text_input = "What is the current weather in San Jose, CA?" display(Markdown(f"**Input:** {text_input}")) msg = { "client_content": { "turns": [{"role": "user", "parts": [{"text": text_input}]}], "turn_complete": True, } } await ws.send(json.dumps(msg)) 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: responses.append(parts[0]["text"]) # End of turn turn_complete = server_content.pop("turnComplete", None) if turn_complete: break # Print the server response display(Markdown("**Response >** {}".format("\n".join(responses))))
Limitações
Consulte a seção de limitações da API Gemini Live da nossa documentação de referência para conferir a lista completa de limitações atuais da API Live.
Preços
Consulte nossa página de preços para mais detalhes.
Mais informações
Para mais informações sobre a API Live, como a referência da API
WebSocket
, consulte a documentação da API
Gemini.