Google 생성형 AI SDK를 사용하여 Live API 시작하기

이 튜토리얼에서는 Google 생성형 AI SDK의 Python 버전을 사용하여 Live API에 연결하는 방법을 보여줍니다. 이 튜토리얼에서는 Gen AI SDK를 사용하여 Live API를 사용하도록Google Cloud 프로젝트를 설정하고, 모델에 오디오 파일을 전송하고, 오디오를 응답으로 수신합니다.

시작하기 전에

요청을 보내려면 Vertex AI로 인증을 설정해야 합니다. API 키를 사용하거나 애플리케이션 기본 사용자 인증 정보 (ADC)를 사용하여 인증을 설정할 수 있습니다.

이 튜토리얼에서는 API 키를 사용하여 가장 빠르게 시작할 수 있습니다.

ADC를 사용하여 인증을 설정하는 방법은 빠른 시작을 참고하세요.

생성형 AI SDK 설치

다음을 실행하여 google-genai 라이브러리를 설치합니다.

pip install --upgrade google-genai

환경 변수 설정

프로젝트 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())

다음 단계