오디오 입출력으로 Gemini Realtime API 호출하기

오디오 입출력으로 Gemini Realtime API 호출하기

LiteLLM Proxy를 사용해 Gemini Realtime API를 오디오 입출력과 함께 호출하는 방법을 알아봐요. 이 기능은 LiteLLM Proxy v1.70.1 이상이 필요해요.

출처: 문서

본문

• LiteLLM Proxy용 config.yaml 설정

model_list:
  - model_name: "gemini-flash-live"
    litellm_params:
      model: gemini/gemini-2.0-flash-live-001
    model_info:
      mode: realtime

• LiteLLM Proxy 시작

litellm --config config.yaml

• 테스트 스크립트 실행

import asyncioimport websocketsimport jsonimport base64from dotenv import load_dotenvimport waveimport base64import soundfile as sfimport sounddevice as sdimport ioimport numpy as np# Load environment variablesOPENAI_API_KEY = "sk-"  # Replace with your LiteLLM API keyOPENAI_API_URL = 'ws://{PROXY_URL}/v1/realtime?model=gemini-flash-live' # REPLACE WITH `wss://{PROXY_URL}/v1/realtime?model=gemini-flash-live` for secure connectionWAV_FILE_PATH = "/path/to/audio.wav"  # Replace with your .wav file pathasync def send_session_update(ws):    session_update = {
        "type": "session.update",
        "session": {
            "conversation_id": "123456",
            "language": "en-US",
            "transcription_mode": "fast",
            "modalities": ["text"]
        }
    }
    await ws.send(json.dumps(session_update))async def send_audio_file(ws, file_path):    with wave.open(file_path, 'rb') as wav_file:        chunk_size = 1024  # Adjust as needed
        while True:
            chunk = wav_file.readframes(chunk_size)
            if not chunk:
                break
            base64_audio = base64.b64encode(chunk).decode('utf-8')
            audio_message = {
                "type": "input_audio_buffer.append",
                "audio": base64_audio
            }
            await ws.send(json.dumps(audio_message))
            await asyncio.sleep(0.1)  # Add a small delay to simulate real-time streaming
    # Send end of audio stream message
    await ws.send(json.dumps({"type": "input_audio_buffer.end"}))def play_base64_audio(base64_string, sample_rate=24000, channels=1):    # Decode the base64 string
    audio_data = base64.b64decode(base64_string)
    # Convert to numpy array
    audio_np = np.frombuffer(audio_data, dtype=np.int16)
    # Reshape if stereo
    if channels == 2:
        audio_np = audio_np.reshape(-1, 2)
    # Normalize
    audio_float = audio_np.astype(np.float32) / 32768.0
    # Play the audio
    sd.play(audio_float, sample_rate)
    sd.wait()def combine_base64_audio(base64_strings):    # Step 1: Decode base64 strings to binary
    binary_data = [base64.b64decode(s) for s in base64_strings]
        # Step 2: Concatenate binary data
    combined_binary = b''.join(binary_data)
        # Step 3: Encode combined binary back to base64
    combined_base64 = base64.b64encode(combined_binary).decode('utf-8')
        return combined_base64async def listen_in_background(ws):    combined_b64_audio_str = []
    try:
         while True:
            response = await ws.recv()
            message_json = json.loads(response)
            print(f"message_json: {message_json}")
            if message_json['type'] == 'response.audio.delta' and message_json.get('delta'):
                play_base64_audio(message_json["delta"])
    except Exception:
         print("END OF STREAM")async def main():    async with websockets.connect(
        OPENAI_API_URL,
        additional_headers={
            "Authorization": f"Bearer {OPENAI_API_KEY}",
            "OpenAI-Beta": "realtime=v1"
        }
    ) as ws:
        asyncio.create_task(listen_in_background(ws=ws))
        await send_session_update(ws)
        await send_audio_file(ws, WAV_FILE_PATH)
        if __name__ == "__main__":
    asyncio.run(main())

더 알아보기 (Learn more)

• LiteLLM Realtime API