이 글은 누구를 위한 것인가
- 고객 상담 음성을 자동으로 텍스트로 변환하고 싶은 팀
- Whisper를 프로덕션에 배포하려는 ML 엔지니어
- 실시간 자막·받아쓰기 기능이 필요한 개발자
들어가며
STT는 클라우드 API(Google, AWS, Azure)가 편하지만, 분당 비용이 쌓인다. 대용량 배치 처리는 Whisper 자체 호스팅이 10배 이상 저렴하다. 실시간 스트리밍은 청크 처리로 레이턴시를 줄인다.
이 글은 bluefoxdev.kr의 음성 AI 인프라 가이드 를 참고하여 작성했습니다.
1. STT 시스템 선택 기준
[STT 솔루션 비교]
클라우드 API:
Google STT: $0.006/15초, 스트리밍 지원 ✓
AWS Transcribe: $0.0004/초, 화자 분리 ✓
Azure STT: $0.001/60초
장점: 관리 없음, 스케일링 자동
단점: 비용, 데이터 외부 전송
자체 호스팅 (Whisper):
Whisper Large-v3: GPU 필요 (A10G, 10분/GPU-시간 처리)
Faster-Whisper: 4배 빠른 CTranslate2 최적화 버전
장점: 데이터 보안, 대용량 저렴
단점: GPU 관리, 배치만 (실시간 어려움)
[한국어 특화 고려사항]
표준 Whisper: 한국어 WER ~10-15%
파인튜닝 버전: WER ~5-8%
전문 용어 (의료, 법률): 커스텀 어휘 필요
억양/방언: 추가 데이터 필요
[실시간 스트리밍 전략]
청크 크기: 1-2초 (레이턴시 vs 정확도 트레이드오프)
VAD (Voice Activity Detection): 침묵 감지로 청크 분할
슬라이딩 윈도우: 이전 컨텍스트 유지로 단어 경계 처리
2. 스트리밍 STT 구현
import asyncio
import anthropic
import io
import numpy as np
from dataclasses import dataclass
client = anthropic.Anthropic()
@dataclass
class TranscriptionSegment:
text: str
start_time: float
end_time: float
speaker: str | None = None
confidence: float = 1.0
class StreamingSTTProcessor:
"""실시간 스트리밍 음성 인식"""
def __init__(self, chunk_seconds: float = 1.5, sample_rate: int = 16000):
self.chunk_seconds = chunk_seconds
self.sample_rate = sample_rate
self.chunk_samples = int(chunk_seconds * sample_rate)
self.buffer = np.array([], dtype=np.float32)
self.transcript_history = []
def is_speech(self, audio_chunk: np.ndarray, threshold: float = 0.01) -> bool:
"""간단한 VAD: RMS 에너지 기반 음성 감지"""
rms = np.sqrt(np.mean(audio_chunk ** 2))
return rms > threshold
async def process_chunk(self, audio_bytes: bytes) -> TranscriptionSegment | None:
"""오디오 청크를 텍스트로 변환 (Whisper API 활용)"""
# Whisper API (OpenAI) 또는 자체 서버 호출
import openai
oai_client = openai.OpenAI()
audio_file = io.BytesIO(audio_bytes)
audio_file.name = "chunk.wav"
transcript = oai_client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language="ko",
response_format="verbose_json",
timestamp_granularities=["segment"],
)
if not transcript.text.strip():
return None
return TranscriptionSegment(
text=transcript.text,
start_time=transcript.segments[0].start if transcript.segments else 0,
end_time=transcript.segments[-1].end if transcript.segments else 0,
)
async def stream_transcribe(self, audio_stream) -> asyncio.AsyncGenerator:
"""실시간 오디오 스트림 처리"""
async for audio_chunk in audio_stream:
chunk_array = np.frombuffer(audio_chunk, dtype=np.float32)
self.buffer = np.append(self.buffer, chunk_array)
if len(self.buffer) >= self.chunk_samples:
process_chunk = self.buffer[:self.chunk_samples]
self.buffer = self.buffer[self.chunk_samples:]
if self.is_speech(process_chunk):
audio_bytes = self._to_wav_bytes(process_chunk)
segment = await self.process_chunk(audio_bytes)
if segment:
self.transcript_history.append(segment)
yield segment
def _to_wav_bytes(self, audio: np.ndarray) -> bytes:
import wave, struct
buf = io.BytesIO()
with wave.open(buf, 'wb') as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(self.sample_rate)
pcm = (audio * 32767).astype(np.int16)
wf.writeframes(struct.pack(f'{len(pcm)}h', *pcm))
return buf.getvalue()
async def enhance_transcript_with_llm(raw_transcript: str, context: str = "") -> str:
"""LLM으로 STT 결과 후처리: 구두점, 전문용어 교정"""
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=1000,
messages=[{
"role": "user",
"content": f"""다음 음성 인식 텍스트를 교정하세요.
맥락: {context or "일반 대화"}
원본: {raw_transcript}
교정 규칙:
1. 구두점 추가 (마침표, 쉼표, 물음표)
2. 명백한 오인식 교정
3. 전문용어 통일
4. 문단 구분
교정된 텍스트만 반환하세요."""
}]
)
return response.content[0].text
마무리
STT 프로덕션의 핵심 결정은 세 가지다: 실시간 vs 배치, 클라우드 vs 자체 호스팅, 범용 vs 도메인 특화. 고객 상담처럼 실시간이 필요하면 클라우드 API, 대용량 녹취록 처리는 Faster-Whisper GPU 배치가 경제적이다. LLM 후처리로 구두점과 오인식을 교정하면 최종 품질이 크게 향상된다.