AI 음성인식 Whisper 실시간 음성인식 한국어 파인튜닝까지

AI 음성인식 Whisper 실시간 음성인식 한국어 파인튜닝까지

실시간 음성 챗봇이나 회의록 자동화, AI 상담 서비스(AI 음성인식)를 만들다 보면 반드시 마주치는 질문이 하나 있습니다. “어떤 음성인식(STT) 모델을 써야 하지?” 그리고 Whisper를 선택한 다음에는 곧바로 두 번째 벽에 부딪힙니다. “이걸 어떻게 실시간처럼 빠르게 돌리지?”

이번 글은 AI 음성인식을 처음부터 끝까지 다룹니다. 한국어 음성인식 모델 비교로 시작해서, Whisper를 실시간처럼 동작하게 만드는 최적화 원칙, FastAPI와 WebSocket을 이용한 실시간 양방향 STT 서버 구축, 서버 리팩토링 전략, 마지막으로 한국어 파인튜닝까지 실전 코드와 함께 정리했습니다.

목차

1. AI 음성 인식 모델 비교: Whisper, CLOVA Speech, GPT-4o Transcribe 중 뭘 써야 할까

한국어 음성인식 모델을 고를 때 정확도만 보는 경우가 많은데, 실제로는 정확도 외에도 실시간성, 언어 지원 범위, 비용, 배포 방식(로컬과 클라우드)까지 함께 따져봐야 합니다. Hugging Face의 Open ASR Leaderboard는 최근 다국어와 장문 전사까지 포함해 비교 기준을 넓혔는데, 여기서 LLM 디코더 기반 모델은 정확도는 높지만 속도가 느릴 수 있다고 정리하고 있습니다. OpenAI 공식 문서 역시 Whisper를 범용 음성인식 모델로 두고, 최신 오디오 API에서는 GPT-4o Transcribe 같은 신규 STT 모델을 함께 제공하는 방향으로 가고 있습니다.

결론부터 말하면, 최신 기준으로 Whisper 계열은 여전히 범용성과 다국어 지원에서 강력하고, 영어 실시간·고속 처리에서는 NVIDIA 계열 최신 ASR 같은 경량·고성능 모델이 유리합니다. 반면 한국어 중심 서비스라면 네이버 CLOVA Speech 같은 상용 API가 실무 만족도가 높다고 평가받습니다.

AI 음성인식 모델별 장단점 비교표

모델/서비스장점단점추천 용도
Whisper오픈소스, 로컬 실행 가능, 다국어 강함, 배치 전사에 좋음기본 구현은 실시간·화자분리에서 약함자막 생성, 회의록, 로컬 보안 환경
GPT-4o Transcribe최신 OpenAI 오디오 계열, API 중심으로 쓰기 편함폐쇄형, 비용과 종속성 고려 필요서비스형 STT, 빠른 프로토타입
Google STT실시간 스트리밍, 안정성, 대규모 운영에 강함유료 API, 커스터마이징 제한콜센터, 실시간 자막
Azure Speech기업 통합, 실시간 처리와 기능 균형유료, 한국어 특화는 전용 서비스 대비 약할 수 있음엔터프라이즈 워크플로
Naver CLOVA Speech한국어 정확도와 실사용 만족도가 높음폐쇄형, 튜닝 자유도 제한한국어 콜센터, 의료·상담, 국내 서비스
NVIDIA 최신 ASR 계열영어 정확도와 속도 모두 강함, 리더보드 상위권영어 중심, 배포 난이도 높을 수 있음영어 중심 고속 전사, 서버 배치
Speechmatics / Deepgram 계열실시간 API, 억양·다국어 대응 강점비용, 벤더 종속성글로벌 서비스, 실시간 API

장단점 요약

  • Whisper는 오픈소스라 직접 통제가 가능하고 다국어 전사에 강하지만, 실시간 서비스에서는 별도의 최적화 작업이 필요합니다.
  • Google STT, Azure Speech, Deepgram, Speechmatics 같은 상용 API는 운영이 편하고 실시간에 강하지만, 비용과 데이터 외부 전송 이슈를 함께 고려해야 합니다.
  • 한국어 품질만 놓고 보면 네이버 CLOVA Speech가 가장 실무적인 선택으로 자주 언급됩니다.

상황별 추천 모델

  • 한국어 최우선, 실서비스 → Naver CLOVA Speech
  • 로컬 실행, 비용 절감, 다국어 자막·회의록 → Whisper
  • 실시간 스트리밍·콜센터 → Google STT 또는 Azure Speech
  • 최신 OpenAI API 중심 개발 → GPT-4o Transcribe
  • 영어 중심 고속 처리 → NVIDIA 최신 ASR 계열

정리하면, 지금 기준으로 “범용 최고”는 Whisper, “한국어 실전 최고”는 CLOVA, “서비스 개발 편의성”은 GPT-4o Transcribe로 보는 것이 가장 균형 잡힌 판단입니다. 한국어 음성인식 품질과 실무 안정성을 우선하면 CLOVA Speech, 로컬 구축과 커스터마이징을 원하면 Whisper, 최신 API로 빠르게 제품화하려면 GPT-4o Transcribe가 현실적인 선택지입니다.

2. Whisper AI 음성인식 최적화 원칙 (feat. 슬라이딩 윈도우, VAD)

여기서 짚고 넘어가야 할 중요한 전제가 하나 있습니다. Whisper 실시간 서비스는 “진짜 스트리밍 모델”이 아니라, 짧은 청크(chunk) + 오버랩(overlap) + 증분 후처리로 실시간처럼 보이게 만드는 방식이라는 점입니다. Whisper는 원래 완결된 오디오 입력에 맞춰 설계된 모델이라서, 이 구조적 한계를 이해하고 접근해야 지연시간(latency) 문제를 풀 수 있습니다.

Whisper 지연시간을 줄이는 가장 효과적인 최적화 조합은 다음과 같습니다.

  • 모델 크기 축소
  • VAD(Voice Activity Detection, 무음 구간 제거)
  • 슬라이딩 윈도우(sliding window)
  • GPU FP16 연산
  • 비동기 파이프라인

최적화의 핵심 원리

1~5초 단위로 오디오 청크를 잘라 넣고, 0.5~1초 정도 오버랩을 두는 방식이 지연과 끊김을 동시에 줄여줍니다. Open ASR Leaderboard 기준으로도 긴 오디오에서는 정확도와 처리량 사이의 트레이드오프가 크고, CTC/TDT 계열 모델이 더 빠르지만 Whisper Large v3는 다국어 기준으로 여전히 강력한 베이스라인으로 평가됩니다. 실무에서는 tiny/base/small 같은 경량 모델이 실시간에 유리하며, 그중에서도 small 모델이 정확도와 속도의 균형점으로 자주 권장됩니다.

추천 처리 구조

  1. 마이크 입력을 계속 버퍼에 쌓는다.
  2. VAD로 무음 구간을 제거한다.
  3. 3초 전후의 윈도우를 만들고 1초 정도 오버랩을 둔다.
  4. GPU에서 FP16으로 추론한다.
  5. 중복 구간 텍스트를 제거하고 부분 결과를 스트리밍한다.

Python 구현 예시: 로컬 마이크 → 슬라이딩 윈도우 → Whisper 추론

아래는 로컬 마이크 입력을 슬라이딩 윈도우 방식으로 잘라 Whisper에 넣는 최소 구현 예시입니다.

python

import asyncio
import queue
import threading
import numpy as np
import sounddevice as sd
import whisper

RATE = 16000
WINDOW_SEC = 3.0
OVERLAP_SEC = 1.0
CHUNK_SEC = 0.25

model = whisper.load_model("small").to("cuda")

audio_q = queue.Queue()
buffer = np.zeros(0, dtype=np.float32)

def callback(indata, frames, time, status):
    if status:
        print(status)
    audio_q.put(indata[:, 0].copy())

def transcribe_chunk(audio):
    result = model.transcribe(
        audio,
        language="ko",
        fp16=True,
        temperature=0.0,
        condition_on_previous_text=False,
        beam_size=1,
        best_of=1,
        no_speech_threshold=0.6
    )
    return result["text"].strip()

async def main():
    global buffer
    with sd.InputStream(
        channels=1,
        samplerate=RATE,
        blocksize=int(RATE * CHUNK_SEC),
        callback=callback,
    ):
        print("start")
        while True:
            chunk = await asyncio.to_thread(audio_q.get)
            buffer = np.concatenate([buffer, chunk])

            need = int(RATE * WINDOW_SEC)
            if len(buffer) >= need:
                audio = buffer[:need]
                keep = int(RATE * OVERLAP_SEC)
                buffer = buffer[need - keep:]

                text = await asyncio.to_thread(transcribe_chunk, audio)
                if text:
                    print(text)

asyncio.run(main())

실서비스에 적용할 때 팁

  • 실서비스에서는 sounddevice 직결 구조보다 WebSocket + 서버 추론 큐 구조가 훨씬 안정적입니다.
  • 동시 사용자 수가 늘어나면 요청별로 즉시 추론하는 것보다, GPU 단위로 배치를 묶어 처리하는 방식이 처리량 면에서 유리합니다.
  • 한국어 서비스라면 처음엔 base 모델로 시작해서 품질이 부족할 때 small로 올리는 순서가 현실적입니다.

Whisper 실시간 처리를 더 빠르게 만드는 구체적 방법

  • tiny 또는 base 모델 사용
  • fp16=True와 GPU 사용
  • VAD로 무음 제거
  • condition_on_previous_text=False로 누적 의존성 감소
  • beam_size=1, best_of=1로 디코딩 단순화
  • 청크 길이를 너무 키우지 않기
  • 오래된 whisper 라이브러리보다 faster-whisper 같은 최신 최적화 스택 검토

한국어 실시간 자막·회의록 서비스라면 Whisper small + VAD + 3초 윈도우 + 1초 오버랩 + GPU FP16 구성이 가장 균형이 좋은 조합입니다. 정확도보다 지연이 더 중요하면 base로 낮추고, 품질이 더 중요하면 small 이상으로 올리면 됩니다.

3. FastAPI + WebSocket으로 실시간 양방향 음성인식 서버 만들기

여기서 말하는 “실시간 양방향“이란, 클라이언트가 서버로 오디오를 실시간 스트리밍 전송하고, 서버는 부분(intermediate) 텍스트를 즉시 클라이언트에 푸시하는 구조를 뜻합니다. 핵심은 WebSocket(FastAPI + websockets)으로 마이크·브라우저에서 PCM 청크를 전송하고, 서버는 VAD·버퍼링(슬라이딩 윈도우) 후 Whisper(또는 faster-whisper) 추론을 비동기 파이프라인으로 수행해 부분 결과를 바로 푸시하는 것입니다.

기존에 마이크 기반으로만 동작하던 STT 모듈(stt_module.py)이 있다면, 그 파일의 동작은 그대로 두고 서버-클라이언트 양방향 스트리밍용 WebSocket 엔드포인트와 비동기 추론 워커(버퍼·VAD·모델 호출)만 추가하는 방식으로 최소 변경이 가능합니다.

변경 포인트 정리

  • FastAPI + WebSocket 서버 추가: 클라이언트(브라우저 또는 로컬 클라이언트)가 오디오 청크를 보내면 서버는 즉시 ACK나 부분 텍스트를 푸시합니다.
  • 비동기 오디오 큐: 클라이언트별 asyncio.Queue를 사용해 오디오 청크를 수집하고, VAD·슬라이딩 윈도우를 적용합니다.
  • 추론 워커: 큐에서 윈도우 단위 오디오를 꺼내 model.transcribe(버퍼)를 비동기(to_thread)로 호출하고, 부분 결과를 WebSocket으로 전송합니다.
  • 모델 설정: faster-whisperWhisperModel을 사용한다면 fp16 / small 모델이 지연-정확도 균형에 유리합니다.
  • 중복 텍스트 처리: 오버랩 구간에서 중복 출력이 나올 수 있으므로 최근 출력과 비교해 제거합니다.
  • 에러·종료 처리: 클라이언트가 종료되면 큐를 정리하되, 모델 관련 리소스는 유지합니다.

필요한 패키지는 다음과 같이 설치합니다.

pip install fastapi "uvicorn[standard]" python-multipart websockets aiofiles soundfile numpy faster-whisper

주의: 이 구조는 서버용으로 설계되어 있으며, GPU에서 faster-whisper가 동작 가능한 환경(CUDA, 드라이버, ffmpeg 등)이 필요합니다. 프로덕션에서는 인증, rate-limit, 멀티 GPU 분배 등을 추가로 고려해야 합니다.

서버 코드: FastAPI WebSocket + 비동기 추론 워커

기존 stt_module.py와 같은 디렉토리에 두고 임포트할 수 있는 형태의 예시입니다.

# stt_realtime_ws.py  (기존 stt_module.py와 같은 디렉토리에 두고 import 가능)
import asyncio
import base64
import io
import json
import time
from typing import Dict

import numpy as np
import soundfile as sf
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import uvicorn
from faster_whisper import WhisperModel
from concurrent.futures import ThreadPoolExecutor

# 모델 초기화 (프로덕션: process-level 하나만 로드)
MODEL_SIZE = "small"  # small 권장(지연-정확도 균형)
model = WhisperModel(MODEL_SIZE, device="cuda", compute_type="float16")
executor = ThreadPoolExecutor(max_workers=2)

app = FastAPI()

# 클라이언트별 상태 보관
clients: Dict[str, Dict] = {}
# 설정
SAMPLE_RATE = 16000
WINDOW_SEC = 3.0
OVERLAP_SEC = 1.0
MIN_CHUNKS_FOR_INFER = int(WINDOW_SEC * SAMPLE_RATE / 160)  # 예시 (청크 크기 160 샘플 가정)

def pcm_bytes_to_float32(pcm_bytes: bytes, dtype="int16"):
    arr = np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32) / 32768.0
    return arr

async def transcribe_buffer(audio_float32: np.ndarray):
    # faster-whisper accepts file-like object or numpy array; 안전하게 wave로 변환
    with io.BytesIO() as wf:
        sf.write(wf, audio_float32, SAMPLE_RATE, format="WAV")
        wf.seek(0)
        # 모델은 블로킹이므로 threadpool에 위임
        loop = asyncio.get_event_loop()
        segments, info = await loop.run_in_executor(
            executor,
            lambda: model.transcribe(
                wf,
                language="ko",
                vad_filter=True,
                beam_size=1,
                condition_on_previous_text=False,
                fp16=True
            )
        )
        text = ""
        for seg in segments:
            if getattr(seg, "no_speech_prob", 0.0) < 0.6:
                text += seg.text
        return text.strip()

async def inference_worker(client_id: str):
    """각 클라이언트에 대해 실행되는 워커: 큐에서 오디오 청크를 모아 윈도우 단위로 추론"""
    state = clients[client_id]
    q: asyncio.Queue = state["queue"]
    ws: WebSocket = state["ws"]
    buffer = np.zeros(0, dtype=np.float32)
    last_sent = ""
    try:
        while True:
            item = await q.get()
            if item is None:
                break
            # item은 PCM 바이트 (int16 little-endian)
            samples = pcm_bytes_to_float32(item)
            buffer = np.concatenate([buffer, samples])

            need = int(WINDOW_SEC * SAMPLE_RATE)
            keep = int(OVERLAP_SEC * SAMPLE_RATE)

            if len(buffer) >= need:
                window_audio = buffer[:need]
                buffer = buffer[need - keep:]
                # 추론
                text = await transcribe_buffer(window_audio)
                # 중복/빈문자 필터
                if text and text != last_sent:
                    await ws.send_text(json.dumps({"type": "partial", "text": text}))
                    last_sent = text
    except WebSocketDisconnect:
        pass
    except Exception as e:
        await ws.send_text(json.dumps({"type": "error", "error": str(e)}))
    finally:
        # 정리
        clients.pop(client_id, None)

@app.get("/")
async def index():
    return HTMLResponse("<html><body>Whisper RT WebSocket endpoint at /ws</body></html>")

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    client_id = str(id(ws)) + "_" + str(time.time())
    q = asyncio.Queue()
    clients[client_id] = {"ws": ws, "queue": q}
    worker_task = asyncio.create_task(inference_worker(client_id))
    await ws.send_text(json.dumps({"type": "ready", "client_id": client_id}))
    try:
        while True:
            msg = await ws.receive_text()
            # 클라이언트가 base64 PCM 청크를 보낸다고 가정
            data = json.loads(msg)
            if data.get("type") == "chunk":
                b64 = data["data"]
                pcm = base64.b64decode(b64)
                await q.put(pcm)
            elif data.get("type") == "end":
                await q.put(None)
                break
            else:
                await ws.send_text(json.dumps({"type": "error", "error": "unknown message type"}))
    except WebSocketDisconnect:
        await q.put(None)
    finally:
        worker_task.cancel()
        clients.pop(client_id, None)

if __name__ == "__main__":
    uvicorn.run("stt_realtime_ws:app", host="0.0.0.0", port=8000, reload=False)

브라우저(클라이언트) 코드: 마이크 → 서버로 base64 PCM 전송

브라우저에서 getUserMedia로 오디오 스트림을 얻고, ScriptProcessor(실무에서는 AudioWorklet 권장)로 PCM int16 변환 후 base64로 인코딩해 WebSocket으로 전송하는 예시입니다.

// 브라우저 예시 (간단히 설명)
// 1) getUserMedia로 오디오 스트림 얻기
// 2) AudioWorklet 또는 ScriptProcessor로 PCM int16 변환 후 base64로 WebSocket 전송
// (여기선 간단 의사 코드)
const ws = new WebSocket("ws://server:8000/ws");
ws.onopen = () => console.log("connected");
ws.onmessage = e => {
  const msg = JSON.parse(e.data);
  if(msg.type === 'partial') {
    console.log("부분텍스트:", msg.text);
  }
};
// 마이크 처리 로직 (실무: AudioWorklet 권장)
navigator.mediaDevices.getUserMedia({ audio: true }).then(stream => {
  const ctx = new AudioContext({ sampleRate: 16000 });
  const src = ctx.createMediaStreamSource(stream);
  const processor = ctx.createScriptProcessor(4096, 1, 1);
  src.connect(processor);
  processor.connect(ctx.destination);
  processor.onaudioprocess = (e) => {
    const float32 = e.inputBuffer.getChannelData(0);
    // float32 -> int16
    let l = float32.length;
    let buffer = new ArrayBuffer(l * 2);
    let view = new DataView(buffer);
    for (let i=0;i<l;i++){
      let s = Math.max(-1, Math.min(1, float32[i]));
      view.setInt16(i*2, s < 0 ? s*0x8000 : s*0x7FFF, true);
    }
    // base64로 인코딩 (브라우저에서 btoa 사용시 먼저 binary string 만들어야 함)
    let binary = '';
    const bytes = new Uint8Array(buffer);
    for (let i=0;i<bytes.byteLength;i++) binary += String.fromCharCode(bytes[i]);
    const b64 = btoa(binary);
    ws.send(JSON.stringify({type:'chunk', data: b64}));
  };
});

기존 stt_module.py와 통합할 때 팁

  • 기존 VoiceAssistant의 transcribe 로직(버퍼 → model.transcribe)을 transcribe_buffer 내부 로직으로 옮기거나 호출하도록 리팩토링하면, 기존 파일 변환 함수(transcribe_file)와 WebSocket 워커가 같은 transcribe 함수를 재사용할 수 있습니다.
  • faster_whisper 모델 객체는 프로세스 단위로 하나만 생성해 여러 코루틴·쓰레드에서 재사용해야 합니다(위 예제에서 전역 모델로 로드한 이유입니다).
  • 로컬 마이크 기반 start_listening은 그대로 두고, WebSocket 엔드포인트에서 클라이언트가 로컬 마이크 대신 오디오를 전송하면 동일한 transcribe 흐름을 타도록 만드는 것이 가장 자연스럽습니다.
  • VAD를 더 엄격히 적용하려면 webrtcvad 라이브러리로 전처리한 뒤 큐에 넣는 방식이 노이즈 많은 환경에서 유리합니다.

기존 클래스에 start_ws_server() 메서드를 추가해 위 FastAPI 앱을 uvicorn으로 내부 스레드에서 띄우는 방식도 가능하지만, 프로덕션에서는 별도 프로세스로 운영하는 것을 권장합니다.

성능·운영 고려사항

  • 동시 사용자 증가 시 GPU 병목: 큐를 통해 배치·스루풋을 최적화(예: 여러 클라이언트 오디오를 시간 창별로 묶어 한 번에 추론)하는 방식을 고려해야 합니다.
  • 모델 스케일: small/base/tiny로 테스트해 지연-정확도 균형을 조정합니다.
  • 보안: WebSocket 인증과 TLS는 필수입니다.
  • 모니터링: 추론 지연, 큐 길이, 에러 로그를 수집해야 합니다.

4. 서버용 구조로 리팩토링하기: 병목 지점과 해결 원칙

실시간 STT 서버를 구축할 때 로컬 마이크 직접 수집 방식을 그대로 서버에 옮기면 성능이 크게 떨어집니다. 지금 구조에서 서버용으로 최적화하려면, 입력·추론·출력을 분리한 비동기 서버 구조로 바꾸는 것이 핵심입니다. 특히 기존 stt_module.py는 그대로 쓰기보다, FastAPI 기반 AI 서버(예: ai_server-2.py)의 생명주기와 통합해서 모델 인스턴스 1개를 공유하고, 요청마다 오디오를 받는 방식으로 바꾸는 것이 맞습니다.

먼저 바꿀 방향

PyAudio로 마이크를 직접 읽는 구조는 서버용으로는 비효율적입니다. 서버에서는 보통 다음 순서로 갑니다.

  1. 클라이언트가 오디오 청크를 전송한다.
  2. 서버는 버퍼에 누적한다.
  3. VAD로 무음을 제거한다.
  4. 일정 길이만큼 쌓이면 Whisper로 추론한다.
  5. 부분 결과를 SSE 또는 WebSocket으로 즉시 반환한다.

이렇게 해야 여러 사용자를 동시에 처리하기 쉽고, is_speaking 같은 단일 전역 상태 충돌도 피할 수 있습니다.

현재 구조에서 흔히 발견되는 병목 지점

  • WhisperModel을 클래스 생성 시마다 로드하면 GPU 메모리 낭비가 큽니다.
  • beam_size=5, condition_on_previous_text=True 설정은 지연을 키웁니다.
  • vad_filter=True와 별도의 RMS 기반 감지 로직이 중복되는 경우가 많습니다.
  • PyAudio 기반 루프는 서버 동시성에 불리합니다.
  • 버퍼를 매번 WAV로 만들고 다시 읽는 구조는 불필요한 오버헤드를 만듭니다.

서버 최적화 원칙

  • 모델 1회 로드: 프로세스 시작 시 한 번만 로드합니다.
  • 비동기 큐: 사용자별 오디오 큐를 둡니다.
  • 짧은 청크: 보통 1~3초 단위로 처리합니다.
  • 오버랩: 0.5~1초 정도 겹치게 처리합니다.
  • 디코딩 단순화: beam_size=1, temperature=0.0.
  • 조건부 문맥 축소: condition_on_previous_text=False.
  • GPU 고정: device="cuda", compute_type="float16".
  • 입력 포맷 고정: 16kHz mono PCM으로 강제합니다.
  • 부분 결과 전송: 완전한 문장을 기다리지 말고 중간 결과를 먼저 전달합니다.

stt_module.py를 서버용으로 수정하는 방향

기존 파일은 서버용으로는 역할을 아래처럼 축소하는 것이 좋습니다.

  • start_listening()은 제거하거나 로컬 테스트 전용으로만 유지합니다.
  • transcribe_file()은 배치 전사용으로 유지합니다.
  • 새로 transcribe_chunk(audio_bytes)를 추가합니다.
  • 새로 transcribe_stream(audio_queue)를 추가합니다.
  • 모델 초기화는 __init__ 밖에서 캐시하거나 싱글톤으로 변경합니다.

예를 들면 이런 구조가 맞습니다.

python

class VoiceAssistant:
    _model = None

    def __init__(self, model_size="small", device="cuda", compute_type="float16"):
        if VoiceAssistant._model is None:
            VoiceAssistant._model = WhisperModel(
                model_size,
                device=device,
                compute_type=compute_type,
                cpu_threads=4,
                num_workers=1,
            )
        self.model = VoiceAssistant._model

이렇게 하면 요청마다 모델을 다시 올리지 않습니다.

실시간 양방향 구조를 위한 권장 엔드포인트 설계

“실시간 양방향”은 보통 다음을 뜻합니다.

  • 클라이언트 → 서버: 마이크 오디오 전송
  • 서버 → 클라이언트: 부분 STT 결과를 즉시 push
  • 서버 → 클라이언트: TTS 응답도 이어서 push

즉, 기존에 있던 스트리밍 채팅 엔드포인트(예: /api/chat/stream)의 스타일을 확장해서, STT까지 동일한 흐름으로 합치는 것이 가장 자연스럽습니다. 권장 구조는 다음과 같습니다.

  • /ws/stt : 오디오 스트림 입력용 WebSocket
  • /ws/chat : 텍스트 응답·상태 전송용 WebSocket
  • /stt : 파일 업로드 배치 전용
  • /api/chat/stream : LLM·TTS 스트리밍 전용

서버 체감 성능을 올리는 우선순위

  1. large-v3 대신 small 또는 medium으로 테스트
  2. beam_size=1
  3. condition_on_previous_text=False
  4. VAD는 한 군데만 사용
  5. 오디오를 WAV 파일로 매번 저장하지 말고 numpy 배열 그대로 처리
  6. 여러 요청이 들어오면 추론 큐를 둬서 GPU 작업을 직렬화하거나 소배치 처리
  7. 로그 출력량 줄이기 (세그먼트마다 print 남발하지 않기)

추천 적용 순서

  1. 1단계: stt_module.py를 배치·실시간 공용 구조로 리팩토링
  2. 2단계: 기존 AI 서버에 WebSocket STT 엔드포인트 추가
  3. 3단계: 모델 싱글톤화
  4. 4단계: 청크 처리 + 오버랩 + 부분 결과 반환
  5. 5단계: 필요하면 faster-whisper 설정을 낮춰 지연 최소화

5. Whisper small vs large, 실시간 서비스엔 어떤 모델이 맞을까

서버 기준으로는 아래처럼 모델을 나눠 시작하는 것이 좋습니다.

  • 정확도 우선: medium
  • 균형형: small
  • 고속 우선: base
  • 대형 모델(large-v3): 동시 사용자 수가 적을 때만 권장

large-v3는 품질은 좋지만 서버 동시성 측면에서는 무겁습니다. 실시간 서비스를 목표로 한다면 small이나 medium이 훨씬 현실적입니다.

small 모델을 선택하면 얻는 구체적 이점

  • 훨씬 빠른 처리 속도: small 모델은 large 모델에 비해 약 2.4배 빠르게 추론합니다. 실시간 대화에서는 이 속도 차이가 체감상 매우 큽니다.
  • 적은 컴퓨팅 자원: large 모델(1550M 매개변수)에 비해 small 모델(244M 매개변수)은 GPU 메모리 사용량이 훨씬 적어, 서버 비용을 절감하거나 더 많은 동시 사용자를 처리할 수 있습니다.

모델 선택 시 참고할 포인트

  • 한국어 특화 모델의 존재: 일부 업체는 small 모델을 한국어 데이터로 추가 학습시켜 large 모델보다 뛰어난 성능을 내는 모델을 만들기도 합니다. 이는 모델 크기보다 데이터와 학습 방식이 더 중요할 수 있음을 보여줍니다.
  • 파인튜닝의 영향: 한국어 데이터로 파인튜닝하면 성능이 크게 향상될 수 있지만, 한국어에만 과적합되면 다른 언어 성능이 급격히 떨어질 수 있다는 점도 함께 고려해야 합니다.
  • 평가 지표의 한계: large 모델처럼 성능이 지나치게 좋아지면, 오히려 발음의 미세한 차이를 평가하는 데는 부적합할 수 있다는 지적도 있습니다.

즉, 무조건 큰 모델이 정답은 아니며, 서비스 목적에 맞는 크기 + 한국어 파인튜닝 조합이 더 나은 결과를 낼 수 있습니다. 다음 장에서는 이 파인튜닝을 직접 진행하는 방법을 다룹니다.

6. 한국어 Whisper 파인튜닝 방법 (ENERZAi 사례 포함)

Whisper small 모델을 한국어로 파인튜닝하는 작업은 생각보다 어렵지 않습니다. 잘 정리된 가이드와 사례를 따라가면 여러분의 환경에 맞춰 직접 진행할 수 있습니다.

왜 한국어 파인튜닝이 필요할까

Whisper 원본 모델은 방대한 다국어 데이터로 학습되었지만, 그중 한국어 데이터의 비중은 매우 작습니다. ENERZAi 사례를 보면, Whisper Large 모델의 영어 오류율(CER)은 3.91%였지만 한국어에서는 11.13%로 크게 증가했습니다. 그런데 한국어 데이터로 추가 학습을 수행하면, 모델 크기는 원래의 0.4% 수준(3GB → 13MB)까지 줄이면서도 더 뛰어난 한국어 인식 성능을 달성할 수 있었습니다. 이는 모델 크기 자체보다 데이터와 학습 방식의 중요성을 잘 보여주는 사례입니다.

파인튜닝을 위한 핵심 단계

Hugging Face의 공식 가이드와 여러 성공적인 한국어 파인튜닝 모델들은 공통적으로 다음 접근 방식을 사용합니다.

(1) 학습 데이터 준비

  • 공개 데이터셋 활용: Zeroth Korean Dataset은 한국어 음성인식 파인튜닝에 자주 사용되는 데이터셋입니다. 이 데이터셋으로 파인튜닝된 seastar105/whisper-small-ko-zeroth 모델이 대표적인 성공 사례입니다.
  • 자체 데이터 구축: 특정 도메인에 특화된 모델이 필요하다면 직접 음성 데이터를 준비할 수 있습니다. 보통 librosa 라이브러리로 .wav 파일을 모델이 이해할 수 있는 Mel Spectrogram으로 변환합니다.

(2) 학습 환경 및 설정 구성

ENERZAi는 대규모 데이터(약 50,000시간)와 맞춤형 토크나이저, 양자화 같은 고급 기법을 사용했지만, 더 단순한 접근 방식으로도 충분히 좋은 결과를 얻을 수 있습니다. Hugging Face에 공개된 한국어 Whisper-small 파인튜닝 모델들이 자주 사용하는 하이퍼파라미터는 다음과 같습니다.

  • learning_rate: 1e-05 (안정적인 학습을 위한 일반적인 선택)
  • batch_size: 16 (GPU 메모리에 따라 조정)
  • optimizer: Adam
  • mixed_precision_training: Native AMP (FP16 혼합 정밀도 학습으로 메모리와 속도 최적화)

추가로 고려할 만한 선택적 기법도 있습니다.

  • 맞춤형 토크나이저: 한국어에 최적화된 토크나이저를 사용하면 성능을 크게 향상시킬 수 있습니다.
  • 정규화: 다양한 출처의 데이터셋에 존재하는 표기 차이(예: “3시에” vs “세 시에”)를 통일하는 정규화 과정이 학습 효율을 높입니다.

(3) 학습 실행 및 평가

대부분의 파인튜닝은 Hugging Face transformers 라이브러리를 기반으로 수행됩니다.

  • 평가 지표: 주로 **WER(Word Error Rate, 단어 오류율)**과 **CER(Character Error Rate, 글자 오류율)**을 사용하며, 수치가 낮을수록 성능이 좋습니다.
  • 성능 예시: 일부 공개 모델은 Zeroth 데이터셋 평가 기준으로 WER 6~9% 수준의 성능을 기록하기도 했습니다.

실용적인 조언

  • 시작은 공개된 모델로: 직접 처음부터 학습시키는 대신, Hugging Face에 이미 공개된 한국어 파인튜닝 모델(whisper-small-ko-zeroth 등)을 활용하는 것이 가장 빠른 방법입니다.
  • 도메인 특화가 필요하다면: 의료, 고객센터 등 특정 분야에서 더 나은 성능이 필요하다면, 해당 도메인 데이터를 추가로 수집해 공개 모델을 기반으로 추가 파인튜닝(Continual Learning)을 고려할 수 있습니다.
  • 하드웨어 고려: 파인튜닝에는 GPU가 필요하며, transformers 라이브러리와 함께 CUDA 지원 GPU를 사용하는 것이 일반적입니다.

이미 만들어진 모델을 먼저 써보기

직접 학습시키기 전에, Hugging Face에 이미 공개된 한국어 파인튜닝 Whisper small 모델을 먼저 사용해보는 방법도 있습니다. 예를 들어 seastar105/whisper-small-ko-zerothdaekeun-ml/whisper-small-ko-finetuned-single-speaker-3922samples 같은 모델은 transformers 라이브러리로 간단히 불러와 바로 사용할 수 있습니다.

원하는 데이터로 직접 학습시키고 싶다면, 아래 단계를 따라가면 됩니다.

1단계: 환경 준비하기

Google Colab을 사용하면 복잡한 설정 없이 무료로 GPU를 사용할 수 있어서 초보자에게 가장 추천되는 방법입니다.

# 필요한 라이브러리 설치
!pip install datasets>=2.6.1
!pip install git+https://github.com/huggingface/transformers
!pip install librosa
!pip install evaluate>=0.30
!pip install jiwer
!pip install accelerate -U
!pip install transformers[torch]

2단계: 데이터 준비하기

모델을 학습시키려면 음성 파일과 그에 맞는 정답 텍스트 데이터가 필요합니다.

  • 공개 데이터셋 활용: Zeroth Korean Dataset이나 KSS(Korean Single Speaker Speech) Dataset을 사용할 수 있습니다.
  • 나만의 데이터 준비: CSV 파일 등에 음성 파일 경로와 정답 텍스트를 함께 정리해 둡니다.
  • 데이터 전처리: Whisper 모델이 데이터를 이해하려면 모든 음성 파일의 샘플링 레이트를 16,000Hz로 통일해야 합니다. 음성 데이터는 모델에 입력되기 전에 log-Mel 스펙트로그램 형태로 변환되어야 하는데, 이 복잡한 변환은 WhisperFeatureExtractor가 알아서 처리해줍니다.

3단계: 모델과 도구 불러오기

from transformers import WhisperFeatureExtractor, WhisperTokenizer, WhisperProcessor, WhisperForConditionalGeneration
from datasets import load_dataset
import torch

# 1. Feature Extractor: 오디오를 모델이 이해하는 형태(log-Mel 스펙트로그램)로 변환
feature_extractor = WhisperFeatureExtractor.from_pretrained("openai/whisper-small")

# 2. Tokenizer: 텍스트를 모델이 이해하는 숫자(token)로 변환
#    반드시 언어(language)와 작업(task)를 'korean'과 'transcribe'로 지정해주세요!
tokenizer = WhisperTokenizer.from_pretrained("openai/whisper-small", language="Korean", task="transcribe")

# 3. Processor: Feature Extractor와 Tokenizer를 하나로 묶어 편리하게 사용
processor = WhisperProcessor.from_pretrained("openai/whisper-small", language="Korean", task="transcribe")

# 4. 모델: 학습시킬 Whisper Small 모델
model = WhisperForConditionalGeneration.from_pretrained("openai/whisper-small")

4단계: 학습시키기

학습 방법은 크게 두 가지로 나뉩니다.

방법 A: 표준 파인튜닝 (Standard Fine-tuning)

가장 일반적인 방법으로, 준비한 전체 데이터로 모델의 모든 가중치를 업데이트합니다. 주요 설정값은 다음과 같습니다(많은 성공 사례에서 검증된 값입니다).

  • learning_rate: 1e-5 ~ 2e-5 (학습 속도)
  • train_batch_size: 4 ~ 16 (한 번에 학습할 데이터 양, GPU 메모리에 따라 조정)
  • num_train_epochs: 5 (전체 데이터를 반복 학습할 횟수)
  • fp16: True (혼합 정밀도 학습으로 메모리 절약과 속도 향상)
  • gradient_checkpointing: True (GPU 메모리 절약에 도움)
  • eval_steps: 200 (몇 스텝마다 모델 성능을 평가할지)
from transformers import Seq2SeqTrainingArguments, Seq2SeqTrainer

training_args = Seq2SeqTrainingArguments(
    output_dir="./whisper-small-ko",       # 학습 결과가 저장될 폴더
    per_device_train_batch_size=16,        # 배치 크기
    gradient_accumulation_steps=1,
    learning_rate=1e-5,                    # 학습률
    warmup_steps=500,                      # 학습률을 서서히 높이는 단계
    max_steps=4000,                        # 전체 학습 스텝 수
    gradient_checkpointing=True,
    fp16=True,                             # 혼합 정밀도 학습
    evaluation_strategy="steps",
    per_device_eval_batch_size=8,
    save_steps=1000,
    eval_steps=500,
    logging_steps=50,
    load_best_model_at_end=True,
    metric_for_best_model="wer",
    greater_is_better=False,
    push_to_hub=False,                     # True로 설정하면 Hugging Face Hub에 업로드
)

trainer = Seq2SeqTrainer(
    args=training_args,
    model=model,
    train_dataset=common_voice["train"],   # 앞서 준비한 학습 데이터셋
    eval_dataset=common_voice["test"],     # 앞서 준비한 평가 데이터셋
    data_collator=processor.feature_extractor.pad,
    tokenizer=processor.tokenizer,
)

trainer.train()

방법 B: LoRA(Low-Rank Adaptation)를 활용한 경량 파인튜닝

리소스가 부족하거나 빠르게 실험해보고 싶다면 LoRA라는 방법이 있습니다. 모델 전체를 학습하는 대신 아주 작은 크기의 추가 파라미터만 학습하는 방식입니다. peft(Parameter-Efficient Fine-Tuning) 라이브러리를 사용하면 훨씬 적은 GPU 메모리로도 파인튜닝이 가능해, 맥북 M1과 같은 환경에서도 성공한 사례가 있습니다.

5단계: 학습된 모델 사용하기

학습이 완료되면 새로운 음성을 텍스트로 변환할 수 있습니다.

python

import librosa
import torch
from transformers import WhisperProcessor, WhisperForConditionalGeneration

# 1. 음성 파일 불러오기 (반드시 16,000Hz로 리샘플링)
file = "내_음성_파일.wav"
arr, sampling_rate = librosa.load(file, sr=16000)

# 2. 학습한 모델과 Processor 불러오기
processor = WhisperProcessor.from_pretrained("openai/whisper-small")
model = WhisperForConditionalGeneration.from_pretrained("내가_학습한_모델_경로")

# 3. 예측하기
input_features = processor(arr, return_tensors="pt", sampling_rate=sampling_rate).input_features
forced_decoder_ids = processor.get_decoder_prompt_ids(language="ko", task="transcribe")
predicted_ids = model.generate(input_features, forced_decoder_ids=forced_decoder_ids)
transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True)

print(transcription)

이 과정을 따라가면 여러분도 직접 한국어에 특화된 나만의 Whisper 음성인식 모델을 만들 수 있습니다.

7. 결론 및 상황별 추천 요약

Whisper 실시간 음성인식을 구현하는 과정은 결국 하나의 흐름으로 이어집니다.

  1. 서비스 성격에 맞는 STT 모델을 고른다 (한국어 실무 안정성이면 CLOVA, 로컬·커스터마이징이면 Whisper)
  2. Whisper를 골랐다면 청크 + 오버랩 + VAD + GPU FP16 + 디코딩 단순화로 실시간처럼 만든다
  3. FastAPI + WebSocket으로 진짜 양방향 스트리밍 구조를 구현한다
  4. 서버 구조를 병목 없는 형태로 리팩토링하고 모델을 싱글톤화한다
  5. 서비스 목적에 맞는 모델 크기(small/medium/large)를 고른다
  6. 필요하다면 한국어 데이터로 파인튜닝해 정확도를 끌어올린다

Whisper로 실시간 음성인식이 정말 가능한가요?

Whisper 자체는 스트리밍 전용 모델이 아니지만, 짧은 청크와 오버랩, VAD, GPU FP16 최적화를 조합하면 실시간에 가깝게 동작시킬 수 있습니다.

실시간 서비스에는 small과 large 중 어떤 모델이 나을까요?

small 모델이 large 대비 약 2.4배 빠르고 GPU 메모리도 훨씬 적게 사용하므로, 동시 사용자가 있는 실시간 서비스에는 small이나 medium이 더 현실적입니다.

한국어 인식 정확도를 높이려면 어떻게 해야 하나요?

Zeroth Korean Dataset 등으로 한국어 파인튜닝을 진행하거나, 이미 공개된 한국어 파인튜닝 모델(whisper-small-ko-zeroth 등)을 먼저 사용해보는 것이 빠른 방법입니다.

FastAPI에서 WebSocket으로 실시간 STT 서버를 만들 때 가장 중요한 점은 무엇인가요?

모델을 프로세스당 한 번만 로드하는 싱글톤 구조, 클라이언트별 비동기 큐, 짧은 청크와 오버랩 처리, 그리고 부분 결과를 즉시 푸시하는 구조가 핵심입니다.

댓글 남기기