Gemini 3.1 Flash Live 오디오 API — WebSocket 세션 설정부터 에러 핸들링까지

목차

• Gemini Live API WebSocket 연결 구조와 모델 스펙
• WebSocket 세션 생성 — Python과 JavaScript
  • Python SDK 세션 연결
  • JavaScript SDK 세션 연결
• 오디오 포맷과 청크 전송
  • Python에서 오디오 청크 전송
  • JavaScript에서 오디오 청크 전송
• 서버 응답 수신과 트랜스크립션 처리
• Barge-in과 VAD 처리
  • 자동 VAD vs 수동 VAD
  • Barge-in 처리 코드
• 음성 설정과 세션 제한
  • 음성과 다국어 설정
  • 세션 시간 제한
  • 가격 정책
• Gemini Live API 오디오 WebSocket 에러 핸들링과 재연결
  • 연결 끊김 대응 — Python
  • 연결 끊김 대응 — JavaScript
  • 오디오 포맷 불일치 디버깅
  • 세션 만료 처리
• 프로덕션 투입 전 체크리스트

Gemini Live API는 WebSocket(WSS) 위에서 오디오 청크를 양방향으로 주고받는 상태 유지형 API다. REST 요청-응답 방식과 달리, 한 번 연결을 맺으면 세션이 유지되는 동안 계속 스트리밍이 가능하다. gemini-3.1-flash-live-preview 모델에서 gemini live api 오디오 websocket 연결을 구현하려면 세션 생성, 오디오 포맷, Barge-in 처리, 세션 만료 대응까지 한 묶음으로 다뤄야 한다.

이 글은 Python SDK(google-genai)와 JavaScript SDK(@google/genai) 기준으로, 복붙 가능한 코드와 함께 전체 흐름을 정리한다.

Gemini Live API WebSocket 연결 구조와 모델 스펙

gemini live api 오디오 websocket 연결의 시작점은 모델 선택이다. 모델 ID는 gemini-3.1-flash-live-preview이며, 공식 모델 문서에 따르면 입력은 텍스트, 이미지, 오디오, 비디오를 모두 지원하고 출력은 텍스트와 오디오를 지원한다 (2026-04 기준).

항목	값
입력 토큰 한도	131,072
출력 토큰 한도	65,536
컨텍스트 윈도우 (네이티브 오디오)	128k 토큰
thinkingLevel 옵션	minimal, low, medium, high
thinkingLevel 기본값	minimal

기존 Gemini 모델과 다른 점이 있다. thinkingBudget 대신 thinkingLevel을 사용하며, Function calling은 동기(synchronous) 방식만 지원된다.

응답 모달리티는 세션당 `TEXT` 또는 `AUDIO` 중 하나만 선택할 수 있다. 한 세션에서 텍스트와 오디오 응답을 동시에 받는 건 안 된다. 모달리티를 바꾸려면 새 세션을 열어야 하므로, 세션 설계 시 가장 먼저 결정해야 할 사항이다.

WebSocket 세션 생성 — Python과 JavaScript

Live API의 핵심은 WSS 연결 위에서 양방향 스트리밍을 유지하는 것이다. Python과 JavaScript SDK 모두 WebSocket 연결을 추상화해서 제공하지만, 연결 방식과 에러 처리 패턴이 다르다.

Python SDK 세션 연결

Python SDK google-genai에서는 client.aio.live.connect()로 세션을 생성한다. 공식 SDK 가이드 기준 코드:

from google import genai

client = genai.Client(api_key="YOUR_API_KEY")

model = "gemini-3.1-flash-live-preview"
config = {"response_modalities": ["AUDIO"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        print("Session started")
        # 이 블록 안에서 오디오 송수신 처리

async with 블록이 끝나면 세션이 자동으로 닫힌다. response_modalities에 "AUDIO"를 넣으면 오디오 응답, "TEXT"를 넣으면 텍스트 응답을 받게 된다. 두 값을 동시에 넣을 수는 없다.

JavaScript SDK 세션 연결

JavaScript SDK @google/genai에서는 콜백 기반으로 동작한다:

import { GoogleGenAI } from "@google/genai";

const ai = new GoogleGenAI({ apiKey: "YOUR_API_KEY" });

const config = { responseModalities: ["AUDIO"] };

const session = await ai.live.connect({
  model: "gemini-3.1-flash-live-preview",
  callbacks: {
    onopen: function () {
      console.debug("Opened");
    },
    onmessage: function (message) {
      console.debug(message);
    },
    onerror: function (e) {
      console.debug("Error:", e.message);
    },
    onclose: function (e) {
      console.debug("Close:", e.reason);
    },
  },
  config: config,
});

Python SDK와 달리 async with로 자동 종료되지 않는다. onerror와 onclose 콜백에서 에러와 연결 종료를 직접 처리해야 하고, 재연결 로직도 이 콜백 안에서 구현해야 한다.

## 오디오 포맷과 청크 전송

Gemini Live API 오디오 WebSocket 통신에서 가장 자주 부딪히는 문제가 포맷 불일치다. 입력과 출력의 샘플레이트가 다르기 때문이다.

방향	포맷	샘플레이트	바이트 순서
입력 (마이크 → API)	raw 16-bit PCM	16kHz	little-endian
출력 (API → 스피커)	raw 16-bit PCM	24kHz	little-endian

마이크에서 16kHz로 녹음해서 보내고, 받은 오디오는 24kHz로 재생해야 한다. 브라우저에서는 AudioContext의 sampleRate를 24000으로 설정해서 재생하고, 서버 사이드에서는 pyaudio나 sounddevice에서 출력 샘플레이트를 24000으로 맞춰야 한다. 이 부분을 놓치면 오디오가 빠르게 또는 느리게 재생된다.

Python에서 오디오 청크 전송

from google import genai
from google.genai import types

client = genai.Client(api_key="YOUR_API_KEY")

model = "gemini-3.1-flash-live-preview"
config = {"response_modalities": ["AUDIO"]}

async def send_audio(session, chunk: bytes):
    await session.send_realtime_input(
        audio=types.Blob(
            data=chunk,
            mime_type="audio/pcm;rate=16000"
        )
    )

chunk는 raw PCM 바이트 데이터다. mime_type에 rate=16000을 반드시 명시해야 한다. 이 값이 누락되거나 틀리면 서버 측에서 오디오를 제대로 해석하지 못한다.

JavaScript에서 오디오 청크 전송

import { GoogleGenAI } from "@google/genai";

// session은 ai.live.connect()로 생성된 세션 객체
function sendAudioChunk(session, chunk) {
  session.sendRealtimeInput({
    audio: {
      data: Buffer.from(chunk).toString("base64"),
      mimeType: "audio/pcm;rate=16000",
    },
  });
}

JavaScript에서는 바이너리 데이터를 base64로 인코딩해서 보낸다. Node.js 환경이라면 Buffer.from(chunk).toString("base64")를 사용하면 되고, 브라우저 환경이라면 btoa()나 FileReader로 변환해야 한다.

마이크 입력을 일시정지할 때는 audioStreamEnd를 전송해야 한다. 이걸 빠뜨리면 서버가 오디오 스트림이 아직 진행 중인 것으로 판단하게 된다.

서버 응답 수신과 트랜스크립션 처리

단일 서버 이벤트에 여러 콘텐츠 파트가 동시에 포함될 수 있다. 오디오 데이터, 입력 트랜스크립션, 출력 트랜스크립션이 한 이벤트 안에 섞여서 오므로, 각 이벤트의 모든 파트를 순회하면서 처리해야 한다.

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="YOUR_API_KEY")

model = "gemini-3.1-flash-live-preview"
config = {"response_modalities": ["AUDIO"]}

playback_queue: asyncio.Queue[bytes] = asyncio.Queue()

async def drain_queue(q: asyncio.Queue) -> None:
    """asyncio.Queue에는 clear()가 없으므로 수동으로 비운다."""
    while not q.empty():
        try:
            q.get_nowait()
        except asyncio.QueueEmpty:
            break

async def receive_responses(session):
    async for response in session.receive():
        content = response.server_content
        if content is None:
            continue

        if content.model_turn:
            for part in content.model_turn.parts:
                if part.inline_data:
                    await playback_queue.put(part.inline_data.data)
                    # playback_queue에서 꺼내서 24kHz PCM으로 재생

        if content.input_transcription:
            user_text = content.input_transcription.text
            # 사용자 발화 텍스트

        if content.output_transcription:
            model_text = content.output_transcription.text
            # 모델 응답 텍스트

        if content.interrupted is True:
            await drain_queue(playback_queue)

`asyncio.Queue`에는 `clear()` 메서드가 없다. Barge-in으로 `content.interrupted`가 `True`일 때 재생 큐를 비우려면 `get_nowait()`을 반복 호출해서 남은 아이템을 모두 꺼내야 한다. 위 코드의 `drain_queue()` 함수가 이 처리를 담당한다.

각 필드의 역할:

• model_turn.parts — 모델이 생성한 오디오 데이터. inline_data.data에 raw PCM 바이트(24kHz)가 담긴다.
• input_transcription.text — 사용자가 말한 내용의 텍스트 변환. 자막 표시 용도.
• output_transcription.text — 모델이 말한 내용의 텍스트 변환. 대화 로깅 용도.
• interrupted — Barge-in 발생 여부. True면 재생 큐를 즉시 비워야 한다.

input_transcription과 output_transcription은 오디오 응답과 별도로 제공된다. UI에 실시간 자막을 띄우거나, 대화 내용을 데이터베이스에 저장하는 데 쓸 수 있다.

Barge-in과 VAD 처리

Barge-in은 모델이 응답하는 도중에 사용자가 말을 끊는 동작이다. Live API capabilities 문서에 따르면, VAD(Voice Activity Detection)가 사용자 음성을 감지하면 진행 중인 모델 생성이 자동 취소된다.

처리 흐름은 다음과 같다:

sequenceDiagram
    participant User as 사용자
    participant Client as 클라이언트
    participant Server as Gemini Live API

    Server->>Client: 오디오 응답 스트리밍 중
    User->>Client: 마이크 입력 시작
    Client->>Server: 오디오 청크 전송
    Server->>Server: VAD가 음성 감지
    Server->>Client: interrupted: true
    Client->>Client: 재생 큐 비우기 (drain_queue)
    Server->>Client: 새로운 응답 스트리밍 시작

자동 VAD vs 수동 VAD

기본 설정은 자동 VAD다. 서버가 입력 오디오에서 음성 구간을 자동으로 감지한다. 별도 설정 없이 오디오를 보내면 서버가 알아서 발화 시작과 끝을 판단한다.

수동 VAD는 클라이언트가 직접 발화 구간을 제어하는 방식이다. 노이즈가 많은 환경이나, 특정 버튼을 눌렀을 때만 입력을 받고 싶을 때 유용하다. 수동 모드에서는 send_realtime_input() 대신 send_client_content()를 써서 클라이언트가 발화 구간을 명시적으로 알려야 한다.

자동 VAD가 맞는 경우와 수동 VAD가 맞는 경우를 구분하면:

상황	권장 방식
일반 음성 대화 UI	자동 VAD
소음이 심한 환경	수동 VAD
Push-to-talk 방식	수동 VAD
연속 대화가 필요한 경우	자동 VAD

Barge-in 처리 코드

Barge-in이 감지되면 클라이언트 측에서 해야 할 일은 재생 중인 오디오를 즉시 멈추는 것이다. 서버는 interrupted: true를 보내면서 이전 응답 생성을 취소하고 새로운 응답을 준비한다.

import asyncio

playback_queue: asyncio.Queue[bytes] = asyncio.Queue()

async def drain_queue(q: asyncio.Queue) -> None:
    while not q.empty():
        try:
            q.get_nowait()
        except asyncio.QueueEmpty:
            break

async def handle_barge_in(content):
    if content.interrupted is True:
        # 1. 재생 큐 비우기
        await drain_queue(playback_queue)
        # 2. 현재 재생 중인 오디오 스트림 중단
        #    (pyaudio 등 오디오 출력 라이브러리 의존)
        # 3. 새로운 응답 수신 대기 상태로 전환

재생 큐를 비울 때 asyncio.Queue에는 clear() 메서드가 존재하지 않으므로 get_nowait()로 하나씩 꺼내야 한다. QueueEmpty 예외를 잡아서 루프를 종료하는 게 안전한 패턴이다.

음성과 다국어 설정

Gemini Live API는 다중 음성 프리셋을 제공한다. 세션 생성 시 speech_config에 음성을 지정할 수 있다:

from google import genai

client = genai.Client(api_key="YOUR_API_KEY")

config = {
    "response_modalities": ["AUDIO"],
    "speech_config": {
        "voice_config": {
            "prebuilt_voice_config": {
                "voice_name": "Kore"
            }
        }
    }
}

async def main():
    async with client.aio.live.connect(
        model="gemini-3.1-flash-live-preview",
        config=config
    ) as session:
        print("Session with Kore voice started")

음성 이름은 대소문자를 구분한다. 사용 가능한 음성 목록은 공식 모델 문서에서 확인할 수 있다. 한국어 입력도 가능하며, 시스템 인스트럭션에 응답 언어를 지정하면 해당 언어로 응답한다.

세션 시간 제한

Live API 세션에는 시간 제한이 있다. 세션이 만료되면 WebSocket 연결이 서버 측에서 종료된다. 클라이언트는 이 종료를 감지하고 새 세션을 열어야 한다. 세션 만료 직전에 별도 경고 이벤트가 오지 않으므로, 클라이언트 측에서 타이머를 걸어두거나 onclose 이벤트에서 재연결을 처리하는 게 일반적인 패턴이다.

가격 정책

gemini live api 오디오 websocket 세션의 과금은 입출력 토큰 단위로 이루어진다. 오디오 입력과 텍스트 입력의 토큰 단가가 다르고, 오디오 출력과 텍스트 출력의 단가도 다르다. 정확한 단가는 Google AI 가격 페이지에서 확인해야 한다. 세션이 열려 있는 것 자체로는 과금되지 않고, 실제 토큰이 오갈 때만 과금된다.

## Gemini Live API 오디오 WebSocket 에러 핸들링과 재연결

실시간 오디오 스트리밍에서 에러 핸들링은 선택이 아니다. WebSocket 연결은 네트워크 상태, 세션 만료, 포맷 불일치 등 여러 이유로 끊길 수 있다.

연결 끊김 대응 — Python

Python SDK에서는 async with 블록 바깥에서 재연결 루프를 감싸는 구조가 기본이다:

import asyncio
from google import genai

client = genai.Client(api_key="YOUR_API_KEY")

model = "gemini-3.1-flash-live-preview"
config = {"response_modalities": ["AUDIO"]}

MAX_RETRIES = 5

async def run_session():
    retries = 0
    while retries < MAX_RETRIES:
        try:
            async with client.aio.live.connect(
                model=model, config=config
            ) as session:
                retries = 0  # 연결 성공 시 카운터 리셋
                print("Connected")
                async for response in session.receive():
                    content = response.server_content
                    if content is None:
                        continue
                    if content.model_turn:
                        for part in content.model_turn.parts:
                            if part.inline_data:
                                pass  # 오디오 처리
        except Exception as e:
            retries += 1
            wait_time = min(2 ** retries, 30)
            print(f"Connection lost: {e}. Retry {retries}/{MAX_RETRIES} in {wait_time}s")
            await asyncio.sleep(wait_time)

    print("Max retries exceeded")

재연결 시 지수 백오프(exponential backoff)를 적용한다. 2 ** retries로 대기 시간을 늘리되 최대 30초로 제한한다. 연결에 성공하면 retries를 0으로 리셋해서, 안정적인 세션 중에 발생한 일시적 끊김에 대비한다.

연결 끊김 대응 — JavaScript

import { GoogleGenAI } from "@google/genai";

const ai = new GoogleGenAI({ apiKey: "YOUR_API_KEY" });
const MAX_RETRIES = 5;
let retries = 0;

async function connectWithRetry() {
  const config = { responseModalities: ["AUDIO"] };

  const session = await ai.live.connect({
    model: "gemini-3.1-flash-live-preview",
    callbacks: {
      onopen: function () {
        retries = 0;
        console.debug("Connected");
      },
      onmessage: function (message) {
        // 오디오 및 트랜스크립션 처리
      },
      onerror: function (e) {
        console.debug("Error:", e.message);
      },
      onclose: function (e) {
        if (retries < MAX_RETRIES) {
          retries++;
          const waitTime = Math.min(2 ** retries, 30) * 1000;
          console.debug(`Reconnecting in ${waitTime}ms...`);
          setTimeout(connectWithRetry, waitTime);
        }
      },
    },
    config: config,
  });

  return session;
}

JavaScript SDK에서는 onclose 콜백 안에서 setTimeout으로 재연결을 예약한다. onopen이 호출되면 retries를 리셋한다.

오디오 포맷 불일치 디버깅

오디오가 재생은 되는데 “다람쥐 목소리”처럼 들리거나 극단적으로 느린 경우, 십중팔구 샘플레이트 불일치다. 입력은 16kHz, 출력은 24kHz라는 점을 다시 확인해야 한다. 가장 흔한 실수 두 가지:

1. 출력 오디오를 16kHz로 재생 → 원래보다 느리고 낮은 목소리
2. 입력 오디오를 48kHz(브라우저 기본값)로 보냄 → 서버에서 3배 느린 음성으로 해석

브라우저의 AudioContext 기본 sampleRate는 보통 44100Hz 또는 48000Hz다. 마이크 입력을 캡처한 뒤 16kHz로 다운샘플링하는 과정이 반드시 필요하다.

세션 만료 처리

세션이 만료되면 서버가 WebSocket을 닫는다. 클라이언트 측에서는 이걸 일반적인 연결 끊김과 구분하기 어렵다. 실무에서 쓸 수 있는 전략:

• 클라이언트 타이머: 세션 시작 시점부터 타이머를 걸고, 만료 예상 시간 전에 선제적으로 새 세션을 연다.
• 대화 로그 저장: output_transcription으로 대화 내용을 계속 저장해두면, 새 세션 시작 시 이전 맥락을 시스템 인스트럭션으로 주입할 수 있다.
• 사용자 알림: 재연결 중이라는 UI 피드백을 즉시 보여줘야 사용자가 “앱이 죽었나?” 하고 이탈하는 걸 막을 수 있다.

프로덕션 투입 전 체크리스트

프로덕션에서 gemini live api 오디오 websocket 연결을 운영하려면, 개발 단계에서 잡기 어려운 엣지 케이스를 미리 점검해야 한다.

연결 안정성:

• 지수 백오프 재연결 로직이 구현되어 있는가
• 최대 재시도 횟수 초과 시 사용자에게 에러 UI를 보여주는가
• 세션 만료 전 선제적 재연결 타이머가 있는가

오디오 품질:

• 입력 오디오가 16kHz 16-bit PCM little-endian인가
• 출력 오디오를 24kHz로 재생하고 있는가
• 브라우저 환경이라면 마이크 입력 다운샘플링 로직이 있는가

Barge-in 처리:

• interrupted 이벤트 수신 시 재생 큐를 비우는가 (asyncio.Queue는 clear()가 없으므로 get_nowait() 반복으로 비워야 한다)
• 현재 재생 중인 오디오 스트림도 중단하는가

데이터 보존:

• input_transcription과 output_transcription을 저장하고 있는가
• 재연결 후 이전 대화 맥락을 새 세션에 주입하는가

전체 흐름을 다이어그램으로 정리하면:

flowchart TD
    A[클라이언트 시작] --> B[WebSocket 세션 생성]
    B --> C{연결 성공?}
    C -- 예 --> D[오디오 청크 전송 시작]
    C -- 아니오 --> E[지수 백오프 대기]
    E --> F{재시도 횟수 초과?}
    F -- 아니오 --> B
    F -- 예 --> G[에러 UI 표시]
    D --> H[서버 응답 수신]
    H --> I{interrupted?}
    I -- 예 --> J[재생 큐 비우기]
    J --> H
    I -- 아니오 --> K[오디오 재생 + 트랜스크립션 저장]
    K --> H
    H --> L{연결 끊김?}
    L -- 예 --> E
    L -- 아니오 --> H

gemini live api 오디오 websocket 연결의 기본 구조는 여기까지다. 다음 단계로는 Function Calling을 Live API 세션 안에서 동기 방식으로 연동하는 것을 다룰 수 있다. 외부 API를 호출해서 날씨, 예약, 검색 결과 등을 음성 응답에 반영하는 구조다. Vertex AI 환경에서 Live API를 서비스 계정으로 인증해서 배포하는 방법도 프로덕션 진입 시 필요한 주제다. 그리고 여러 사용자의 동시 세션을 관리하는 세션 풀링 패턴도 규모가 커지면 고민해야 할 영역이 된다.

관련 글

• Claude Code 자동화 시스템 구축 7단계: 스킬·훅·배치 실전 아키텍처