5. 스트리밍 시스템에서의 청킹 + 발화 세그멘테이션

청크(Chunk)

오디오는 청크라는 크기 단위로 나누어서 처리한다.

마이크에서 나오는 오디오는 끊임없이 흐르는 연속 신호이다.

• 이걸 통째로 처리할 경우 Whisper가 답을 줄 때 까지 오디오가 계속 쌓이게 되어 실시간 처리가 불가능해진다.

[연속 오디오 신호]
──────────────────────────────────────────────────
  [30ms] [30ms] [30ms] [30ms] [30ms] [30ms] ...
    ↑      ↑      ↑      ↑
  청크1  청크2  청크3  청크4   →  VAD → Buffer → Whisper

• 따라서 위처럼 오디오를 일정 크기의 조각(청크)로 나누어 빠르게 연속으로 처리하는게 스트리밍 방식임.

https://mariblossom.tistory.com/177

• 위 글에 청크를 나누는 다양한 방식이 나와있음.

일반적인 chunking

# PipelineConfig
sample_rate: int = 16000        # 1초 = 16,000개 샘플
chunk_duration_ms: int = 30     # 청크 1개 = 30ms

@property
def chunk_samples(self) -> int:
    return int(self.sample_rate * self.chunk_duration_ms / 1000)
    # = 16000 × 30 / 1000 = 480 샘플

• 위 코드는 30ms별로 한 청크씩 청킹하는 방식을 채용했다.
• 즉, 청크 1개 = 480개의 float32 숫자 = 30ms 분량의 오디오이다.

청크 전달

# MicrophoneASRTest.run()
with sd.InputStream(
    samplerate=self.config.sample_rate,   # 16000Hz
    channels=self.config.channels,         # 1 (mono)
    dtype="float32",
    blocksize=self.config.chunk_samples,   # 480샘플마다 콜백 호출
    callback=self._audio_callback,         # ← 30ms마다 여기 호출됨
):

이렇게 나눈 청크를 Sounddevice가 넘겨주는데

• blocksize를 청크의 샘플 개수 (16000 x 30 / 1000)만큼 모이면 자동으로 오디오 콜백을 호출한다.

def _audio_callback(self, indata: np.ndarray, _frames, _time_info, status):
    # indata.shape = (480, 1)  ← (샘플수, 채널수)
    chunk = indata[:, 0].copy()  # shape: (480,) — mono float32 추출

    # VAD: 이 청크가 음성인지 묵음인지 판별
    indicator = "█" if self._core.vad.is_speech(chunk) else "·"

    # ASRCore로 전달 → 버퍼 축적
    self._core.push(chunk)

• 그리고 받아온 청크가 음성인지 묵음인지 로그로 판별하도록 설계
• VAD와 연동한다.

SpeechBuffer

# SpeechBuffer.push() 내부 로직

if is_speech:
    self._chunks.append(chunk)   # 음성 청크 → 버퍼에 추가

else:
    if self._is_speaking:
        self._silence_count += 1
        self._chunks.append(chunk)   # 묵음도 포함 (자연스러운 끝 처리)

# 발화 종료 조건: 묵음이 silence_chunks개 연속
# silence_chunks = 0.8s / 0.03s = 약 26개 청크
speech_ended = self._silence_count >= self.config.silence_chunks

• 해당 청크가 쌓여서 발화 되는 과정은 위와 같다.
• 음성 청크를 버퍼에 추가하며 쌓다가 묵음이 연속으로 감지되면 그때 청크를 합쳐 whisper로 넘긴다.

청킹 구조

[마이크]
  │
  │ ← 1단계: 처리 단위 청크 (30ms)
  ▼
[_audio_callback] → VAD 판별
  │
  │ ← 2단계: 전사 단위 청크 (묵음 기준)
  ▼
[SpeechBuffer] → Whisper

• 1단계: 30ms단위로 나누는 청크는 고정길이로 나누는 청크이다. 이건 묵음인지 발화인지 판별하기 위한 용도로 쓰인다.
• 2단계: 전사단위 청크는 문장의 묵음을 기준으로 청킹한다. 이전 글에서 말했듯 SileroVAD기반 청킹과 Energy기반청킹이 있다.

묵음 청킹 구조 길이에 따른 Trade-off

# 현재
silence_duration_s: float = 0.8   # 0.8초 묵음 → 전사 시작

# 줄이면
silence_duration_s: float = 0.3   # 0.3초 묵음 → 전사 시작

[안녕하세요 │0.3s묵음│ 저는 오늘]
            ↑ 여기서 분할 → Whisper 전송

vs

[안녕하세요 저는 오늘│0.8s묵음│]
                      ↑ 여기서 분할

• 잘게 나눌수록 실시간성이 오른다. 그러나

문제점

• 말하다가 숨쉬는 등의 묵음으로 불완전한 문맥이 넘어간다.
• 이로인해 짧은 전사가 넘어가게 된다.
• Whisper는 짧은 오디오에 취약한데 이에 따라 전사 품질이 감소된다.
• 또한, whisper추론 횟수가 늘어나 처리시간이 늘어날 수 있다.

묵음 시간을 줄이는대신…

현재 방식 (순차):
[발화 축적 중────────────][묵음 감지][Whisper 전사]
                                     ← 여기서야 시작

더 나은 방식 (점진적):
[발화 축적 중────────────]
        [중간 전사 시작──][묵음 감지][최종 전사 보정]
         ↑ 일정 길이 쌓이면 미리 시작

• 전사를 병렬로 실행하는 방식을 고려하자

발화 세그멘테이션

연속으로 들어오는 오디오 스트림 중에서 의미있는 발화 한 덩어리를 잘라내는 과정

ex) 묵음 →말소리1 → 숨소리 → 말소리2 → 묵음 → 말소리3 → 묵음 → 말소리 1, 2를 합쳐 하나의 발화로 생성한다.

whisper에게 30ms의 청크 조각을 하나만 넘겨주면 의미있는 전사가 불가능하다.

SpeechBuffer

@dataclass
class SpeechBuffer:
    config: PipelineConfig
    _chunks: list = field(default_factory=list)
    _silence_count: int = 0
    _is_speaking: bool = False

• 상태는 3개의 변수로 나눠진다.
  • _is_speaking: 현재 말하는가?(
  • _silence_count: 묵음이 얼마나 쌓여있는가?
  • _chunks: 그 리스트 내역

이때, 발화 종료를 판단하는 두 가지 조건

조건 1: 침묵 기반 분할

speech_ended = (
    self._is_speaking
    and self._silence_count >= self.config.silence_chunks
    # silence_duration_s=0.8s → 0.8s ÷ 0.03s = 약 27청크 연속 묵음
)

• 말하고 있는지 상태를 받아오고, 해당 상태일때, 카운트가 설정한 값(0.8s / 27청크)만큼 연속으로 묵음일때 분할한다.

조건 2: 버퍼 오버플로우

buffer_full = len(self._chunks) >= self.config.max_buffeR_chunks

• 설정한 최대 버퍼 크기보다 더 커질경우 분할한다.(강제 분할)

if self._is_speaking:
    self._silence_count += 1
    self._chunks.append(chunk)  # ← 묵음도 버퍼에 포함

• 여기서 보면 묵음 청크도 버퍼에 그냥 포함해버린다.
• → 그 이유는 Whisper가 문맥을 잃지 않고 해석하기 때문이다.

if duration >= self.config.min_speech_duration_s:
    return audio
# min_speech_duration_s=0.3s → 0.3초 미만은 그냥 버림

• 여기서 아주 작은 발화는 버려버린다.
  • Whisper는 짧은 오디오는 환각 문제를 심각하게 일으킴
  • 또한, 노이즈/불완전한 음절을 필터링 하는 수단이 되기도 함.
  • 불필요한 whisper를 부르는 연산을 방지한다.

위에서 구현한 발화 세그멘테이션을 나누는 알고리즘은

30ms 청크 입력
      ↓
  RMS VAD 판별
  ┌────┴────┐
발화      묵음
  ↓         ↓
버퍼에     is_speaking 중이면
추가       버퍼에 추가 + silence_count++
              ↓
         silence_count >= 27?
              ↓ YES
         전체 버퍼 flush → float32 배열
              ↓
         duration >= 0.3s?
              ↓ YES
         Whisper에게 전달 → 전사 시작

• 위와 같이 구성된다.