단계별 튜토리얼: Gemini Multimodal Live API로 저지연 음성·영상 AI 어시스턴트 구축하기

Gemini Live API는 오디오 입력과 출력을 기본 지원하며, 엔드투엔드 아키텍처를 통해 ASR/TTS 중계 없이 지연 시간을 500ms 이내로 낮춥니다. 이 글에서는 이 API를 사용해 실제로 실시간 대화가 가능한 AI 어시스턴트를 만드는 방법을 소개합니다.

Gemini Multimodal Live API란?

먼저 개념부터 명확히 해보겠습니다. 기존 Gemini API는 어떻게 작동할까요? 텍스트를 보내면 텍스트로 답합니다. 단순하고 직관적입니다. 하지만 음성 상호작용을 구현하려면 ASR(음성 인식)과 TTS(음성 합성)를 직접 연결해야 하고, 여러 단계를 거치면서 지연 시간이 크게 늘어납니다.

Gemini Multimodal Live API의 차별점은 오디오 입력과 출력을 기본 지원한다는 것입니다. 즉, 마이크로 수집한 소리를 그대로 전송할 수 있고 응답도 순수 오디오 스트림으로 받기 때문에 중간에 별도의 형식 변환을 할 필요가 없습니다. 이러한 엔드투엔드 아키텍처는 지연 시간을 500ms 이내로 직접 낮춰 줍니다.

스마트 홈 프로젝트를 만들면서 이 기능을 사용해 본 적이 있습니다. 사용자가 “거실 조명을 조금 어둡게 해줘”라고 말하자 거의 말이 끝나는 순간 AI가 응답했습니다. 너무 자연스러워서 상대가 프로그램이라는 사실을 잊을 정도였습니다.

현재 지원되는 모델은 gemini-2.0-flash-native-audio-preview입니다. 이 버전 번호를 기억해 두세요. Google이 빠르게 개선하고 있으므로 업데이트를 정기적으로 확인하는 것이 좋습니다.

아키텍처 설계와 기술 선택

이제 시스템을 어떻게 구성할지 살펴보겠습니다. 제가 권하는 방식은 프런트엔드와 백엔드를 분리하는 아키텍처입니다. 이유는 간단합니다. API Key를 프런트엔드에 노출해서는 안 되기 때문입니다.

전체 데이터 흐름은 다음과 같습니다.

[브라우저] --WebSocket--> [Python 백엔드 프록시] --WebSocket--> [Gemini Live API]
    |                             |                              |
마이크 수집                  중계+비즈니스 로직                  AI 처리
스피커 재생                  VAD 감지/끼어들기 제어              오디오 생성

브라우저를 Gemini에 직접 연결하면 안 되는지 궁금할 수 있습니다. 기술적으로는 가능하지만, 그렇게 하려면 API Key를 JavaScript에 넣어야 합니다. 누구든 개발자 도구를 열어 키를 가져갈 수 있다는 뜻입니다. 저도 한 번 그런 실수를 했다가 다음 날 요금이 폭증했습니다. 뼈아픈 경험이었습니다.

따라서 기술 스택은 다음과 같이 정했습니다.

Web Audio API의 AudioWorklet은 꽤 유용합니다. 별도 스레드에서 오디오를 처리하므로 메인 스레드를 차단하지 않습니다. 뒤에서 구체적인 구현 코드를 살펴보겠습니다.

WebSocket 연결 설정과 세션 관리

이제 코드를 작성해 보겠습니다. 먼저 Gemini 서비스에 연결하는 방법부터 해결해야 합니다.

Railway - 현대적인 배포 플랫폼, 번거로운 운영 없이 몇 분 만에 라이브

지금 시작 →광고

Live API의 WebSocket 엔드포인트는 다음과 같습니다.

wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1alpha.GenerativeService.BidiGenerateContent?key=YOUR_API_KEY

여기서 v1alpha에 주목하세요. 아직 프리뷰 버전이므로 인터페이스가 바뀔 수 있습니다. 프로덕션 환경에서 사용할 때는 이 점을 염두에 두어야 합니다.

연결이 설정되면 가장 먼저 Setup 메시지를 보내 Gemini에 원하는 대화 방식을 알려 줍니다.

import asyncio
import json
import websockets

GEMINI_API_KEY = "your-api-key-here"
GEMINI_WS_URL = (
    f"wss://generativelanguage.googleapis.com/ws/"
    f"google.ai.generativelanguage.v1alpha.GenerativeService.BidiGenerateContent"
    f"?key={GEMINI_API_KEY}"
)

CONFIG = {
    "setup": {
        "model": "models/gemini-2.0-flash-native-audio-preview",
        "generation_config": {
            "response_modalities": ["AUDIO"],
            "speech_config": {
                "voice_config": {
                    "prebuilt_voice_config": {
                        "voice_name": "Charon"  # 선택 항목: Charon, Aoede 등
                    }
                }
            }
        },
        "system_instruction": {
            "parts": [{"text": "당신은 helpful한 AI 어시스턴트입니다. 간결하고 자연스럽게 답하세요."}]
        }
    }
}

async def connect():
    async with websockets.connect(GEMINI_WS_URL) as ws:
        # setup 설정 전송
        await ws.send(json.dumps(CONFIG))

        # setup complete 응답 대기
        response = await ws.recv()
        data = json.loads(response)

        if "setupComplete" in data:
            print("✅ 연결에 성공했습니다. 이제 대화를 시작할 수 있습니다")
            return ws
        else:
            raise Exception(f"Setup 실패: {data}")

여기에는 몇 가지 살펴볼 만한 매개변수가 있습니다.

• response_modalities: ['AUDIO']로 설정하면 음성 응답만 받습니다. 텍스트도 원한다면 ['AUDIO', 'TEXT']로 변경할 수 있습니다.
• voice_name: Gemini는 몇 가지 사전 설정 음성을 제공합니다. 저는 차분하게 들리는 Charon을 선호합니다.

연결이 끊어진 뒤 재연결할 때는 지수 백오프 전략을 권합니다. 처음부터 무작정 재시도하면 서비스에 과도한 부하를 줄 수 있습니다.

async def connect_with_retry(max_retries=5):
    for attempt in range(max_retries):
        try:
            return await connect()
        except Exception as e:
            wait_time = min(2 ** attempt, 30)  # 최대 30초 대기
            print(f"연결 실패 ({e}), {wait_time}초 후 재시도합니다...")
            await asyncio.sleep(wait_time)
    raise Exception("여러 번 재시도했지만 연결하지 못했습니다")

16kHz PCM 오디오 스트림 수집과 전송

연결을 준비했으니 이제 오디오를 어디서 가져와 어떻게 보낼지 해결할 차례입니다.

먼저 16kHz를 선택하는 이유를 설명하겠습니다. 사람 목소리의 주파수 범위는 일반적으로 85Hz에서 255Hz 사이입니다. 남성 목소리는 낮고 여성 목소리는 높은 편입니다. 나이퀴스트 샘플링 정리에 따르면 이론적으로는 8kHz면 충분합니다. 하지만 실제로 세부 정보를 어느 정도 보존하려면 16kHz가 음질과 데이터 크기의 균형을 맞추는 sweet spot입니다. Gemini도 이 샘플링 레이트를 권장합니다.

프런트엔드 수집 코드는 다음과 같습니다.

class AudioRecorder {
  constructor() {
    this.sampleRate = 16000;
    this.bufferSize = 1024;
    this.audioContext = null;
    this.workletNode = null;
    this.stream = null;
    this.onAudioData = null; // 콜백 함수
  }

  async start() {
    // 마이크 권한 요청
    this.stream = await navigator.mediaDevices.getUserMedia({
      audio: {
        sampleRate: 16000,
        channelCount: 1,
        echoCancellation: true,
        noiseSuppression: true
      }
    });

    // AudioContext를 생성하고 샘플링 레이트 강제 지정
    this.audioContext = new AudioContext({
      sampleRate: 16000
    });

    // AudioWorklet 프로세서 로드
    await this.audioContext.audioWorklet.addModule('pcm-processor.js');

    const source = this.audioContext.createMediaStreamSource(this.stream);
    this.workletNode = new AudioWorkletNode(this.audioContext, 'pcm-processor');

    // 오디오 데이터 처리
    this.workletNode.port.onmessage = (event) => {
      const float32Data = event.data;

      // Int16 PCM으로 변환
      const int16Data = this.float32ToInt16(float32Data);

      // Base64로 인코딩한 뒤 전송
      const base64Data = btoa(String.fromCharCode(...new Uint8Array(int16Data.buffer)));

      if (this.onAudioData) {
        this.onAudioData(base64Data);
      }
    };

    source.connect(this.workletNode);
    console.log('🎤 오디오 수집을 시작했습니다');
  }

  float32ToInt16(float32Array) {
    const int16Array = new Int16Array(float32Array.length);
    for (let i = 0; i < float32Array.length; i++) {
      // Float32 (-1.0 ~ 1.0) -> Int16 (-32768 ~ 32767)
      const s = Math.max(-1, Math.min(1, float32Array[i]));
      int16Array[i] = s < 0 ? s * 0x8000 : s * 0x7FFF;
    }
    return int16Array;
  }

  stop() {
    if (this.workletNode) {
      this.workletNode.disconnect();
    }
    if (this.audioContext) {
      this.audioContext.close();
    }
    if (this.stream) {
      this.stream.getTracks().forEach(track => track.stop());
    }
    console.log('🛑 오디오 수집을 중지했습니다');
  }
}

AudioWorklet에는 별도의 pcm-processor.js 파일이 필요합니다.

// pcm-processor.js
class PCMProcessor extends AudioWorkletProcessor {
  process(inputs, outputs, parameters) {
    const input = inputs[0];
    if (input && input[0]) {
      // 메인 스레드로 전송
      this.port.postMessage(input[0].slice());
    }
    return true; // 프로세서를 활성 상태로 유지
  }
}

registerProcessor('pcm-processor', PCMProcessor);

백엔드가 데이터를 받으면 Gemini로 전달해야 합니다.

async def send_audio(ws, base64_pcm_data):
    """오디오 데이터를 Gemini로 전송합니다"""
    message = {
        "realtime_input": {
            "media_chunks": [{
                "mime_type": "audio/pcm;rate=16000",
                "data": base64_pcm_data
            }]
        }
    }
    await ws.send(json.dumps(message))

여기서 주의할 점이 있습니다. 일부 브라우저의 getUserMedia는 지정한 sampleRate를 무시하고 실제로 44.1kHz나 48kHz를 반환할 수 있습니다. 안전하게 처리하려면 AudioContext에서 다시 리샘플링하거나 audiobuffer-to-wav 같은 서드파티 라이브러리를 사용하는 것이 좋습니다.

VAD 음성 활동 감지 구현

이제 또 다른 문제가 생깁니다. 아무 조건 없이 모든 오디오를 Gemini에 전송하면 아무도 말하지 않는 동안에도 데이터가 전달되어 대역폭과 비용이 모두 낭비됩니다. 이때 필요한 것이 VAD(Voice Activity Detection, 음성 활동 감지)입니다.

VAD의 역할은 단순합니다. 해당 오디오 구간에 사람이 말하고 있는지 판단합니다. 말할 때만 보내고, 아무도 말하지 않을 때는 전송을 멈춥니다.

Google이 오픈 소스로 공개한 WebRTC VAD를 권합니다. 가볍고 빠르며 성능도 좋습니다. Python에는 webrtcvad라는 래퍼 라이브러리가 있습니다.

import webrtcvad
import collections
import numpy as np

class VADProcessor:
    def __init__(self, aggressiveness=2, frame_duration_ms=20):
        """
        aggressiveness: 0~3, 높을수록 엄격함(음성을 무음으로 판단하기 쉬움)
        frame_duration_ms: 10, 20, or 30
        """
        self.vad = webrtcvad.Vad(aggressiveness)
        self.frame_duration_ms = frame_duration_ms
        self.sample_rate = 16000

        # 평활화를 위한 링 버퍼
        self.ring_buffer = collections.deque(maxlen=30)  # 600ms
        self.triggered = False

    def process_frame(self, pcm_bytes):
        """
        오디오 프레임 하나를 처리하고 전송 여부를 반환합니다
        """
        is_speech = self.vad.is_speech(pcm_bytes, self.sample_rate)

        if not self.triggered:
            # 비활성 상태: 음성 프레임 누적
            self.ring_buffer.append((pcm_bytes, is_speech))
            num_voiced = sum(1 for _, speech in self.ring_buffer if speech)

            # 프레임의 90%가 음성이면 활성화
            if num_voiced > 0.9 * self.ring_buffer.maxlen:
                self.triggered = True
                # 버퍼의 데이터도 함께 전송
                return b''.join([f for f, _ in self.ring_buffer])
            return None
        else:
            # 활성 상태
            if is_speech:
                self.ring_buffer.append((pcm_bytes, True))
                return pcm_bytes
            else:
                self.ring_buffer.append((pcm_bytes, False))
                num_unvoiced = sum(1 for _, speech in self.ring_buffer if not speech)

                # 90%가 무음이면 비활성화
                if num_unvoiced > 0.9 * self.ring_buffer.maxlen:
                    self.triggered = False
                    self.ring_buffer.clear()
                return pcm_bytes

사용 방법은 대략 다음과 같습니다.

Alibaba Cloud - 개발자 클라우드 1순위, Lighthouse 서버 15% 할인

15% 할인 받기 →광고

vad = VADProcessor(aggressiveness=2)

async def handle_client_audio(websocket, gemini_ws):
    async for message in websocket:
        data = json.loads(message)

        if 'audio' in data:
            pcm_bytes = base64.b64decode(data['audio'])

            # VAD 감지
            result = vad.process_frame(pcm_bytes)

            if result:
                # 음성이 있으면 Gemini로 전달
                await send_audio(gemini_ws, base64.b64encode(result).decode())

aggressiveness 매개변수는 조정이 다소 까다롭습니다. 너무 낮으면 작은 배경 소음까지 음성으로 판단하고, 너무 높으면 작은 목소리를 놓칠 수 있습니다. 제 경험으로는 2부터 시작해 실제 환경에 맞게 미세 조정하는 것이 좋습니다.

배포 환경에 webrtcvad를 설치할 수 없다면 간단한 에너지 임계값 감지를 대안으로 사용할 수도 있습니다.

// 프런트엔드 대안: RMS 에너지 기반의 간단한 감지
function detectVoiceActivity(audioData, threshold = 0.015) {
    const sum = audioData.reduce((acc, val) => acc + val * val, 0);
    const rms = Math.sqrt(sum / audioData.length);
    return rms > threshold;
}

자연스러운 끼어들기(Barge-in) 기능 구현

일부 음성 어시스턴트와 대화할 때 답답함을 느낀 적이 있을 겁니다. 일단 긴 설명을 시작하면 끝날 때까지 기다려야 하고 중간에 끼어들 수도 없습니다.

Barge-in, 즉 끼어들기 기능은 바로 이 문제를 해결합니다. AI가 말하는 중에도 사용자가 바로 말을 시작해 중단시킬 수 있으며, AI는 현재 출력을 즉시 멈추고 사용자의 말을 듣습니다.

좋은 소식은 Gemini Live API가 이 기능을 기본 지원하며 구현도 꽤 지능적이라는 점입니다. 설정에서 자동 활동 감지만 활성화하면 됩니다.

CONFIG = {
    "setup": {
        "model": "models/gemini-2.0-flash-native-audio-preview",
        "generation_config": {
            "response_modalities": ["AUDIO"],
        },
        "realtime_input_config": {
            "automatic_activity_detection": {
                "disabled": False,
                "start_of_speech_sensitivity": "START_SENSITIVITY_HIGH",
                "end_of_speech_sensitivity": "END_SENSITIVITY_LOW"
            }
        }
    }
}

sensitivity 설정에는 약간의 요령이 필요합니다.

• start_of_speech_sensitivity를 HIGH로 설정하면 AI가 사용자의 발화 시작에 더 민감하게 반응해 끼어들기가 쉽게 활성화됩니다.
• end_of_speech_sensitivity를 LOW로 설정하면 사용자가 실제로 말을 마쳤는지 AI가 조금 더 기다린 뒤 응답하므로 오판을 줄일 수 있습니다.

클라이언트에서는 interrupted 이벤트를 수신한 뒤 즉시 재생을 중지하면 됩니다.

class GeminiClient {
  constructor() {
    this.audioQueue = [];
    this.isPlaying = false;
    this.currentSource = null;
  }

  async handleMessage(event) {
    const message = JSON.parse(event.data);

    // 중단 신호 처리
    if (message.server_content?.interrupted) {
      console.log('⚡ 사용자가 끼어들었습니다. 재생을 중지합니다');
      this.stopPlayback();
      return;
    }

    // AI가 반환한 오디오 처리
    if (message.server_content?.model_turn) {
      const parts = message.server_content.model_turn.parts;

      for (const part of parts) {
        if (part.inline_data?.mime_type.startsWith('audio/')) {
          const audioData = base64ToArrayBuffer(part.inline_data.data);
          this.queueAudio(audioData);
        }
      }
    }
  }

  stopPlayback() {
    // 재생 큐 비우기
    this.audioQueue = [];
    this.isPlaying = false;

    // 현재 재생 중인 오디오 중지
    if (this.currentSource) {
      try {
        this.currentSource.stop();
      } catch (e) {
        // 이미 중지되었을 수 있음
      }
      this.currentSource = null;
    }
  }

  async queueAudio(audioData) {
    this.audioQueue.push(audioData);
    if (!this.isPlaying) {
      this.playNext();
    }
  }

  async playNext() {
    if (this.audioQueue.length === 0) {
      this.isPlaying = false;
      return;
    }

    this.isPlaying = true;
    const audioData = this.audioQueue.shift();

    // 디코딩 후 재생
    const audioBuffer = await this.audioContext.decodeAudioData(audioData.slice());
    this.currentSource = this.audioContext.createBufferSource();
    this.currentSource.buffer = audioBuffer;
    this.currentSource.connect(this.audioContext.destination);

    this.currentSource.onended = () => {
      this.playNext();
    };

    this.currentSource.start();
  }
}

여기서 주의할 점이 하나 있습니다. 오디오가 이미 자연스럽게 재생을 마쳤다면 stop() 메서드가 예외를 던질 수 있습니다. 그래서 콘솔이 오류로 가득 차지 않도록 try-catch를 추가했습니다.

성능 최적화와 지연 시간 제어

마지막으로 시스템 지연 시간을 최소화하는 방법을 살펴보겠습니다.

먼저 지연 시간이 어디서 발생하는지 알아야 합니다.

1. 네트워크 전송: 데이터 패킷이 브라우저에서 서버를 거쳐 Gemini까지 왕복하는 시간
2. 오디오 인코딩과 디코딩: PCM 압축 및 압축 해제 시간(다만 PCM 자체가 무손실이라 이 부분의 오버헤드는 매우 작습니다)
3. 버퍼 누적: 원활한 재생을 위해 설정한 버퍼 깊이

이 요소들을 고려해 제가 사용하는 최적화 방법은 다음과 같습니다.

버퍼 깊이 줄이기

재생 버퍼는 너무 크게 설정하지 말고 필요한 만큼만 사용하세요. 저는 보통 100~200ms를 사용합니다.

// 작은 버퍼 설정
const audioContext = new AudioContext({
  sampleRate: 16000,
  latencyHint: 'interactive'  // 저지연 모드
});

적응형 비트 전송률(실제로는 주로 적응형 버퍼)

KiloClaw - 기업용 매니지드 OpenClaw, AI Agent 오케스트레이션 및 스마트 라우팅

시작하기 →광고

네트워크 변동이 크다고 감지되면 버퍼를 조금 늘리고, 네트워크가 안정적일 때는 다시 줄일 수 있습니다.

로컬 에코 제거

사용자가 헤드폰 대신 스피커를 켜 두면 AI의 목소리가 마이크로 다시 들어가 순환이 발생합니다. 다행히 getUserMedia에는 에코 제거 기능이 포함되어 있습니다.

navigator.mediaDevices.getUserMedia({
  audio: {
    echoCancellation: true,
    noiseSuppression: true,
    autoGainControl: true
  }
})

지표 모니터링

최적화가 효과가 있는지 어떻게 알 수 있을까요? Performance API로 측정 지점을 기록할 수 있습니다.

// 지연 시간 지표 기록
class LatencyMonitor {
  constructor() {
    this.metrics = [];
  }

  recordSendTime() {
    this.lastSendTime = performance.now();
  }

  recordReceiveTime() {
    const latency = performance.now() - this.lastSendTime;
    this.metrics.push(latency);

    // 최근 100개 기록 유지
    if (this.metrics.length > 100) {
      this.metrics.shift();
    }

    // 평균 지연 시간 계산
    const avg = this.metrics.reduce((a, b) => a + b, 0) / this.metrics.length;
    console.log(`📊 평균 지연 시간: ${avg.toFixed(2)}ms`);
  }
}

테스트 환경에서 측정한 결과는 대략 다음과 같았습니다.

• 엔드투엔드 지연 시간: 300~500ms(네트워크 상태에 따라 달라짐)
• 첫 패킷 응답 시간: 200~400ms
• 연속 대화 지연 시간: 150~300ms

지연 시간이 이보다 눈에 띄게 길다면 다음 체크리스트를 따라 확인해 보세요.

• WebSocket 연결이 HTTPS/WSS를 사용하고 있나요? HTTP에는 추가 오버헤드가 발생합니다.
• 서버가 어디에 배포되어 있나요? Google 데이터 센터와 가까울수록 좋습니다.
• VAD 감지로 인해 지연 시간이 너무 많이 늘어나지는 않았나요? 프레임 길이를 줄여 보세요.
• 프런트엔드 재생 버퍼를 너무 크게 설정하지 않았나요?

오디오 컨텍스트와 관련된 또 다른 주의점도 있습니다. Chrome은 사용자가 상호작용한 뒤에만 소리 재생을 허용합니다. 따라서 페이지가 열리자마자 자동으로 재생하지 말고 ‘대화 시작’ 버튼을 추가하세요.

마무리

여기까지 Gemini Live API 애플리케이션을 개발하는 전체 과정을 살펴봤습니다. 기본 개념과 아키텍처 설계부터 WebSocket 연결, 오디오 수집, VAD 감지, 끼어들기 기능, 마지막 성능 최적화까지 제가 직접 겪은 시행착오를 단계마다 최대한 공유했습니다.

솔직히 실시간 음성 상호작용 분야는 여전히 빠르게 발전하고 있으며 Gemini Live API 자체도 계속 업데이트되고 있습니다. 하지만 이 기본 아키텍처는 충분히 검증을 견딜 수 있다고 생각합니다. 적어도 제 프로젝트에서는 몇 달 동안 안정적으로 작동했습니다.

실제 개발 과정에서 문제가 생기면 언제든 함께 이야기해 보세요. 기술은 혼자 탐색하는 것보다 여러 사람이 함께 논의할 때 훨씬 빠르게 발전할 수 있으니까요.