本頁面由 Cloud Translation API 翻譯而成。

Live API

Live API 可讓您與 Gemini 進行低延遲的雙向語音/視訊互動，讓您在與 Gemini 進行即時對話的同時，也能串流傳輸視訊輸入內容或分享螢幕畫面。使用 Live API，您可以為使用者提供自然流暢的語音對話體驗。

您可以在 Google AI Studio 中試用 Live API。如要在 Google AI Studio 中使用 Live API，請選取「串流」。

Live API 的運作方式

串流

Live API 會透過 WebSocket 連線使用串流模型。您與 API 互動時，系統會建立持續連線。您的輸入內容 (音訊、影片或文字) 會持續串流至模型，而模型的回應 (文字或音訊) 會透過相同的連線即時串流回傳。

這項雙向串流功能可確保低延遲，並支援語音活動偵測、工具使用和語音生成等功能。

Live API 總覽

如要進一步瞭解基礎 WebSockets API，請參閱 WebSockets API 參考資料。

使用 Live API 建構 {:#building-with-live-api}ß

建立連線

以下範例說明如何使用 API 金鑰建立連線：

PythonJavaScript

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:
        print("Session started")

if __name__ == "__main__":
    asyncio.run(main())

import { GoogleGenAI } from '@google/genai';

const ai = new GoogleGenAI({ apiKey: "GOOGLE_API_KEY" });
const model = 'gemini-2.0-flash-live-001';
const config = { responseModalities: [Modality.TEXT] };

async function main() {

    const session = await ai.live.connect({
        model: model,
        callbacks: {
            onopen: function () {
                console.debug('Opened');
            },
            onmessage: function (message) {
                console.debug(message);
            },
            onerror: function (e) {
                console.debug('Error:', e.message);
            },
            onclose: function (e) {
                console.debug('Close:', e.reason);
            },
        },
        config: config,
    });

    // Send content...

    session.close();
}

main();

收發文字

以下說明如何傳送及接收文字訊息：

PythonJavaScript

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:
        message = "Hello, how are you?"
        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())

import { GoogleGenAI, Modality } from '@google/genai';

const ai = new GoogleGenAI({ apiKey: "GOOGLE_API_KEY" });
const model = 'gemini-2.0-flash-live-001';
const config = { responseModalities: [Modality.TEXT] };

async function live() {
    const responseQueue = [];

    async function waitMessage() {
        let done = false;
        let message = undefined;
        while (!done) {
            message = responseQueue.shift();
            if (message) {
                done = true;
            } else {
                await new Promise((resolve) => setTimeout(resolve, 100));
            }
        }
        return message;
    }

    async function handleTurn() {
        const turns = [];
        let done = false;
        while (!done) {
            const message = await waitMessage();
            turns.push(message);
            if (message.serverContent && message.serverContent.turnComplete) {
                done = true;
            }
        }
        return turns;
    }

    const session = await ai.live.connect({
        model: model,
        callbacks: {
            onopen: function () {
                console.debug('Opened');
            },
            onmessage: function (message) {
                responseQueue.push(message);
            },
            onerror: function (e) {
                console.debug('Error:', e.message);
            },
            onclose: function (e) {
                console.debug('Close:', e.reason);
            },
        },
        config: config,
    });

    const simple = 'Hello how are you?';
    session.sendClientContent({ turns: simple });

    const turns = await handleTurn();
    for (const turn of turns) {
        if (turn.text) {
            console.debug('Received text: %s\n', turn.text);
        }
        else if (turn.data) {
            console.debug('Received inline data: %s\n', turn.data);
        }
    }

    session.close();
}

async function main() {
    await live().catch((e) => console.error('got error', e));
}

main();

傳送及接收音訊

您可以將音訊轉換為 16 位元 PCM、16 kHz、單聲道格式，這個範例會讀取 WAV 檔案，並以正確格式傳送：

PythonJavaScript

# Test file: https://storage.googleapis.com/generativeai-downloads/data/16000.wav
# Install helpers for converting files: pip install librosa soundfile
import asyncio
import io
from pathlib import Path
from google import genai
from google.genai import types
import soundfile as sf
import librosa

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:

        buffer = io.BytesIO()
        y, sr = librosa.load("sample.wav", sr=16000)
        sf.write(buffer, y, sr, format='RAW', subtype='PCM_16')
        buffer.seek(0)
        audio_bytes = buffer.read()

        # If already in correct format, you can use this:
        # audio_bytes = Path("sample.pcm").read_bytes()

        await session.send_realtime_input(
            audio=types.Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000")
        )

        async for response in session.receive():
            if response.text is not None:
                print(response.text)

if __name__ == "__main__":
    asyncio.run(main())

// Test file: https://storage.googleapis.com/generativeai-downloads/data/16000.wav
// Install helpers for converting files: npm install wavefile
import { GoogleGenAI, Modality } from '@google/genai';
import * as fs from "node:fs";
import pkg from 'wavefile';
const { WaveFile } = pkg;

const ai = new GoogleGenAI({ apiKey: "GOOGLE_API_KEY" });
const model = 'gemini-2.0-flash-live-001';
const config = { responseModalities: [Modality.TEXT] };

async function live() {
    const responseQueue = [];

    async function waitMessage() {
        let done = false;
        let message = undefined;
        while (!done) {
            message = responseQueue.shift();
            if (message) {
                done = true;
            } else {
                await new Promise((resolve) => setTimeout(resolve, 100));
            }
        }
        return message;
    }

    async function handleTurn() {
        const turns = [];
        let done = false;
        while (!done) {
            const message = await waitMessage();
            turns.push(message);
            if (message.serverContent && message.serverContent.turnComplete) {
                done = true;
            }
        }
        return turns;
    }

    const session = await ai.live.connect({
        model: model,
        callbacks: {
            onopen: function () {
                console.debug('Opened');
            },
            onmessage: function (message) {
                responseQueue.push(message);
            },
            onerror: function (e) {
                console.debug('Error:', e.message);
            },
            onclose: function (e) {
                console.debug('Close:', e.reason);
            },
        },
        config: config,
    });

    // Send Audio Chunk
    const fileBuffer = fs.readFileSync("sample.wav");

    // Ensure audio conforms to API requirements (16-bit PCM, 16kHz, mono)
    const wav = new WaveFile();
    wav.fromBuffer(fileBuffer);
    wav.toSampleRate(16000);
    wav.toBitDepth("16");
    const base64Audio = wav.toBase64();

    // If already in correct format, you can use this:
    // const fileBuffer = fs.readFileSync("sample.pcm");
    // const base64Audio = Buffer.from(fileBuffer).toString('base64');

    session.sendRealtimeInput(
        {
            audio: {
                data: base64Audio,
                mimeType: "audio/pcm;rate=16000"
            }
        }

    );

    const turns = await handleTurn();
    for (const turn of turns) {
        if (turn.text) {
            console.debug('Received text: %s\n', turn.text);
        }
        else if (turn.data) {
            console.debug('Received inline data: %s\n', turn.data);
        }
    }

    session.close();
}

async function main() {
    await live().catch((e) => console.error('got error', e));
}

main();

您可以將 AUDIO 設為回應模式，接收音訊。這個範例會將收到的資料儲存為 WAV 檔案：

PythonJavaScript

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 how are you?"
        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())

import { GoogleGenAI, Modality } from '@google/genai';

const ai = new GoogleGenAI({ apiKey: "GOOGLE_API_KEY" });
const model = 'gemini-2.0-flash-live-001';
const config = { responseModalities: [Modality.AUDIO] };

async function live() {
    const responseQueue = [];

    async function waitMessage() {
        let done = false;
        let message = undefined;
        while (!done) {
            message = responseQueue.shift();
            if (message) {
                done = true;
            } else {
                await new Promise((resolve) => setTimeout(resolve, 100));
            }
        }
        return message;
    }

    async function handleTurn() {
        const turns = [];
        let done = false;
        while (!done) {
            const message = await waitMessage();
            turns.push(message);
            if (message.serverContent && message.serverContent.turnComplete) {
                done = true;
            }
        }
        return turns;
    }

    const session = await ai.live.connect({
        model: model,
        callbacks: {
            onopen: function () {
                console.debug('Opened');
            },
            onmessage: function (message) {
                responseQueue.push(message);
            },
            onerror: function (e) {
                console.debug('Error:', e.message);
            },
            onclose: function (e) {
                console.debug('Close:', e.reason);
            },
        },
        config: config,
    });

    const simple = 'Hello how are you?';
    session.sendClientContent({ turns: simple });

    const turns = await handleTurn();

    // Combine audio data strings and save as wave file
    const combinedAudio = turns.reduce((acc, turn) => {
        if (turn.data) {
            const buffer = Buffer.from(turn.data, 'base64');
            const intArray = new Int16Array(buffer.buffer, buffer.byteOffset, buffer.byteLength / Int16Array.BYTES_PER_ELEMENT);
            return acc.concat(Array.from(intArray));
        }
        return acc;
    }, []);

    const audioBuffer = new Int16Array(combinedAudio);

    const wf = new WaveFile();
    wf.fromScratch(1, 24000, '16', audioBuffer);
    fs.writeFileSync('output.wav', wf.toBuffer());

    session.close();
}

async function main() {
    await live().catch((e) => console.error('got error', e));
}

main();

音訊格式

Live API 中的音訊資料一律為原始、小端序的 16 位元 PCM。音訊輸出一律會使用 24 kHz 的取樣率。輸入音訊的預設取樣率為 16kHz，但 Live API 會視需要重新取樣，因此可傳送任何取樣率。如要傳達輸入音訊的取樣率，請將每個含有音訊的 Blob 的 MIME 類型設為 audio/pcm;rate=16000 之類的值。

接收音訊轉錄

您可以在設定設定檔中傳送 output_audio_transcription，啟用模型的音訊輸出轉錄功能。系統會根據模型的回覆推斷語音轉錄語言。

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

您可以在設定設定檔中傳送 input_audio_transcription，啟用音訊輸入的轉錄功能。

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": ["TEXT"],
    "realtime_input_config": {
        "automatic_activity_detection": {"disabled": True},
        "activity_handling": "NO_INTERRUPTION",
    },
    "input_audio_transcription": {},
}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        audio_data = Path("sample.pcm").read_bytes()

        await session.send_realtime_input(activity_start=types.ActivityStart())
        await session.send_realtime_input(
            audio=types.Blob(data=audio_data, mime_type='audio/pcm;rate=16000')
        )
        await session.send_realtime_input(activity_end=types.ActivityEnd())

        async for msg in session.receive():
            if msg.server_content.input_transcription:
                print('Transcript:', msg.server_content.input_transcription.text)

if __name__ == "__main__":
    asyncio.run(main())

串流音訊和影片

系統指示

系統指令可讓您根據特定需求和用途，控制模型的行為。系統指示可在設定配置中設定，並在整個工作階段中持續生效。

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"],
}

逐步更新內容

使用遞增更新功能傳送文字輸入內容、建立工作階段背景資訊或還原工作階段背景資訊。如果是短時間的內容，您可以傳送逐向互動，以代表確切的事件順序：

PythonJSON

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 物件中設定語音名稱，做為工作階段設定的一部分：

PythonJSON

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

generateContent

Live API 支援多種語言。

如要變更語言，請在 speechConfig 物件中設定語言代碼，做為工作階段設定的一部分：

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    speech_config=types.SpeechConfig(
        language_code="de-DE",
    )
)

使用工具搭配 Live API

您可以使用 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 訊息。

非同步函式呼叫

根據預設，執行作業會暫停，直到每個函式呼叫的結果可用為止，以確保按順序處理。也就是說，您無法在函式執行期間繼續與模型互動。

如果不想封鎖對話，您可以指示模型以非同步方式執行函式。

如要這麼做，您必須先在函式定義中新增 behavior：

  # Non-blocking function definitions
  turn_on_the_lights = {"name": "turn_on_the_lights", "behavior": "NON_BLOCKING"} # turn_on_the_lights will run asynchronously
  turn_off_the_lights = {"name": "turn_off_the_lights"} # turn_off_the_lights will still pause all interactions with the model

NON-BLOCKING 可確保函式以非同步方式執行，同時讓您繼續與模型互動。

接著，您需要使用 scheduling 參數，告訴模型在收到 FunctionResponse 時的行為方式。可以是：

中斷其執行的工作，並立即告知您收到的回應 (scheduling="INTERRUPT")。
等待目前作業完成 (scheduling="WHEN_IDLE")，
或者不採取任何行動，稍後在討論中使用該知識 (scheduling="SILENT")

# Non-blocking function definitions
  function_response = types.FunctionResponse(
      id=fc.id,
      name=fc.name,
      response={
          "result": "ok",
          "scheduling": "INTERRUPT" # Can also be WHEN_IDLE or SILENT
      }
  )

程式碼執行

您可以將程式碼執行作業定義為工作階段設定的一部分。詳情請參閱程式碼執行教學課程。

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
from google.genai import types

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,
        }
    }
}

停用自動語音偵測功能

或者，您也可以在設定訊息中將 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,
)

限制

規劃專案時，請考量 Live API 的下列限制。

回應模式

您只能在工作階段設定中為每個工作階段設定一個回應模式 (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

第三方整合

如要部署網站和行動應用程式，您可以參考以下選項：

後續步驟

使用暫時性權杖實作用戶端對伺服器的連線。
請試用 Google AI Studio 中的 Live API。
如要進一步瞭解 Gemini 2.0 Flash Live，請參閱模型頁面。
請參閱 Live API 食譜、Live API 工具食譜和 Live API 入門指南指令碼，瞭解更多範例。