Die Gemini Live API ermöglicht bidirektionale Sprach- und Videointeraktionen mit Gemini bei niedriger Latenz. Mit der Live API können Sie Endnutzern natürliche, menschenähnliche Sprachunterhaltungen bieten und ihnen die Möglichkeit geben, die Antworten des Modells per Sprachbefehl zu unterbrechen. Die Live API kann Text-, Audio- und Videoeingaben verarbeiten und Text- und Audioausgaben bereitstellen.
Features
Die Live API hat die folgenden technischen Spezifikationen:
- Eingabedaten:Text, Audio und Video
- Ausgaben:Text und Audio (synthetische Sprache)
- Standarddauer der Sitzung:10 Minuten
- Die Sitzungsdauer kann nach Bedarf in Schritten von 10 Minuten verlängert werden.
- Kontextfenster:32.000 Tokens
- Auswahl zwischen 8 Stimmen für Antworten
- Unterstützung für Antworten in 31 Sprachen
Live API verwenden
In den folgenden Abschnitten findest du Beispiele für die Verwendung der Funktionen der Live API.
Weitere Informationen findest du im Referenzleitfaden für die Gemini Live API.
Text senden und Audio empfangen
Gen AI SDK für 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))
SMS senden und empfangen
Gen AI SDK for Python
Installieren
pip install --upgrade google-genai
Legen Sie Umgebungsvariablen fest, um das Gen AI SDK mit Vertex AI zu verwenden:
# 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
Audio senden
Gen AI SDK für 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())
Die Live API unterstützt die folgenden Audioformate:
- Eingabeaudioformat:Rohes 16-Bit-PCM-Audio mit 16 kHz und Little Endian
- Audioausgabeformat:Rohes 16-Bit-PCM-Audio mit 24 kHz Little Endian
Audiotranskription
Die Live API kann sowohl Eingabe- als auch Ausgabeaudio transkribieren:
Gen AI SDK für 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)}"))
Sprach- und Spracherkennungseinstellungen ändern
Die Live API verwendet Chirp 3, um synthetisierte Sprachantworten in 8 HD-Stimmen und 31 Sprachen zu unterstützen.
Sie können zwischen den folgenden Stimmen wählen:
Aoede(weiblich)Charon(männlich)Fenrir(männlich)Kore(weiblich)Leda(weiblich)Orus(männlich)Puck(männlich)Zephyr(weiblich)
Demos dieser Stimmen und eine vollständige Liste der verfügbaren Sprachen finden Sie unter Chirp 3: HD-Stimmen.
So legen Sie die Stimme und Sprache für die Antwort fest:
Gen AI SDK für 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
- Öffnen Sie Vertex AI Studio > Live API.
- Wählen Sie im maximierten Bereich Ausgaben im Drop-down-Menü Stimme eine Stimme aus.
- Wählen Sie im selben Drop-down-Menü Sprache eine Sprache aus.
- Klicken Sie auf Sitzung starten, um die Sitzung zu starten.
Wenn Sie das Modell auffordern und anweisen möchten, in einer anderen Sprache als Englisch zu antworten, sollten Sie Folgendes in Ihre Systemanweisungen aufnehmen:
RESPOND IN LANGUAGE. YOU MUST RESPOND UNMISTAKABLY IN LANGUAGE.
Gestreamte Unterhaltungen
Gen AI SDK für Python
Richten Sie eine Unterhaltung mit der API ein, über die Sie Textprompts senden und Audioantworten erhalten können:
# 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()
Starten Sie die Unterhaltung, geben Sie Ihre Prompts ein oder geben Sie q, quit oder exit ein, um sie zu beenden.
await main()
Console
- Öffnen Sie Vertex AI Studio > Live API.
- Klicken Sie auf Sitzung starten, um die Unterhaltungssitzung zu starten.
Klicken Sie auf Sitzung beenden, um die Sitzung zu beenden.
Sitzungsdauer
Die Standardmaximaldauer einer Unterhaltungssitzung beträgt 10 Minuten. 60 Sekunden vor Ende der Sitzung wird eine go_away-Benachrichtigung (BidiGenerateContentServerMessage.go_away) an den Client gesendet.
Wenn Sie die API verwenden, können Sie die Sitzungsdauer in Schritten von 10 Minuten verlängern. Sie können eine Sitzung beliebig oft verlängern. Ein Beispiel dazu, wie Sie die Sitzungsdauer verlängern, finden Sie unter Sitzungswiederaufnahme aktivieren und deaktivieren. Diese Funktion ist derzeit nur in der API und nicht in Vertex AI Studio verfügbar.
Verlaufszeitraum
Die maximale Kontextlänge für eine Sitzung in der Live API beträgt standardmäßig 32.768 Token. Diese werden zum Speichern von Echtzeitdaten zugewiesen, die mit einer Rate von 25 Token pro Sekunde (TPS) für Audio und 258 TPS für Video sowie anderen Inhalten wie textbasierten Eingaben und Modellausgaben gestreamt werden.
Wenn das Kontextfenster die maximale Kontextlänge überschreitet, werden die Kontexte der ältesten Abbiege aus dem Kontextfenster abgeschnitten, sodass die Gesamtgröße des Kontextfensters unter der Einschränkung liegt.
Die Standardkontextlänge der Sitzung und die Zielkontextlänge nach der Kürzung können jeweils mit den Feldern context_window_compression.trigger_tokens und context_window_compression.sliding_window.target_tokens der Einrichtungsnachricht konfiguriert werden.
Gleichzeitig geöffnete Sitzungen
Standardmäßig können Sie pro Projekt bis zu 10 gleichzeitige Sitzungen haben.
Systemanweisungen während einer Sitzung aktualisieren
Mit der Live API können Sie die Systemanweisungen auch während einer aktiven Sitzung aktualisieren. So können Sie die Antworten des Modells während einer Sitzung anpassen, z. B. die Sprache, in der das Modell antwortet, oder den Tonfall, mit dem das Modell antworten soll.
Einstellungen für die Erkennung von Sprachaktivitäten ändern
Standardmäßig führt das Modell automatisch eine Sprachaktivitätserkennung (VAD) auf einem kontinuierlichen Audioeingangsstream durch. Die VAD-Funktion kann mit dem Feld realtimeInputConfig.automaticActivityDetection der Einrichtungsnachricht konfiguriert werden.
Wenn der Audiostream länger als eine Sekunde pausiert wird (z. B. weil der Nutzer das Mikrofon ausgeschaltet hat), sollte ein audioStreamEnd-Ereignis gesendet werden, um den Cache zu leeren. Der Client kann das Senden von Audiodaten jederzeit fortsetzen.
Alternativ kann die automatische VAD deaktiviert werden, indem Sie in der Einrichtungsnachricht realtimeInputConfig.automaticActivityDetection.disabled auf true festlegen. Bei dieser Konfiguration ist der Client dafür verantwortlich, die Sprache des Nutzers zu erkennen und activityStart- und activityEnd-Nachrichten zum richtigen Zeitpunkt zu senden. In dieser Konfiguration wird kein audioStreamEnd gesendet. Stattdessen wird jede Unterbrechung des Streams durch eine activityEnd-Nachricht gekennzeichnet.
Sitzungsfortsetzung aktivieren und deaktivieren
Diese Funktion ist standardmäßig deaktiviert. Sie muss jedes Mal vom Nutzer aktiviert werden, wenn er die API aufruft, indem er das Feld in der API-Anfrage angibt. Für im Cache gespeicherte Daten gilt der Datenschutz auf Projektebene. Wenn Sie die Fortsetzung der Sitzung aktivieren, kann der Nutzer innerhalb von 24 Stunden eine Verbindung zu einer vorherigen Sitzung herstellen. Dazu werden bis zu 24 Stunden lang im Cache gespeicherte Daten wie Text-, Video- und Audioprompts sowie Modellausgaben verwendet. Wenn Sie keine Daten aufbewahren möchten, aktivieren Sie diese Funktion nicht.
Wenn Sie die Funktion zur Sitzungsfortsetzung aktivieren möchten, legen Sie das Feld session_resumption der BidiGenerateContentSetup-Nachricht fest. Wenn diese Option aktiviert ist, erstellt der Server regelmäßig einen Snapshot der aktuellen zwischengespeicherten Sitzungskontexte und speichert ihn im internen Speicher. Wenn ein Snapshot erfolgreich erstellt wurde, wird eine resumption_update mit der Handle-ID zurückgegeben, die du aufzeichnen und später verwenden kannst, um die Sitzung aus dem Snapshot fortzusetzen.
Hier ein Beispiel für die Aktivierung der Funktion zur Sitzungswiederaufnahme und das Erfassen der Informationen zur Handle-ID:
Gen AI SDK für 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}"))
Wenn Sie die vorherige Sitzung fortsetzen möchten, können Sie das Feld handle der setup.session_resumption-Konfiguration auf die zuvor aufgezeichnete Handle-ID festlegen:
Gen AI SDK für 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}"))
Wenn Sie die Sitzung nahtlos fortsetzen möchten, können Sie den transparenten Modus aktivieren:
Gen AI SDK für 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, }, } } ) )
Nachdem der transparente Modus aktiviert wurde, wird der Index der Clientnachricht, der dem Kontext-Snapshot entspricht, explizit zurückgegeben. So kannst du leichter feststellen, welche Clientnachricht du noch einmal senden musst, wenn du die Sitzung über den Wiederanlauf-Handle fortsetzt.
Funktionsaufrufe verwenden
Mit Funktionsaufrufen können Sie eine Funktionsbeschreibung erstellen und diese dann in einer Anfrage an das Modell übergeben. Die Antwort des Modells enthält den Namen einer Funktion, die der Beschreibung entspricht, sowie die Argumente, um diese aufzurufen.
Alle Funktionen müssen zu Beginn der Sitzung deklariert werden. Dazu müssen Sie Tooldefinitionen als Teil der setup-Nachricht senden.
Gen AI SDK für 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))))
Codeausführung verwenden
Mit der Codeausführung in der Live API können Sie Python-Code direkt generieren und ausführen.
Gen AI SDK für 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)}"))
Fundierung mit der Google Suche verwenden
Du kannst Fundierung mit der Google Suche mit der Live API und google_search verwenden:
Gen AI SDK für 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))))
Beschränkungen
Eine vollständige Liste der aktuellen Einschränkungen für die Live API findest du in der Referenzdokumentation im Abschnitt Einschränkungen der Gemini Live API.
Preise
Weitere Informationen finden Sie auf unserer Preisseite.
Weitere Informationen
Weitere Informationen zur Live API, z. B. die WebSocket API-Referenz, finden Sie in der Gemini API-Dokumentation.