Comienza a usar la API de Live con el SDK de IA generativa de Google

En este instructivo, se muestra cómo conectarse a la API de Live con la versión de Python del SDK de IA generativa de Google. En este instructivo, configurarás un proyecto deGoogle Cloud para usar la API de Live con el SDK de IA generativa, enviarás un archivo de audio al modelo y recibirás audio como respuesta.

Antes de comenzar

Antes de enviar solicitudes, debes configurar la autenticación con Vertex AI. Puedes configurar la autenticación con una clave de API o con credenciales predeterminadas de la aplicación (ADC).

En este instructivo, la forma más rápida de comenzar es usar una clave de API:

Si deseas obtener instrucciones para configurar la autenticación con las ADC, consulta nuestra guía de inicio rápido.

Instala el SDK de IA generativa

Ejecuta el siguiente comando para instalar la biblioteca de google-genai:

pip install --upgrade google-genai

Configura variables de entorno

Establece variables de entorno para el ID y la ubicación de tu proyecto. Reemplaza PROJECT_ID por el ID del proyecto de Google Cloud.

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

Cómo iniciar una sesión de audio

En este ejemplo, se establece una sesión, se transmite audio desde un archivo y se imprime el tamaño de los fragmentos de audio en la respuesta.

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

¿Qué sigue?