시리즈 · 음성봇 직접 만들기 — 실시간 음성봇의 뼈대를 바닥부터 만든 8편 기록입니다.
- 1. 끼어들기(barge-in) — 재생 중인 TTS를 실시간으로 끊는 법
- 2. "말이 끝났다"는 판단 — VAD 엔드포인터
- 3. 엔진을 코드 수정 없이 바꾸는 인터페이스 설계 — 지금 보는 글
- 4. 로컬 LLM 대화 두뇌와 도구 호출 루프
- 5. 매 턴 지연시간을 재는 이유
- 6. 브라우저 마이크에서 서버까지 — 오디오 프로토콜
- 7. 로컬 Whisper STT 연결하기
- 8. 스텁부터 만든 이유 — 테스트 우선 설계 (완결)
음성봇의 엔진 인터페이스란, STT(음성인식)와 TTS(음성합성)를 실제로 처리하는 부품을 오케스트레이터가 직접 알지 못해도 되게 만드는 추상 계층을 말합니다. 지난 편에서 끼어들기(barge-in) 처리를 다뤘다면, 이번 편은 그 끼어들기를 가능하게 만든 더 아래 층, 즉 엔진을 통째로 갈아 끼울 수 있게 짜둔 구조에 관한 기록입니다.
혼자 음성봇을 만들면서 가장 먼저 부딪힌 문제는 이것이었습니다. STT 엔진 하나를 정해서 오케스트레이터 코드 곳곳에 그 엔진의 함수를 직접 호출해버리면, 나중에 다른 엔진으로 바꾸고 싶을 때 오케스트레이터 코드를 전부 뜯어고쳐야 합니다. 테스트용 스텁 엔진과 실제 엔진을 나란히 두고 검증하고 싶었던 저에게는 이 방식이 애초에 성립하지 않았습니다. 그래서 오케스트레이터가 의존할 대상을 구체 엔진이 아니라 추상 인터페이스 하나로 좁히는 쪽을 택했습니다.
왜 blocking transcribe()가 아니라 요청과 이벤트를 분리했는가
가장 단순한 설계는 transcribe(audio)를 호출하면 텍스트가 바로 리턴되는 blocking 함수였을 겁니다. 하지만 실제 음성 인식은 그렇게 동작하지 않습니다. 사용자가 말을 시작한 순간, 말이 끝나 최종 텍스트가 확정된 순간, 아무 말도 하지 않고 시간이 초과된 순간이 각각 다른 시점에 발생합니다. 하나의 리턴값으로는 이 시간의 흐름을 표현할 수 없습니다.
그래서 저는 STT·TTS 인터페이스를 "요청을 던지고 비동기 이벤트 스트림을 받는" 형태로 설계했습니다. recognize()나 speak()를 호출하면 즉시 결과가 오는 게 아니라, 진행 상황을 알리는 이벤트가 하나씩 흘러나오는 async generator를 돌려받습니다. 오케스트레이터는 이 스트림을 async for로 순회하면서 지금 무슨 일이 일어나고 있는지 그때그때 판단합니다. 이렇게 하면 발화 시작 감지 같은 중간 신호도, 최종 결과도, 타임아웃도 같은 통로로 자연스럽게 흘려보낼 수 있습니다.
abc.ABC로 오케스트레이터와 엔진 사이에 벽을 세우다
이 구조를 코드로 강제하기 위해 STTEngine과 TTSEngine을 파이썬 abc.ABC로 선언했습니다. 추상 메서드로 계약만 정의하고 실제 구현은 전부 하위 클래스에 맡기는 방식입니다.
import abc
from dataclasses import dataclass
from typing import AsyncIterator, Iterable
@dataclass
class EngineEvent:
type: str
text: str | None = None
audio: bytes | None = None
meta: dict | None = None
AudioSource = Iterable[bytes] | AsyncIterator[bytes]
class STTEngine(abc.ABC):
@abc.abstractmethod
async def recognize(self, audio_iter: AudioSource) -> AsyncIterator[EngineEvent]:
...
class TTSEngine(abc.ABC):
@abc.abstractmethod
async def speak(self, text: str) -> AsyncIterator[EngineEvent]:
...
@abc.abstractmethod
async def stop(self) -> None:
...
recognize와 speak의 본문이 ...로만 되어 있는 이유가 있습니다. 여기에 yield를 넣어버리면 추상 메서드 자체가 async generator가 되어버려서, 계약을 선언하는 역할과 실제로 값을 흘려보내는 역할이 뒤섞입니다. yield는 오직 하위 구현체 안에만 있어야 합니다. 오케스트레이터는 이 두 추상 클래스만 보고 코드를 짜고, 그 뒤에 어떤 엔진이 붙는지는 신경 쓰지 않습니다. 그 덕분에 테스트용 Echo·Beep 스텁 엔진과, faster-whisper나 MeloTTS 같은 실제 엔진을 인터페이스만 지키면 코드 수정 없이 서로 바꿔 끼울 수 있습니다.
이벤트 하나에 상태를 다 실어 보내는 법
EngineEvent는 필드 네 개짜리 단순한 dataclass입니다. type이 이벤트 이름이고, text·audio·meta는 그 이벤트 종류에 따라 채워지거나 비워집니다. STT 쪽 이벤트 어휘는 세 가지로 정리했습니다. START_OF_INPUT은 사용자가 말을 시작했다는 신호이고, RECOGNITION_COMPLETE는 최종 텍스트가 확정됐다는 신호로 이때 text 필드가 채워집니다. NO_INPUT_TIMEOUT은 정해진 시간 동안 아무 입력도 없었다는 신호입니다.
이 세 가지만으로도 오케스트레이터가 대화의 진행을 추적하는 데 충분합니다. 말을 시작했는지, 끝난 텍스트가 무엇인지, 아니면 아예 말이 없었는지를 구분해서 처리 흐름을 분기하면 되기 때문입니다. 이벤트 타입을 문자열로 넓게 열어둔 이유도 여기 있습니다. 나중에 이벤트 종류가 늘어나도 EngineEvent의 구조 자체는 그대로 두고 새 type 값만 추가하면 됩니다.
오디오 규격을 한 줄로 통일해야 스왑이 실제로 된다
인터페이스를 아무리 잘 짜도 오디오 포맷이 엔진마다 제각각이면 교체는 말뿐인 구조가 됩니다. 그래서 16kHz, mono, PCM16(16비트 리틀엔디안)이라는 오디오 규격을 인터페이스 전체에서 하나로 고정했습니다. recognize에 넘기는 입력 프레임도, speak가 흘려보내는 출력 청크도 전부 같은 규격을 따릅니다. 이 부분에서 실시간 음성 시스템이 오래전부터 써온 MRCP(Media Resource Control Protocol, IETF RFC 6787) 같은 표준의 발상을 참고했습니다. 클라이언트가 리소스에 요청을 보내고 리소스가 비동기로 이벤트를 돌려주는 구조 자체가 이미 그쪽 세계에서 검증된 방식이었습니다.
자주 묻는 질문
Q. 이벤트 스트림 방식이 blocking 함수보다 코드가 복잡해지지 않나요.
호출부만 보면 await transcribe() 한 줄이 async for 루프보다 짧은 건 사실입니다. 하지만 발화 시작 감지, 최종 확정, 타임아웃을 blocking 함수 하나로 표현하려면 콜백이나 폴링을 따로 얹어야 해서 결국 복잡도가 사라지는 게 아니라 다른 곳으로 옮겨갈 뿐입니다. 이벤트 스트림은 그 복잡도를 인터페이스 안으로 끌어와 한곳에서 관리하게 해줍니다.
Q. 스텁 엔진과 실제 엔진을 같은 인터페이스로 묶으면 테스트가 실제 동작을 보장하나요.
아닙니다. 인터페이스가 같다는 건 오케스트레이터 코드가 수정 없이 돌아간다는 뜻이지, 스텁 테스트를 통과했다고 실제 엔진의 인식 정확도나 지연 시간까지 검증됐다는 뜻은 아닙니다. 스텁은 구조 검증용이고, 실제 엔진 검증은 별도로 필요합니다.
이번 편에서는 오케스트레이터가 엔진의 구체적인 구현을 몰라도 되게 만든 추상 인터페이스 설계를 정리했습니다. 다음 편에서는 이 인터페이스 뒤에서 실제로 스텁 엔진이 어떻게 이벤트를 만들어내는지, Echo·Beep 스텁의 구체적인 동작을 다루겠습니다.
음성봇 직접 만들기 시리즈 (전체 6편)
댓글
댓글 쓰기