通过 Gen AI SDK 开始使用 Live API

本教程介绍了如何使用 Gen AI SDK 连接到 Live API。您将向模型发送音频文件，并收到音频回答。

准备工作

  1. 向 Vertex AI 进行身份验证。如需了解详情，请参阅身份验证方法。

  2. 安装google-genai库：

    pip install --upgrade google-genai

  3. 为项目 ID 和位置设置环境变量。将 PROJECT_ID 替换为您的Google Cloud 项目 ID。

    export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True

开始音频会话

此示例将建立一个会话，通过文件流式传输音频，并输出回答中的音频块大小。

import asyncio
import os
import sys
from pathlib import Path
from google import genai
from google.genai import types

# Configuration
MODEL = "gemini-live-2.5-flash-preview-native-audio-09-2025"
config = {
   "response_modalities": ["audio"],
}

client = genai.Client()

async def main():
   # Establish WebSocket session
   async with client.aio.live.connect(model=MODEL, config=config) as session:
       print("Session established. Sending audio...")

       # Download sample if missing
       if not os.path.exists("input.wav"):
           !wget -q https://storage.googleapis.com/cloud-samples-data/generative-ai/audio/where_the_nearest_train_station_is.wav -O input.wav

       # Send Input (Simulated from file)
       # In production, this would be a microphone stream
       # Format: PCM, 16kHz, 16-bit, Mono, Little-Endian
       with open("input.wav", "rb") as f:
           while chunk := f.read(1024):
               await session.send_realtime_input(
                   audio=types.Blob(data=chunk, mime_type="audio/pcm;rate=16000")
               )
               await asyncio.sleep(0.01) # Simulate real-time stream

       # Receive Output
       async for message in session.receive():
           if message.server_content:
               # Check for interruptions (User barge-in)
               if message.server_content.interrupted:
                   print("[Interrupted] Clear client audio buffer immediately.")
                   continue

               # Process Audio Chunks
               model_turn = message.server_content.model_turn
               if model_turn and model_turn.parts:
                   for part in model_turn.parts:
                       if part.inline_data:
                           # Output is PCM, 24kHz, 16-bit, Mono
                           audio_data = part.inline_data.data
                           print(f"Received audio chunk: {len(audio_data)} bytes")

           if message.server_content.turn_complete:
               print("Turn complete.")

if "ipykernel" in sys.modules:
   # Run directly in notebook
   await main()
else:
   # Run as standard .py script
   asyncio.run(main())

后续步骤