실시간 음성 에이전트 2026년 2월판: STT/TTS를 “파이프라인”이 아닌 “스트리밍 런타임”으로 다루는 법

게시 2026/02/05

By Daewook Kwon

14 분읽는 시간

들어가며

2026년 2월 기준, “음성 AI”의 승부처는 모델 성능 자체보다 대화의 리듬(턴 전환, 끼어들기, 지연, 끊김 복구) 입니다. 텍스트 챗봇은 1~2초 지연도 용납되지만, 음성 대화는 첫 소리(First audio)까지 600~900ms, 턴 전체가 1~2초대를 넘기면 사용자가 즉시 “기계랑 말한다”는 느낌을 받습니다. (현업에서 흔히 보는 3초+ 지연은 대개 “버퍼링/턴 감지” 설계 문제로 귀결됩니다.)

최근 트렌드는 명확합니다.

Speech-to-Speech(=STS) / Realtime 모델 + 스트리밍 프로토콜(WebRTC/WebSocket) 로 “텍스트 중간 단계를 최소화” (OpenAI Realtime/Voice Agents, Azure Realtime) (openai.github.io)
VAD(Voice Activity Detection) 기반의 barge-in(끼어들기) + 서버가 즉시 generation cancel (Gemini Live API) (ai.google.dev)
STT/TTS/LLM orchestration을 한 API로 묶어 지연과 복잡도를 줄이는 통합형 Voice Agent API (Deepgram Voice Agent API GA) (deepgram.com)
엔터프라이즈는 “내부망/클라우드 경계” 때문에 SageMaker 실시간 엔드포인트 같은 배포 경로를 선호 (Deepgram on SageMaker) (press.aboutamazon.com)

이 글은 “실시간 음성 대화 구현”을 목표로, 턴 설계/스트리밍/인터럽트를 중심으로 구조를 잡아봅니다.

🔧 핵심 개념

1) 파이프라인(ASR→LLM→TTS) vs Realtime/STS 런타임

전통 구조는 다음과 같습니다.

1) STT가 문장을 “확정(final)”
2) LLM이 텍스트 응답 생성
3) TTS가 음성 합성

문제는 (1)에서 final을 기다리는 순간 대화가 죽는다는 점입니다. 그래서 2025~2026의 실시간 API들은 다음을 기본으로 깝니다.

Streaming input: 20ms~60ms 단위 오디오 프레임을 지속 전송
Partial 결과: partial transcript/partial audio를 즉시 받음
Interrupt: 사용자가 말하면 서버가 현재 생성 중 응답을 “취소(cancellation)”

Gemini Live API 문서에선 VAD로 끼어들기를 감지하면 진행 중 generation을 cancel/discard하고, 취소된 tool call까지 정리한다고 명시합니다. (ai.google.dev)

Azure OpenAI(=GPT Realtime) 쪽은 특히 WebRTC가 저지연에 유리하며 WebSocket은 “서버-서버 시나리오”에 더 가깝다고 가이드합니다. 즉, 브라우저/모바일 실시간 음성은 WebRTC를 우선 고려하는 게 맞습니다. (learn.microsoft.com)

OpenAI Agents SDK의 Voice Agents도 WebSocket/WebRTC 연결과 interruption handling을 “SDK 레벨 기능”으로 올려둔 게 포인트입니다. (openai.github.io)

2) 지연을 쪼개서 관리하라 (Latency Budget)

실시간 음성 에이전트의 지연은 대개 다음 합입니다.

Capture/Encode: 마이크 캡처 + Opus/PCM 인코딩
Network jitter: 업링크 지터/손실
VAD 결정 지연: “사용자 발화가 끝났다”를 판단하는 hangover
Model thinking: 첫 토큰/첫 오디오 청크
Playback buffer: 재생측 버퍼(너무 두껍게 잡으면 체감 지연 급증)

여기서 가장 흔한 함정은:

VAD를 너무 보수적으로 잡아 턴 종료를 늦게 확정
서버가 “사용자 오디오를 일정 길이로 모아서” 보내 가짜 스트리밍
재생 버퍼를 안전하게 잡다가 대화 리듬 파괴

3) “에이전트”는 음성만이 아니라 Tool/State 머신이다

요즘 Voice Agent API들이 “STT/TTS + orchestration”을 강조하는 이유는 단순합니다.
실전 음성봇은 결국:

상태(state): 인증/주문/예약/CS 티켓 등
툴(tool): DB 조회, CRM 업데이트, 검색, 결제, 콜 라우팅
가드레일(guardrails): 금칙어/개인정보/정책

Deepgram은 Voice Agent API를 단일 스트리밍 API로 통합해 latency/복잡도를 줄였다고 강조합니다. (deepgram.com)

💻 실전 코드

아래 예시는 “WebSocket 기반 실시간 음성 세션”의 최소 골격입니다. (브라우저 실시간은 WebRTC 권장이라는 점은 유지하되, 서버에서 빠르게 검증하기 좋은 형태로 작성) Azure OpenAI Realtime WebSocket 구조(세션/이벤트 기반)와 동일한 “이벤트 스트림” 패턴으로 설계합니다. (learn.microsoft.com)

예제 목표:

마이크 입력(PCM 16k) → 서버 → Realtime API
응답 오디오 chunk 수신 → 즉시 재생 큐에 적재
사용자 발화 감지 시 cancel(인터럽트) 를 넣을 수 있는 구조

  
# python 3.11+
# pip install websockets sounddevice numpy

import asyncio, base64, json
import numpy as np
import sounddevice as sd
import websockets

SAMPLE_RATE = 16000
FRAME_MS = 20
FRAME_SAMPLES = SAMPLE_RATE * FRAME_MS // 1000

# Azure OpenAI Realtime(WebSocket) 연결 예시 형식:
# wss://{resource}.openai.azure.com/openai/v1/realtime?model={deployment}
# (환경에 따라 api-version 파라미터가 필요할 수 있음)  ([learn.microsoft.com](https://learn.microsoft.com/en-us/azure/ai-foundry/openai/how-to/realtime-audio-websockets?utm_source=openai))
AZURE_WSS_URL = "wss://YOUR_RESOURCE.openai.azure.com/openai/v1/realtime?model=YOUR_DEPLOYMENT"
AZURE_API_KEY = "YOUR_API_KEY"

def pcm16_to_b64(pcm16: np.ndarray) -> str:
    # pcm16: int16 mono
    return base64.b64encode(pcm16.tobytes()).decode("ascii")

async def mic_stream(q: asyncio.Queue):
    # 마이크 캡처는 별도 스레드 콜백으로 들어오므로 asyncio queue로 브릿지
    loop = asyncio.get_running_loop()

    def callback(indata, frames, time, status):
        if status:
            # 실제 서비스라면 로깅/메트릭
            pass
        pcm16 = (indata[:, 0] * 32767.0).astype(np.int16)
        loop.call_soon_threadsafe(q.put_nowait, pcm16)

    with sd.InputStream(
        channels=1,
        samplerate=SAMPLE_RATE,
        blocksize=FRAME_SAMPLES,
        dtype="float32",
        callback=callback,
    ):
        while True:
            await asyncio.sleep(1)

async def run():
    mic_q = asyncio.Queue()

    # 1) 마이크 태스크 시작
    asyncio.create_task(mic_stream(mic_q))

    headers = {
        # WebSocket 핸드셰이크 헤더로 api-key 전달(브라우저는 불가한 경우가 많음) ([learn.microsoft.com](https://learn.microsoft.com/en-us/azure/ai-foundry/openai/how-to/realtime-audio-websockets?utm_source=openai))
        "api-key": AZURE_API_KEY,
    }

    async with websockets.connect(AZURE_WSS_URL, extra_headers=headers) as ws:
        # 2) (선택) 세션 설정 이벤트 전송
        # 실제 이벤트 명/스키마는 제공자별로 다르지만,
        # 핵심은 "입력 오디오 포맷/출력 모달리티(audio)"를 명시하는 것.
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "input_audio_format": {"type": "pcm16", "sample_rate_hz": SAMPLE_RATE},
                "output_audio_format": {"type": "pcm16", "sample_rate_hz": SAMPLE_RATE},
                "response_modalities": ["audio"],
            }
        }))

        async def sender():
            while True:
                pcm16 = await mic_q.get()
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": pcm16_to_b64(pcm16),
                }))

        async def receiver():
            while True:
                msg = json.loads(await ws.recv())

                # 3) 서버가 보내는 스트리밍 이벤트를 받아 처리
                t = msg.get("type")

                if t == "response.audio.delta":
                    # 응답 오디오 조각(PCM16 base64)을 바로 재생 큐로 넣는 구조를 권장
                    audio_b64 = msg["delta"]
                    pcm = np.frombuffer(base64.b64decode(audio_b64), dtype=np.int16)
                    sd.play(pcm.astype(np.float32) / 32767.0, SAMPLE_RATE, blocking=False)

                elif t == "response.completed":
                    # 한 턴 완료
                    pass

                elif t == "input_audio_buffer.speech_started":
                    # 사용자 발화 시작을 감지했다면, 지금 재생 중인 TTS를 멈추고
                    # 생성 중 응답을 취소하는 "barge-in"을 구현할 수 있음
                    # (Gemini Live는 VAD 인터럽트 시 생성 취소를 명시) ([ai.google.dev](https://ai.google.dev/gemini-api/docs/live-guide))
                    await ws.send(json.dumps({"type": "response.cancel"}))

        await asyncio.gather(sender(), receiver())

if __name__ == "__main__":
    asyncio.run(run())

포인트:

“실시간”은 append(오디오 프레임) → delta(오디오 청크) 가 끊기지 않는 게 핵심입니다.
인터럽트는 “UI에서 버튼”이 아니라 VAD 이벤트를 트리거로 자동화해야 체감이 좋아집니다.
브라우저/모바일이라면 WebRTC로 옮기고, 서버는 인증/툴 실행/로깅만 담당시키는 구성이 일반적으로 더 낮은 지연을 얻습니다. (learn.microsoft.com)

⚡ 실전 팁

1) VAD hangover(침묵 판정) 튜닝이 대화 리듬을 좌우

턴 종료를 너무 늦게 잡으면 응답이 느려지고,
너무 빠르게 잡으면 말을 끊고 끼어드는 느낌이 납니다.
“사용자 발화 시작” 감지는 공격적으로, “발화 종료”는 약간 보수적으로(짧은 hangover) 가져가는 게 경험상 안정적입니다.

2) 재생 버퍼는 짧게, 대신 끊김 복구 전략을

버퍼를 길게 잡아 끊김을 숨기면 지연이 커져 UX가 망가집니다.
짧은 버퍼 + 패킷 손실 시 “짧은 무음/타임스트레치” 같은 완충을 두세요. (WebRTC가 이 쪽에 강함) (learn.microsoft.com)

3) Tool calling은 “음성 턴”과 분리해서 설계

음성 응답을 길게 끌면서 중간에 DB/외부 API를 때리면 지연이 폭발합니다.
패턴 추천:
- (a) 먼저 짧게 “확인 멘트”를 음성으로 내보내고
- (b) 백그라운드로 툴 실행
- (c) 결과가 오면 다음 턴에서 확정 답변

4) 통합형 Voice Agent API vs BYO 파이프라인 선택 기준

빠른 출시/운영 단순화: 통합형(예: Deepgram Voice Agent API GA) (deepgram.com)
커스텀 모델/정교한 정책/온프레미스 제약: BYO 파이프라인(+SageMaker/사내 배포 경로) (press.aboutamazon.com)

5) “3초 지연”의 80%는 진짜 모델이 아니라 구현

커뮤니티에서도 end-to-end 3초+는 “버퍼링/가짜 스트리밍”을 의심하라는 얘기가 반복됩니다(물론 경험담 수준).
측정은 반드시 구간별로:
- last user audio frame → first server ack
- VAD end → first model audio delta
- first delta → speaker output

🚀 마무리

2026년 2월의 실시간 음성 에이전트는 “STT/TTS 붙이면 끝”이 아니라, Streaming + VAD + Interrupt + Tool/State 를 하나의 런타임으로 다루는 싸움입니다.
정리하면:

저지연 목표라면 WebRTC 우선, WebSocket은 서버-서버/단순 검증에 적합 (learn.microsoft.com)
끼어들기(barge-in) 는 필수 기능이며, VAD 기반 cancel이 핵심 (ai.google.dev)
통합형 Voice Agent API가 “복잡도/지연”을 실제로 줄여주는 구간이 있고(Deepgram), 엔터프라이즈는 배포 경로까지 같이 봅니다. (deepgram.com)

다음 학습 추천:

WebRTC 기반 실시간 오디오 파이프라인(Opus, jitter buffer, echo cancellation)
이벤트 기반 Realtime API(세션/대화/취소/partial) 상태 머신 설계
음성 UX(턴 길이, 백채널링 “네/좋아요” 같은 짧은 음성 토큰) 실험

원하시면, 위 예제를 브라우저(WebRTC) + 서버(툴 실행) + 음성 인터럽트까지 포함한 “프로덕션 아키텍처”로 확장한 버전(구성도/메트릭/부하 테스트 체크리스트 포함)으로 이어서 작성해드릴게요.

AI, Multimodal

ai multimodal trend 2026-02