发送音频和视频流

本文档介绍了如何将音频和视频流发送到 Live API，以便与 Gemini 模型进行实时双向通信。了解如何配置和传输音频和视频数据，以构建动态互动式应用。

发送音频流

实现实时音频需要严格遵守采样率规范并进行精细的缓冲区管理，以确保低延迟和自然的可打断性。

Live API 支持以下音频格式：

  • 输入音频：原始 16 位 PCM 音频，采样率 16 kHz，小端字节序
  • 输出音频：原始 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")
)

客户端实现从视频 Feed 中捕获一帧，将其编码为 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 字段，以指定输入媒体的分辨率。较低的分辨率可减少 token 用量并降低延迟，而较高的分辨率可提高细节识别能力。支持的值包括 MEDIA_RESOLUTION_LOW、MEDIA_RESOLUTION_MEDIUM 和 MEDIA_RESOLUTION_HIGH。

from google.genai import types

config = {
    "response_modalities": ["audio"],
    "media_resolution": types.MediaResolution.MEDIA_RESOLUTION_LOW,
}

后续步骤