傳送音訊和視訊串流

本文說明如何將音訊和視訊串流傳送至 Live API，與 Gemini 模型進行即時雙向通訊。瞭解如何設定及傳輸音訊和視訊資料，建構動態互動式應用程式。

傳送音訊串流

如要實作即時音訊，必須嚴格遵守取樣率規格，並謹慎管理緩衝區，確保低延遲和自然中斷。

Live API 支援下列音訊格式：

  • 輸入音訊：16 kHz 的原始 16 位元 PCM 音訊，小端序
  • 輸出音訊：24 kHz 的原始 16 位元 PCM 音訊，小端序

下列程式碼範例說明如何傳送串流音訊資料：

import asyncio
# Assumes session is an active Live API session
# and chunk_data contains bytes of raw 16-bit PCM audio at 16 kHz.
from google.genai import types
# Send audio input data in chunks
await session.send_realtime_input(
    audio=types.Blob(data=chunk_data, mime_type="audio/pcm;rate=16000")
)

用戶端必須維護播放緩衝區。伺服器會在 server_content 訊息中串流音訊區塊。用戶端負責解碼、緩衝處理及播放資料。

下列程式碼範例說明如何處理串流音訊資料：

import asyncio
# Assumes session is an active Live API session
# and audio_queue is an asyncio.Queue for buffering audio for playback.
import numpy as np

async for msg in session.receive():
    server_content = msg.server_content
    if server_content:
        # 1. Handle Interruption
        if server_content.interrupted:
            print("\n[Interrupted] Flushing buffer...")
            # Clear the Python queue
            while not audio_queue.empty():
                try: audio_queue.get_nowait()
                except asyncio.QueueEmpty: break
            # Send signal to worker to reset hardware buffers if needed
            await audio_queue.put(None)
            continue

        # 2. Process Audio chunks
        if server_content.model_turn:
            for part in server_content.model_turn.parts:
                if part.inline_data:
                    # Add PCM data to playback queue
                    await audio_queue.put(np.frombuffer(part.inline_data.data, dtype='int16'))

傳送影片串流

影片串流可提供視覺背景資訊。Live API 預期會收到一系列不連續的影像畫面，並支援以 1 FPS 的速度輸入影片畫面。為獲得最佳結果，請使用 768x768 的原生解析度，並將影格速率設為 1 FPS。

下列程式碼範例說明如何傳送串流影片資料：

import asyncio
# Assumes session is an active Live API session
# and chunk_data contains bytes of a JPEG image.
from google.genai import types
# Send video input data in chunks
await session.send_realtime_input(
    media=types.Blob(data=chunk_data, mime_type="image/jpeg")
)

用戶端實作會從影片串流擷取影格，將其編碼為 JPEG Blob，然後使用 realtime_input 訊息結構傳輸。

import cv2
import asyncio
from google.genai import types

async def send_video_stream(session):
    # Open webcam
    cap = cv2.VideoCapture(0)

    while True:
        ret, frame = cap.read()
        if not ret: break

        # 1. Resize to optimal resolution (768x768 max)
        frame = cv2.resize(frame, (768, 768))

        # 2. Encode as JPEG
        _, buffer = cv2.imencode('.jpg', frame,)

        # 3. Send as realtime input
        await session.send_realtime_input(
            media=types.Blob(data=buffer.tobytes(), mime_type="image/jpeg")
        )

        # 4. Wait 1 second (1 FPS)
        await asyncio.sleep(1.0)

    cap.release()

設定媒體解析度

您可以在工作階段設定中設定 media_resolution 欄位，指定輸入媒體的解析度。降低解析度可減少權杖用量和延遲時間，提高解析度則可提升細節辨識度。支援的值包括 low、medium 和 high。

config = {
    "response_modalities": ["audio"],
    "media_resolution": "low",
}

後續步驟