このドキュメントでは、Gemini モデルとのリアルタイムの双方向通信のために、音声ストリームと動画ストリームを Live API に送信する方法について説明します。音声データと動画データを構成して送信し、動的でインタラクティブなアプリケーションを構築する方法を学習します。
音声ストリームを送信する
リアルタイム オーディオを実装するには、低レイテンシと自然な割り込み可能性を確保するために、サンプルレートの仕様を厳守し、バッファを慎重に管理する必要があります。
Live API がサポートする音声形式は、次のとおりです。
- 入力音声: RAW 16 ビット PCM 音声、16 kHz、リトル エンディアン
- 出力音声: RAW 16 ビット PCM 音声、24 kHz、リトル エンディアン
次のコードサンプルは、ストリーミング音声データを送信する方法を示しています。
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 の動画フレーム入力をサポートしています。最良の結果を得るには、1 FPS で 768x768 のネイティブ解像度を使用します。
次のコードサンプルは、ストリーミング動画データを送信する方法を示しています。
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",
}