Mulai menggunakan Gemini Live API menggunakan Google Gen AI SDK

Tutorial ini menunjukkan cara terhubung ke Gemini Live API menggunakan Google Gen AI SDK versi Python. Dalam tutorial ini, Anda akan menyiapkan project Google Cloud untuk menggunakan Live API dengan Gen AI SDK, mengirimkan file audio ke model, dan menerima audio sebagai respons.

Sebelum memulai

Sebelum dapat mengirim permintaan, Anda harus menyiapkan autentikasi dengan Vertex AI. Anda dapat menyiapkan autentikasi menggunakan kunci API atau menggunakan kredensial default aplikasi (ADC).

Untuk tutorial ini, cara tercepat untuk memulai adalah dengan menggunakan kunci API:

Untuk mengetahui petunjuk tentang cara menyiapkan autentikasi menggunakan ADC, lihat panduan memulai kami.

Menginstal Gen AI SDK

Jalankan perintah berikut untuk menginstal library google-genai:

pip install --upgrade google-genai

Menyiapkan variabel lingkungan

Tetapkan variabel lingkungan untuk project ID dan lokasi Anda. Ganti PROJECT_ID dengan ID project Google Cloud Anda.

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

Mulai sesi audio

Contoh ini membuat sesi, melakukan streaming audio dari file, dan mencetak ukuran potongan audio dalam respons.

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

Langkah berikutnya