Vertex AI Gemini Live - Realtime API

Vertex AI Gemini Live - Realtime API

LiteLLM의 통합 /realtime 엔드포인트를 통해 Vertex AI의 Gemini Live API(BidiGenerateContent)를 사용하는 방법을 알아봐요. OpenAI Realtime 프로토콜을 말해요.

출처: 문서

본문

Vertex AI의 Gemini Live API(BidiGenerateContent)를 OpenAI Realtime 프로토콜을 말하는 LiteLLM의 통합 /realtime 엔드포인트로 사용해요.

기능 지원
Proxy (/realtime) ✅
Voice in / Voice out ✅
Text in / Text out ✅
Server VAD ✅
Output transcription ✅

설정

1. 인증

LiteLLM은 API 키가 아닌 Google Cloud 자격 증명(OAuth2 Bearer 토큰)을 사용해요.

gcloud auth application-default login

또는 서비스 계정 키 파일 설정:

export GOOGLE_APPLICATION_CREDENTIALS=/path/to/sa-key.json

2. Proxy 구성

model_list:
  - model_name: vertex-gemini-live
    litellm_params:
      model: vertex_ai/gemini-2.0-flash-live-001
      vertex_project: your-gcp-project-id
      vertex_location: us-east4   # or any supported region, or "global"

general_settings:
  master_key: «redacted:sk-…»  # your master key

3. Proxy 시작

litellm --config config.yaml --port 4000

사용법

Python (websockets)

import asyncio
import json
import websockets

PROXY_URL = "ws://localhost:4000/realtime?model=vertex-gemini-live"
API_KEY = "sk-your-key"

async def main():
    async with websockets.connect(
        PROXY_URL,
        additional_headers={"api-key": API_KEY},
    ) as ws:
        # Wait for session.created
        event = json.loads(await ws.recv())
        print(f"session.created: {event['session']['id']}")

        # Send a text message
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Say hello in one sentence."}],
            },
        }))

        # Collect the response
        async for raw in ws:
            ev = json.loads(raw)
            t = ev.get("type", "")
            if t == "response.text.delta":
                print(ev.get("delta", ""), end="", flush=True)
            elif t == "response.done":
                print("\n[done]")
                break

asyncio.run(main())

Node.js

const WebSocket = require("ws");

const ws = new WebSocket(
  "ws://localhost:4000/realtime?model=vertex-gemini-live",
  { headers: { "api-key": "sk-your-key" } }
);

ws.on("open", () => {
  ws.send(JSON.stringify({
    type: "conversation.item.create",
    item: {
      type: "message",
      role: "user",
      content: [{ type: "input_text", text: "Say hello." }],
    },
  }));
});

ws.on("message", (data) => {
  const ev = JSON.parse(data);
  if (ev.type === "response.text.delta") process.stdout.write(ev.delta);
  if (ev.type === "response.done") ws.close();
});

OpenAI SDK (Python)

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="http://localhost:4000",
    api_key="sk-your-key",
)

async def main():
    async with client.beta.realtime.connect(
        model="vertex-gemini-live"
    ) as conn:
        await conn.session.update(session={"modalities": ["text"]})

        await conn.conversation.item.create(
            item={
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Say hello."}],
            }
        )

        async for event in conn:
            if event.type == "response.text.delta":
                print(event.delta, end="", flush=True)
            elif event.type == "response.done":
                print()
                break

asyncio.run(main())

Voice in / Voice out

완전한 음성 예시는 voice_realtime_test.py를 참고해요.

오디오 핵심 설정:

  • 마이크 입력: 16 kHz PCM16 (audio/pcm;rate=16000)
  • 스피커 출력: 24 kHz PCM16 (Vertex AI가 24 kHz로 오디오 반환)
  • Server VAD는 800ms 침묵 임계값으로 기본 활성화
# session.update with server VAD — the proxy ignores this for Vertex AI
# because VAD is already configured in the initial setup message.
await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "modalities": ["audio"],
        "turn_detection": {"type": "server_vad", "silence_duration_ms": 800},
    },
}))

도구 호출

import asyncio
import json
import websockets

PROXY_URL = "ws://localhost:4000/v1/realtime?model=vertex-gemini-live"

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a location.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"},
                    "unit": {"type": "string", "enum": ["fahrenheit", "celsius"]},
                },
                "required": ["location"],
            },
        },
    }
]

def get_weather(location: str, unit: str = "fahrenheit") -> dict:
    return {
        "location": location,
        "temperature": 72 if unit == "fahrenheit" else 22,
        "unit": unit,
        "conditions": "sunny",
    }

TOOL_FUNCTIONS = {"get_weather": get_weather}

async def main():
    async with websockets.connect(
        PROXY_URL,
        additional_headers={
            "Authorization": "Bearer sk-",
            "X-Serverless-Authorization": "Bearer sk-",
        },
    ) as ws:
        _ = json.loads(await ws.recv())  # session.created

        # Required for tool calling: send tools in session.update
        await ws.send(
            json.dumps(
                {
                    "type": "session.update",
                    "session": {
                        "instructions": "Use get_weather for weather questions.",
                        "modalities": ["audio"],
                        "tools": TOOLS,
                    },
                }
            )
        )

        await ws.send(
            json.dumps(
                {
                    "type": "conversation.item.create",
                    "item": {
                        "type": "message",
                        "role": "user",
                        "content": [
                            {"type": "input_text", "text": "What's the weather in San Francisco?"}
                        ],
                    },
                }
            )
        )
        await ws.send(json.dumps({"type": "response.create"}))

        async for raw in ws:
            ev = json.loads(raw)
            t = ev.get("type", "")

            if t == "response.text.delta":
                print(ev.get("delta", ""), end="", flush=True)
            elif t == "response.function_call_arguments.done":
                fn_name = ev.get("name", "")
                call_id = ev.get("call_id", "")
                args = json.loads(ev.get("arguments", "{}"))
                result = TOOL_FUNCTIONS[fn_name](**args)

                await ws.send(
                    json.dumps(
                        {
                            "type": "conversation.item.create",
                            "item": {
                                "type": "function_call_output",
                                "call_id": call_id,
                                "output": json.dumps(result),
                            },
                        }
                    )
                )
                await ws.send(json.dumps({"type": "response.create"}))
            elif t == "response.done":
                print("\n[done]")
                break
            elif t == "error":
                print(ev)
                break

if __name__ == "__main__":
    asyncio.run(main())

구성 + 실행

model_list:
  - model_name: vertex-gemini-live
    litellm_params:
      model: vertex_ai/gemini-live-2.5-flash-native-audio
      vertex_project: your-gcp-project-id
      vertex_location: us-central1

litellm_settings:
  # Required for tool calling with Gemini/Vertex Live:
  # defer setup until client sends session.update (with tools)
  gemini_live_defer_setup: true

litellm --config config.yaml --port 4000
python test_realtime_tool_calling.py

지원되는 OpenAI Realtime 이벤트

클라이언트 → Proxy (→ Vertex AI):

OpenAI 이벤트 비고
input_audio_buffer.append realtime_input.audio로 전달
conversation.item.create realtime_input.text로 전달
session.update 조용히 무시 — Vertex AI는 세션 중 재구성을 지원하지 않음
response.create 조용히 무시 — Vertex AI가 각 턴 후 자동으로 응답

Vertex AI → Proxy (→ 클라이언트):

내보내는 OpenAI 이벤트 Vertex AI 소스
session.created setupComplete 후 합성
response.text.delta serverContent.modelTurn.parts[].text
response.audio.delta serverContent.modelTurn.parts[].inlineData
response.audio_transcript.delta serverContent.outputTranscription.text
conversation.item.input_audio_transcription.completed serverContent.inputTranscription.text
response.done serverContent.turnComplete

제한 사항

  • session.update는 전달되지 않음 (Vertex AI는 연결당 하나의 setup 메시지만 허용)
  • 오디오 전사에는 초기 setup에서 outputAudioTranscription: {}를 설정해야 함 (LiteLLM이 자동 수행)

주의 사항

  • 도구 호출은 tools가 있는 session.update에 의존
  • session.update를 건너뛰면 도구 호출이 트리거되지 않음
  • gemini_live_defer_setup은 하위 호환을 위해 기본값 false

더 알아보기 (Learn more)

  • Gemini Live API 문서
  • LiteLLM Realtime API