Gemini Realtime API - Google AI Studio

Gemini Realtime API - Google AI Studio

출처: 문서

본문

Proxy 사용법

config에 모델 추가

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

Proxy 시작

litellm --config /path/to/config.yaml
# RUNNING on http://0.0.0.0:4000

테스트

// test.js
const WebSocket = require("ws");
const url = "ws://0.0.0.0:4000/v1/realtime?model=openai-gemini-2.0-flash";
const ws = new WebSocket(url, {
  headers: {
    "api-key": `${LITELLM_API_KEY}`,
    "OpenAI-Beta": "realtime=v1",
  },
});

ws.on("open", function open() {
  console.log("Connected to server.");
  ws.send(JSON.stringify({
    type: "response.create",
    response: {
      modalities: ["text"],
      instructions: "Please assist the user.",
    },
  }));
});

ws.on("message", function incoming(message) {
  console.log(JSON.parse(message.toString()));
});

ws.on("error", function handleError(error) {
  console.error("Error: ", error);
});

이 스크립트를 node test.js로 실행해요.

Tool Calling

import asyncio
import json
import websockets

PROXY_URL = "ws://localhost:4000/v1/realtime?model=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-<your-litellm-api-key>",
            "X-Serverless-Authorization": "Bearer sk-<your-litellm-api-key>",
        },
    ) 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...")

config + 실행

model_list:
  - model_name: gemini-live
    litellm_params:
      model: gemini/gemini-2.5-flash-native-audio-latest
      api_key: os.environ/GEMINI_API_KEY

litellm_settings:
  # Required for tool calling with Gemini 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

제한 사항 (Limitations)

• 오디오 전사를 지원하지 않아요.
• 첫 session.update 이후의 세션 구성 업데이트는 무시돼요 (Gemini setup은 연결당 일회성).

주의 사항 (Precaution)

• 도구를 보내려면 먼저 session.update를 보내는 것이 아니면 tool calling이 동작하지 않아요. 해당 websocket 세션의 첫 번째 구성 메시지로 보내세요.
• gemini_live_defer_setup은 하위 호환성을 위해 기본적으로 false예요.

지원 OpenAI Realtime 이벤트

• session.created
• response.created
• response.output_item.added
• conversation.item.created
• response.content_part.added
• response.text.delta
• response.audio.delta
• response.text.done
• response.audio.done
• response.content_part.done
• response.output_item.done
• response.done

지원 Session 파라미터

session 파라미터는 Gemini Realtime 설정에 따라 지원돼요.

더 알아보기 (Learn more)

• LiteLLM 개요
• LiteLLM Python SDK 사용법
• LiteLLM Proxy Server 설정
• 지원되는 공급업체 전체 목록