Live API 可讓您與 Gemini 進行低延遲的雙向語音/視訊互動。您可以使用 Live API，為使用者提供自然流暢的對話體驗，並讓使用者透過語音指令中斷模型的回覆。模型可處理文字、音訊和影片輸入內容，並提供文字和音訊輸出內容。
您可以在 Google AI Studio 中試用 Live API。
最新資訊
請查看變更記錄，瞭解 Live API 的最新功能和能力！
使用 Live API
本節說明如何搭配使用 Live API 和我們的 SDK。如要進一步瞭解基礎 WebSockets API，請參閱 WebSockets API 參考資料。
如要使用所有功能，請務必安裝最新版 SDK，例如：pip install -U google-genai。
收發簡訊
import asyncio
from google import genai
client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"
config = {"response_modalities": ["TEXT"]}
async def main():
async with client.aio.live.connect(model=model, config=config) as session:
while True:
message = input("User> ")
if message.lower() == "exit":
break
await session.send_client_content(
turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
)
async for response in session.receive():
if response.text is not None:
print(response.text, end="")
if __name__ == "__main__":
asyncio.run(main())
接收音訊
以下範例說明如何接收音訊資料，並將資料寫入 .wav 檔案。
import asyncio
import wave
from google import genai
client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"
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={"role": "user", "parts": [{"text": message}]}, turn_complete=True
)
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())
音訊格式
Live API 支援下列音訊格式：
- 輸入音訊格式：原始 16 位元 PCM 音訊，位元深度為 16 位元，取樣率為 16 kHz 小端序
- 輸出音訊格式：24 kHz 小端序的原始 16 位元 PCM 音訊
串流音訊和視訊
系統指示
系統指令可讓您根據特定需求和用途，控制模型的行為。系統指示可在設定配置中設定，並在整個工作階段中持續生效。
from google.genai import types
config = {
"system_instruction": types.Content(
parts=[
types.Part(
text="You are a helpful assistant and answer in a friendly tone."
)
]
),
"response_modalities": ["TEXT"],
}
逐步更新內容
使用遞增更新功能傳送文字輸入內容、建立工作階段背景資訊或還原工作階段背景資訊。針對短時間內的互動，您可以傳送逐向互動，以代表確切的事件順序：
turns = [
{"role": "user", "parts": [{"text": "What is the capital of France?"}]},
{"role": "model", "parts": [{"text": "Paris"}]},
]
await session.send_client_content(turns=turns, turn_complete=False)
turns = [{"role": "user", "parts": [{"text": "What is the capital of Germany?"}]}]
await session.send_client_content(turns=turns, turn_complete=True)
{
"clientContent": {
"turns": [
{
"parts":[
{
"text": ""
}
],
"role":"user"
},
{
"parts":[
{
"text": ""
}
],
"role":"model"
}
],
"turnComplete": true
}
}
對於較長的背景資訊，建議您提供單一訊息摘要，以便釋出脈絡窗口，供後續互動使用。
變更語音
Live API 支援 Puck、Charon、Kore、Fenrir、Aoede、Leda、Orus 和 Zephyr 等語音。
如要指定語音，請在 speechConfig 物件中設定語音名稱，做為工作階段設定的一部分：
from google.genai import types
config = types.LiveConnectConfig(
response_modalities=["AUDIO"],
speech_config=types.SpeechConfig(
voice_config=types.VoiceConfig(
prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Kore")
)
)
)
{
"voiceConfig": {
"prebuiltVoiceConfig": {
"voiceName": "Kore"
}
}
}
變更語言
Live API 支援多種語言。
如要變更語言，請在 speechConfig 物件中設定語言代碼，做為工作階段設定的一部分：
from google.genai import types
config = types.LiveConnectConfig(
response_modalities=["AUDIO"],
speech_config=types.SpeechConfig(
language_code="de-DE",
)
)
使用工具
您可以使用 Live API 定義工具，例如函式呼叫、程式碼執行和 Google 搜尋。
使用函式呼叫
您可以將函式宣告定義為工作階段設定的一部分。詳情請參閱函式呼叫教學課程。
收到工具呼叫後，用戶端應使用 session.send_tool_response 方法回應 FunctionResponse 物件清單。
import asyncio
from google import genai
from google.genai import types
client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"
# Simple function definitions
turn_on_the_lights = {"name": "turn_on_the_lights"}
turn_off_the_lights = {"name": "turn_off_the_lights"}
tools = [{"function_declarations": [turn_on_the_lights, turn_off_the_lights]}]
config = {"response_modalities": ["TEXT"], "tools": tools}
async def main():
async with client.aio.live.connect(model=model, config=config) as session:
prompt = "Turn on the lights please"
await session.send_client_content(turns={"parts": [{"text": prompt}]})
async for chunk in session.receive():
if chunk.server_content:
if chunk.text is not None:
print(chunk.text)
elif chunk.tool_call:
function_responses = []
for fc in tool_call.function_calls:
function_response = types.FunctionResponse(
id=fc.id,
name=fc.name,
response={ "result": "ok" } # simple, hard-coded function response
)
function_responses.append(function_response)
await session.send_tool_response(function_responses=function_responses)
if __name__ == "__main__":
asyncio.run(main())
模型可從單一提示產生多個函式呼叫，以及連結輸出的必要程式碼。這個程式碼會在沙箱環境中執行，產生後續的 BidiGenerateContentToolCall 訊息。執行作業會暫停，直到每個函式呼叫的結果可用為止，藉此確保按順序處理。
音訊輸入和輸出會對模型使用函式呼叫的功能造成負面影響。
使用程式碼執行
您可以將程式碼執行作業定義為工作階段設定的一部分。詳情請參閱程式碼執行教學課程。
import asyncio
from google import genai
from google.genai import types
client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"
tools = [{'code_execution': {}}]
config = {"response_modalities": ["TEXT"], "tools": tools}
async def main():
async with client.aio.live.connect(model=model, config=config) as session:
prompt = "Compute the largest prime palindrome under 100000."
await session.send_client_content(turns={"parts": [{"text": prompt}]})
async for chunk in session.receive():
if chunk.server_content:
if chunk.text is not None:
print(chunk.text)
model_turn = chunk.server_content.model_turn
if model_turn:
for part in model_turn.parts:
if part.executable_code is not None:
print(part.executable_code.code)
if part.code_execution_result is not None:
print(part.code_execution_result.output)
if __name__ == "__main__":
asyncio.run(main())
使用「以 Google 搜尋建立基準」
您可以將 Google 搜尋設為工作階段設定的一部分，啟用基準建立功能。詳情請參閱基礎教學課程。
import asyncio
from google import genai
from google.genai import types
client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"
tools = [{'google_search': {}}]
config = {"response_modalities": ["TEXT"], "tools": tools}
async def main():
async with client.aio.live.connect(model=model, config=config) as session:
prompt = "When did the last Brazil vs. Argentina soccer match happen?"
await session.send_client_content(turns={"parts": [{"text": prompt}]})
async for chunk in session.receive():
if chunk.server_content:
if chunk.text is not None:
print(chunk.text)
# The model might generate and execute Python code to use Search
model_turn = chunk.server_content.model_turn
if model_turn:
for part in model_turn.parts:
if part.executable_code is not None:
print(part.executable_code.code)
if part.code_execution_result is not None:
print(part.code_execution_result.output)
if __name__ == "__main__":
asyncio.run(main())
結合多種工具
您可以在 Live API 中結合多種工具：
prompt = """
Hey, I need you to do three things for me.
1. Compute the largest prime palindrome under 100000.
2. Then use Google Search to look up information about the largest earthquake in California the week of Dec 5 2024?
3. Turn on the lights
Thanks!
"""
tools = [
{"google_search": {}},
{"code_execution": {}},
{"function_declarations": [turn_on_the_lights, turn_off_the_lights]},
]
config = {"response_modalities": ["TEXT"], "tools": tools}
處理中斷
使用者隨時可以中斷模型的輸出內容。當語音活動偵測功能 (VAD) 偵測到中斷情形時，系統會取消並捨棄目前的產生作業。只有已傳送至用戶端的資訊會保留在工作階段記錄中。接著，伺服器會傳送 BidiGenerateContentServerContent 訊息，回報中斷情形。
此外，Gemini 伺服器會捨棄所有待處理的函式呼叫，並傳送含有取消呼叫 ID 的 BidiGenerateContentServerContent 訊息。
async for response in session.receive():
if response.server_content.interrupted is True:
# The generation was interrupted
設定語音活動偵測 (VAD)
您可以設定或停用語音活動偵測 (VAD) 功能。
使用自動 VAD
根據預設，模型會自動對連續音訊輸入串流執行 VAD。您可以使用設定配置的 realtimeInputConfig.automaticActivityDetection 欄位設定 VAD。
如果音訊串流暫停超過一秒 (例如因為使用者關閉麥克風)，就應傳送 audioStreamEnd 事件，以便清除任何已快取的音訊。用戶端隨時可以恢復傳送音訊資料。
# example audio file to try:
# URL = "https://storage.googleapis.com/generativeai-downloads/data/hello_are_you_there.pcm"
# !wget -q $URL -O sample.pcm
import asyncio
from pathlib import Path
from google import genai
client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"
config = {"response_modalities": ["TEXT"]}
async def main():
async with client.aio.live.connect(model=model, config=config) as session:
audio_bytes = Path("sample.pcm").read_bytes()
await session.send_realtime_input(
audio=types.Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000")
)
# if stream gets paused, send:
# await session.send_realtime_input(audio_stream_end=True)
async for response in session.receive():
if response.text is not None:
print(response.text)
if __name__ == "__main__":
asyncio.run(main())
有了 send_realtime_input，API 就會根據 VAD 自動回應音訊。send_client_content 會依序將訊息新增至模型內容，而 send_realtime_input 則會犧牲確定性排序，以提升回應速度。
設定自動語音偵測功能
如要進一步控制 VAD 活動，您可以設定下列參數。詳情請參閱 API 參考資料。
from google.genai import types
config = {
"response_modalities": ["TEXT"],
"realtime_input_config": {
"automatic_activity_detection": {
"disabled": False, # default
"start_of_speech_sensitivity": types.StartSensitivity.START_SENSITIVITY_LOW,
"end_of_speech_sensitivity": types.EndSensitivity.END_SENSITIVITY_LOW,
"prefix_padding_ms": 20,
"silence_duration_ms": 100,
}
}
}
停用自動 VAD
或者，您也可以在設定訊息中將 realtimeInputConfig.automaticActivityDetection.disabled 設為 true，停用自動語音偵測功能。在這個設定中，用戶端負責偵測使用者的語音，並在適當時間傳送 activityStart 和 activityEnd 訊息。這個設定不會傳送 audioStreamEnd。而是會在串流中斷時標示 activityEnd 訊息。
config = {
"response_modalities": ["TEXT"],
"realtime_input_config": {"automatic_activity_detection": {"disabled": True}},
}
async with client.aio.live.connect(model=model, config=config) as session:
# ...
await session.send_realtime_input(activity_start=types.ActivityStart())
await session.send_realtime_input(
audio=types.Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000")
)
await session.send_realtime_input(activity_end=types.ActivityEnd())
# ...
取得符記數
您可以在傳回的伺服器訊息的 usageMetadata 欄位中，找到已使用的符記總數。
async for message in session.receive():
# The server will periodically send messages that include UsageMetadata.
if message.usage_metadata:
usage = message.usage_metadata
print(
f"Used {usage.total_token_count} tokens in total. Response token breakdown:"
)
for detail in usage.response_tokens_details:
match detail:
case types.ModalityTokenCount(modality=modality, token_count=count):
print(f"{modality}: {count}")
延長工作階段時間
工作階段持續時間上限可透過兩種機制延長至無限：
此外，您會在工作階段結束前收到GoAway 訊息，方便您採取進一步行動。
啟用內容視窗壓縮功能
如要啟用較長的工作階段，並避免連線突然中斷，您可以將 contextWindowCompression 欄位設為工作階段設定的一部分，啟用內容視窗壓縮功能。
在 ContextWindowCompressionConfig 中，您可以設定滑動視窗機制和觸發壓縮功能的符記數量。
from google.genai import types
config = types.LiveConnectConfig(
response_modalities=["AUDIO"],
context_window_compression=(
# Configures compression with default parameters.
types.ContextWindowCompressionConfig(
sliding_window=types.SlidingWindow(),
)
),
)
設定工作階段恢復功能
如要避免在伺服器定期重設 WebSocket 連線時終止工作階段，請在設定配置中設定 sessionResumption 欄位。
傳遞這項設定會導致伺服器傳送 SessionResumptionUpdate 訊息，可用於將上一個恢復權杖傳遞為後續連線的 SessionResumptionConfig.handle，藉此恢復工作階段。
import asyncio
from google import genai
from google.genai import types
client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"
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())
在工作階段中斷前收到訊息
伺服器會傳送「GoAway」GoAway訊息，表示目前的連線即將終止。這則訊息包含 timeLeft，可顯示剩餘時間，並讓您在連線因「已中止」而終止前採取進一步行動。
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)
產生完成後收到訊息
伺服器會傳送 generationComplete 訊息，表示模型已完成產生回應。
async for response in session.receive():
if response.server_content.generation_complete is True:
# The generation is complete
變更媒體解析度
您可以將 mediaResolution 欄位設為工作階段設定的一部分，藉此指定輸入媒體的媒體解析度：
from google.genai import types
config = types.LiveConnectConfig(
response_modalities=["AUDIO"],
media_resolution=types.MediaResolution.MEDIA_RESOLUTION_LOW,
)
接收音訊轉錄
您可以啟用模型音訊輸出的轉錄功能。系統會根據模型的回應推斷轉錄語言。
import asyncio
from google import genai
from google.genai import types
client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"
config = {"response_modalities": ["AUDIO"],
"output_audio_transcription": {}
}
async def main():
async with client.aio.live.connect(model=model, config=config) as session:
message = "Hello? Gemini are you there?"
await session.send_client_content(
turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
)
async for response in session.receive():
if response.server_content.model_turn:
print("Model turn:", response.server_content.model_turn)
if response.server_content.output_transcription:
print("Transcript:", response.server_content.output_transcription.text)
if __name__ == "__main__":
asyncio.run(main())
限制
規劃專案時，請考量 Live API 和 Gemini 2.0 的下列限制。
回應模式
您只能在工作階段設定中為每個工作階段設定一種回應模式 (TEXT 或 AUDIO)。嘗試設定這兩者會導致設定錯誤訊息。也就是說，您可以設定模型以文字或音訊回應，但在同一個工作階段中，不能同時使用這兩種回應。
用戶端驗證
Live API 只提供伺服器對伺服器驗證，不建議直接用於用戶端。用戶端輸入內容應透過中介應用程式伺服器進行轉送，以便透過 Live API 進行安全驗證。
工作階段持續時間
啟用工作階段壓縮功能，即可延長工作階段時間長度。如未壓縮，僅音訊的會話時間上限為 15 分鐘，音訊加視訊的會話時間上限為 2 分鐘。如果超過這些限制而未進行壓縮，連線就會終止。
此外，您可以設定工作階段恢復功能，讓用戶端可以恢復已終止的工作階段。
脈絡窗口
工作階段的脈絡窗口限制為 32k 個符號。
支援的語言
Live API 支援下列語言：
| 語言 | BCP-47 代碼 |
|---|---|
| 德文 (德國) | de-DE |
| 英文 (澳洲) | en-AU |
| 英文 (英國) | en-GB |
| 英文 (印度) | en-IN |
| 英文 (美國) | en-US |
| 西班牙文 (美國) | es-US |
| 法文 (法國) | fr-FR |
| 北印度文 (印度) | hi-IN |
| 葡萄牙文 (巴西) | pt-BR |
| 阿拉伯文 (一般) | ar-XA |
| 西班牙文 (西班牙) | es-ES |
| 法文 (加拿大) | fr-CA |
| 印尼文 (印尼) | id-ID |
| 義大利文 (義大利) | it-IT |
| 日文 (日本) | ja-JP |
| 土耳其文 (土耳其) | tr-TR |
| 越南文 (越南) | vi-VN |
| 孟加拉文 (印度) | bn-IN |
| 古吉拉特文 (印度) | gu-IN |
| 卡納達文 (印度) | kn-IN |
| 馬拉雅拉姆文 (印度) | ml-IN |
| 馬拉地文 (印度) | mr-IN |
| 泰米爾文 (印度) | ta-IN |
| 泰盧固文 (印度) | te-IN |
| 荷蘭文 (荷蘭) | nl-NL |
| 韓文 (南韓) | ko-KR |
| 中文 (中國) | cmn-CN |
| 波蘭文 (波蘭) | pl-PL |
| 俄文 (俄羅斯) | ru-RU |
| 泰文 (泰國) | th-TH |
第三方整合
如要部署網站和行動應用程式，您可以探索下列選項：