Gemini 3.1 Flash Live 완전 가이드 — STT+LLM+TTS 파이프라인을 단일 WebSocket으로

음성 AI 에이전트를 만들 때마다 세 가지 서비스를 붙여야 했습니다. STT, LLM, TTS. Gemini 3.1 Flash Live는 이 전체를 하나의 WebSocket 연결로 교체합니다.

[핵심 요약]
→ 출시: 2026년 3월 26일 (Gemini Live API)
→ 모델 ID: gemini-3.1-flash-live-preview
→ 핵심: 오디오 in → 오디오 out, STT/TTS 별도 불필요
→ 레이턴시: 200ms 이하 — 실제 대화 속도
→ 방식: 양방향 WebSocket 스트리밍 (REST 아님)
→ 인터럽션: 사용자가 말 끊으면 모델이 멈추고 들음
→ 언어: 90개+ 지원, 멀티링구얼 실시간 전환
→ 툴 호출: Function Calling 지원 (단, 현재 블로킹 방식)
→ 멀티모달: 오디오 + 비디오 동시 처리 가능
→ 벤치마크: ComplexFuncBench Audio 90.8% (1위)
→ 가격: Google AI Studio 무료 티어 사용 가능

 

---

기존 파이프라인 vs Flash Live

기존 음성 AI 에이전트를 만들 때 구조입니다.

기존 파이프라인:
사용자 음성
  ↓ (1) Whisper/Google STT — 300~500ms
텍스트
  ↓ (2) LLM 처리 — 500~1500ms
응답 텍스트
  ↓ (3) TTS 합성 — 200~500ms
응답 음성
─────────────────────────────
총 레이턴시: 1~2.5초
관리 서비스: 3개
비용: 3개 API 요금 합산
실패 포인트: 3개

Gemini 3.1 Flash Live:
사용자 음성
  ↓ (1) WebSocket 스트리밍 — 50~200ms
응답 음성
─────────────────────────────
총 레이턴시: < 200ms
관리 서비스: 1개
비용: 단일 API
실패 포인트: 1개

[핵심 차이]
→ 기존: 텍스트 중간 단계 필수 (음성 → 텍스트 → 음성)
→ Flash Live: 오디오 네이티브 처리 (음성 → 음성 직접)
→ 인터럽션: 기존은 VAD가 침묵 감지 후 처리, Flash Live는 실시간 감지
→ 음색 유지: 기존 TTS는 세션마다 음색 달라질 수 있음, Flash Live는 일관성

---

실전 1 — 설치 및 기본 WebSocket 연결

# 설치
pip install google-generativeai pyaudio numpy

# API 키 설정
export GEMINI_API_KEY="your-api-key"
# Google AI Studio (aistudio.google.com) → API Key 발급

# 가장 간단한 Flash Live 세션 (텍스트 입출력 테스트)
import asyncio
from google import genai

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

model = "gemini-3.1-flash-live-preview"
config = {
    "response_modalities": ["AUDIO"],  # 오디오 응답
    "speech_config": {
        "voice_config": {
            "prebuilt_voice_config": {
                "voice_name": "Kore"  # 30개 보이스 중 선택
            }
        }
    }
}

async def basic_session():
    async with client.aio.live.connect(
        model=model,
        config=config
    ) as session:
        print("세션 연결됨")

        # 텍스트로 먼저 테스트
        await session.send(
            input="안녕하세요! 오늘 날씨 어때요?",
            end_of_turn=True
        )

        # 오디오 응답 수신
        async for response in session.receive():
            if response.data:
                # PCM16 오디오 데이터
                audio_data = response.data
                print(f"오디오 수신: {len(audio_data)} bytes")
                # 여기서 오디오 재생 처리

asyncio.run(basic_session())

[보이스 선택]
→ Kore: 권위 있고 차분한 여성
→ Puck: 밝고 경쾌한 중성
→ Charon: 깊고 안정적인 남성
→ Aoede: 부드럽고 따뜻한 여성
→ Fenrir: 활기차고 젊은 남성
→ Leda, Orus 등 30개 프리셋

---

실전 2 — 실시간 마이크 입력 음성 에이전트

실제로 마이크에서 입력받고 스피커로 출력하는 완전한 음성 에이전트입니다.

import asyncio
import queue
import threading
import numpy as np
import pyaudio
from google import genai

# 오디오 설정
INPUT_SAMPLE_RATE = 16000   # 마이크: 16kHz (Flash Live 요구사항)
OUTPUT_SAMPLE_RATE = 24000  # 스피커: 24kHz (Flash Live 출력)
CHUNK_SIZE = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1

class AudioHandler:
    def __init__(self):
        self.pa = pyaudio.PyAudio()
        self.input_queue = queue.Queue()
        self.output_queue = queue.Queue()
        self.running = False

    def start_input_stream(self):
        """마이크 오디오 캡처 시작"""
        def callback(in_data, frame_count, time_info, status):
            if self.running:
                self.input_queue.put(in_data)
            return (None, pyaudio.paContinue)

        self.input_stream = self.pa.open(
            format=FORMAT,
            channels=CHANNELS,
            rate=INPUT_SAMPLE_RATE,
            input=True,
            frames_per_buffer=CHUNK_SIZE,
            stream_callback=callback
        )
        self.input_stream.start_stream()

    def start_output_stream(self):
        """스피커 오디오 출력 시작"""
        self.output_stream = self.pa.open(
            format=FORMAT,
            channels=CHANNELS,
            rate=OUTPUT_SAMPLE_RATE,
            output=True,
            frames_per_buffer=CHUNK_SIZE
        )

    def play_audio(self, data: bytes):
        self.output_stream.write(data)

    def stop(self):
        self.running = False
        if hasattr(self, 'input_stream'):
            self.input_stream.stop_stream()
            self.input_stream.close()
        if hasattr(self, 'output_stream'):
            self.output_stream.close()
        self.pa.terminate()


class VoiceAgent:
    def __init__(self, api_key: str, system_prompt: str = ""):
        self.client = genai.Client(api_key=api_key)
        self.audio = AudioHandler()
        self.system_prompt = system_prompt

    async def run(self):
        config = {
            "response_modalities": ["AUDIO"],
            "speech_config": {
                "voice_config": {
                    "prebuilt_voice_config": {
                        "voice_name": "Kore"
                    }
                }
            }
        }

        if self.system_prompt:
            config["system_instruction"] = self.system_prompt

        async with self.client.aio.live.connect(
            model="gemini-3.1-flash-live-preview",
            config=config
        ) as session:
            self.audio.running = True
            self.audio.start_input_stream()
            self.audio.start_output_stream()

            print("음성 에이전트 시작 (Ctrl+C로 종료)")

            # 마이크 → API 전송 태스크
            async def send_audio():
                while self.audio.running:
                    if not self.audio.input_queue.empty():
                        chunk = self.audio.input_queue.get()
                        await session.send(
                            input={"data": chunk, "mime_type": "audio/pcm;rate=16000"}
                        )
                    await asyncio.sleep(0.01)

            # API → 스피커 수신 태스크
            async def receive_audio():
                async for response in session.receive():
                    if response.data:
                        self.audio.play_audio(response.data)

            # 병렬 실행
            await asyncio.gather(
                send_audio(),
                receive_audio()
            )

    def stop(self):
        self.audio.stop()


# 실행
async def main():
    agent = VoiceAgent(
        api_key="YOUR_GEMINI_API_KEY",
        system_prompt="""
        당신은 친절한 AI 음성 어시스턴트입니다.
        짧고 명확하게 대답하세요.
        한국어로 대화합니다.
        """
    )

    try:
        await agent.run()
    except KeyboardInterrupt:
        agent.stop()
        print("\n종료됨")

asyncio.run(main())

[오디오 포맷 주의사항]
→ 입력: PCM16, 16kHz, 모노 — 필수 (다른 포맷은 오류)
→ 출력: PCM16, 24kHz (세션 메타데이터에서 확인)
→ 브라우저 Web Audio API: 기본 32bit float → Int16으로 변환 필요
→ pyaudio 없는 경우: sounddevice 라이브러리로 대체 가능

---

실전 3 — Function Calling (툴 호출) 연동

음성 대화 중에 외부 API를 호출할 수 있습니다.

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

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

# 툴 정의
tools = [
    {
        "function_declarations": [
            {
                "name": "get_weather",
                "description": "특정 도시의 현재 날씨를 가져옵니다",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "날씨를 조회할 도시 이름"
                        }
                    },
                    "required": ["city"]
                }
            },
            {
                "name": "search_product",
                "description": "상품을 검색하고 가격을 조회합니다",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "검색할 상품명"
                        }
                    },
                    "required": ["query"]
                }
            }
        ]
    }
]

# 툴 실행 함수
def execute_tool(name: str, args: dict) -> str:
    if name == "get_weather":
        city = args.get("city", "서울")
        # 실제로는 날씨 API 호출
        return json.dumps({
            "city": city,
            "temperature": "18°C",
            "condition": "맑음",
            "humidity": "60%"
        }, ensure_ascii=False)

    elif name == "search_product":
        query = args.get("query", "")
        # 실제로는 쇼핑 API 호출
        return json.dumps({
            "results": [
                {"name": f"{query} 상품1", "price": "15,000원"},
                {"name": f"{query} 상품2", "price": "23,000원"},
            ]
        }, ensure_ascii=False)

    return json.dumps({"error": "알 수 없는 툴"})


async def voice_agent_with_tools():
    config = {
        "response_modalities": ["AUDIO"],
        "tools": tools,
        "speech_config": {
            "voice_config": {
                "prebuilt_voice_config": {"voice_name": "Kore"}
            }
        },
        "system_instruction": """
        당신은 쇼핑과 날씨 정보를 제공하는 음성 어시스턴트입니다.
        사용자가 날씨나 상품을 물으면 툴을 사용해서 정확한 정보를 제공하세요.
        """
    }

    async with client.aio.live.connect(
        model="gemini-3.1-flash-live-preview",
        config=config
    ) as session:
        print("툴 연동 음성 에이전트 시작")

        # 텍스트로 테스트 (실제로는 마이크 입력)
        await session.send(
            input="서울 날씨 어때요?",
            end_of_turn=True
        )

        async for response in session.receive():
            # 툴 호출 처리
            if hasattr(response, 'tool_call') and response.tool_call:
                for call in response.tool_call.function_calls:
                    print(f"툴 호출: {call.name}({call.args})")

                    # 툴 실행
                    result = execute_tool(call.name, dict(call.args))

                    # 결과 반환 (주의: 현재 블로킹 방식)
                    await session.send(
                        input=types.LiveClientToolResponse(
                            function_responses=[
                                types.FunctionResponse(
                                    id=call.id,
                                    name=call.name,
                                    response={"result": result}
                                )
                            ]
                        )
                    )

            # 오디오 응답
            if response.data:
                print(f"오디오 응답 수신: {len(response.data)} bytes")

asyncio.run(voice_agent_with_tools())

[Function Calling 주의사항]
→ 현재 블로킹 방식: 툴 실행 완료까지 대화 중단됨
→ 이전 2.5 Flash의 비동기 툴 호출 → 3.1에서 아직 미지원
→ 복잡한 툴 체인: 응답 시간 길어질 수 있음
→ 해결책: 빠른 API만 툴로 등록, 느린 건 백그라운드 처리

---

실전 4 — 브라우저 연동 (JavaScript/WebSocket)

프론트엔드에서 직접 WebSocket으로 연결하는 방법입니다.

// 브라우저 오디오 캡처 및 Flash Live 연결
class GeminiVoiceClient {
    constructor(apiKey) {
        this.apiKey = apiKey;
        this.wsUrl = `wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key=${apiKey}`;
        this.ws = null;
        this.audioContext = null;
        this.mediaStream = null;
    }

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

        // WebSocket 연결
        this.ws = new WebSocket(this.wsUrl);

        this.ws.onopen = () => {
            // 세션 초기화
            this.ws.send(JSON.stringify({
                setup: {
                    model: "models/gemini-3.1-flash-live-preview",
                    generation_config: {
                        response_modalities: ["AUDIO"],
                        speech_config: {
                            voice_config: {
                                prebuilt_voice_config: {
                                    voice_name: "Kore"
                                }
                            }
                        }
                    },
                    system_instruction: "당신은 친절한 AI 어시스턴트입니다."
                }
            }));

            // 마이크 스트리밍 시작
            this.startAudioCapture();
        };

        this.ws.onmessage = (event) => {
            const data = JSON.parse(event.data);

            // 오디오 응답 재생
            if (data.serverContent?.modelTurn?.parts) {
                for (const part of data.serverContent.modelTurn.parts) {
                    if (part.inlineData?.mimeType?.startsWith('audio/')) {
                        this.playAudio(part.inlineData.data);
                    }
                }
            }
        };
    }

    startAudioCapture() {
        this.audioContext = new AudioContext({ sampleRate: 16000 });
        const source = this.audioContext.createMediaStreamSource(this.mediaStream);

        // AudioWorklet으로 PCM16 변환
        const processor = this.audioContext.createScriptProcessor(1024, 1, 1);
        processor.onaudioprocess = (e) => {
            const input = e.inputBuffer.getChannelData(0);

            // Float32 → Int16 변환
            const pcm16 = new Int16Array(input.length);
            for (let i = 0; i < input.length; i++) {
                pcm16[i] = Math.max(-32768, Math.min(32767, input[i] * 32768));
            }

            // Base64 인코딩 후 전송
            const base64 = btoa(String.fromCharCode(...new Uint8Array(pcm16.buffer)));
            this.ws.send(JSON.stringify({
                realtime_input: {
                    media_chunks: [{
                        data: base64,
                        mime_type: "audio/pcm;rate=16000"
                    }]
                }
            }));
        };

        source.connect(processor);
        processor.connect(this.audioContext.destination);
    }

    playAudio(base64Data) {
        // Base64 디코딩 후 재생
        const binary = atob(base64Data);
        const bytes = new Uint8Array(binary.length);
        for (let i = 0; i < binary.length; i++) {
            bytes[i] = binary.charCodeAt(i);
        }

        const audioContext = new AudioContext({ sampleRate: 24000 });
        const pcm16 = new Int16Array(bytes.buffer);
        const float32 = new Float32Array(pcm16.length);

        for (let i = 0; i < pcm16.length; i++) {
            float32[i] = pcm16[i] / 32768;
        }

        const buffer = audioContext.createBuffer(1, float32.length, 24000);
        buffer.getChannelData(0).set(float32);

        const source = audioContext.createBufferSource();
        source.buffer = buffer;
        source.connect(audioContext.destination);
        source.start();
    }

    stop() {
        if (this.ws) this.ws.close();
        if (this.mediaStream) {
            this.mediaStream.getTracks().forEach(t => t.stop());
        }
    }
}

// 사용
const client = new GeminiVoiceClient("YOUR_API_KEY");
document.getElementById('startBtn').onclick = () => client.start();
document.getElementById('stopBtn').onclick = () => client.stop();

[브라우저 연동 주의사항]
→ API 키를 프론트에 노출하면 안 됨
→ 프로덕션: 에페머럴 토큰 사용 (서버에서 단기 토큰 발급)
→ 에페머럴 토큰 발급: Google AI Studio → API → Ephemeral Tokens
→ 브라우저 기본 AudioContext: 48kHz → 16kHz로 다운샘플링 필수
→ HTTPS 필수: getUserMedia는 보안 컨텍스트에서만 동작

---

실전 5 — 에페머럴 토큰으로 보안 강화

브라우저에 API 키를 노출하지 않는 방법입니다.

# 서버에서 단기 토큰 발급 (FastAPI 예시)
from fastapi import FastAPI
from google import genai
import time

app = FastAPI()
client = genai.Client(api_key="YOUR_SERVER_API_KEY")

@app.post("/api/get-token")
async def get_ephemeral_token():
    """
    클라이언트에게 단기 토큰 발급
    토큰은 1분간만 유효
    """
    token_response = await client.auth_tokens.create(
        config={
            "model": "gemini-3.1-flash-live-preview",
            "ttl": "60s",  # 1분 유효
            "live_connect_constraints": {
                "model": "gemini-3.1-flash-live-preview"
            }
        }
    )

    return {
        "token": token_response.name,
        "expires_at": int(time.time()) + 60
    }

// 브라우저: 서버에서 토큰 받아 연결
async function startWithEphemeralToken() {
    // 서버에서 단기 토큰 받기
    const { token } = await fetch('/api/get-token', {
        method: 'POST'
    }).then(r => r.json());

    // 에페머럴 토큰으로 WebSocket 연결 (API 키 대신)
    const wsUrl = `wss://generativelanguage.googleapis.com/ws/...?key=${token}`;
    const ws = new WebSocket(wsUrl);
    // ... 이하 동일
}

---

경쟁사 비교

               Gemini 3.1 Flash Live   GPT-4o Voice   ElevenLabs Conversational
방식:           오디오 네이티브         오디오 네이티브  STT+LLM+TTS
레이턴시:       < 200ms                < 300ms         300~500ms
언어:           90개+                  50개+           30개+
Function Call:  ✅ (블로킹)            ✅              제한적
비디오 입력:    ✅                     ✅              ❌
인터럽션:       ✅                     ✅              ✅
가격:           AI Studio 무료 티어     유료             유료
워터마크:       ✅ SynthID 자동         ❌              ❌
ComplexFuncBench: 90.8%               미공개           미공개

---

마무리

✅ Gemini 3.1 Flash Live 써야 할 때
→ 음성 AI 에이전트 (고객 서비스, 인터랙티브 앱)
→ 실시간 대화가 필요한 서비스 (200ms 이하 레이턴시)
→ 다국어 음성 처리 (90개+ 언어 동시 지원)
→ 음성 + 비전 동시 처리 (화면 보면서 대화)
→ 기존 STT+LLM+TTS 파이프라인 단순화
→ Function Calling이 필요한 음성 에이전트

❌ 다른 방식이 나을 때
→ 비동기 툴 호출 필요: 현재 블로킹 방식 — 2.5 Flash 유지 권장
→ 프리레코딩 음성 처리: 표준 Gemini API가 더 적합
→ 4K 영상 + 음성 동시: Veo 3.1 Pro 조합 고려
→ 음성 합성만 필요: Gemini 3.1 Flash TTS가 더 저렴
→ 즉각적인 스트리밍 없이 배치 처리: Flash Live 오버헤드

 

---

관련 글:

 

 

Gemini 3.1 Flash TTS 완전 가이드 — 자연어로 AI 목소리를 연출하는 법

 

Google ADK 실전 가이드 — 에이전트를 백엔드 시스템처럼 만드는 법