음성 챗봇 구축 가이드: STT/TTS 파이프라인과 실시간 음성 인터페이스 구현

사용자 음성 ──▶ [STT] ──▶ 텍스트 ──▶ [LLM] ──▶ 응답 텍스트 ──▶ [TTS] ──▶ 음성 응답
                 │                      │                         │
           Whisper/Deepgram     GPT-4o/Claude          ElevenLabs/Google TTS

import whisper
import numpy as np
import soundfile as sf

# 모델 로드 (turbo는 Large V3 Turbo를 의미)
model = whisper.load_model("turbo")

def transcribe_audio(audio_path: str, language: str = "ko") -> dict:
    """오디오 파일을 텍스트로 변환한다."""
    result = model.transcribe(
        audio_path,
        language=language,
        task="transcribe",
        fp16=True,  # GPU 사용 시 FP16 활성화
        beam_size=5,
        best_of=5,
        temperature=(0.0, 0.2, 0.4, 0.6, 0.8, 1.0),
    )
    return {
        "text": result["text"],
        "segments": [
            {
                "start": seg["start"],
                "end": seg["end"],
                "text": seg["text"],
                "confidence": seg.get("avg_logprob", 0),
            }
            for seg in result["segments"]
        ],
        "language": result["language"],
    }

# 사용 예시
result = transcribe_audio("user_input.wav", language="ko")
print(f"인식된 텍스트: {result['text']}")
for seg in result["segments"]:
    print(f"  [{seg['start']:.1f}s ~ {seg['end']:.1f}s] {seg['text']}")

from faster_whisper import WhisperModel

# INT8 양자화 모델 로드 (GPU VRAM 절약)
model = WhisperModel(
    "large-v3-turbo",
    device="cuda",
    compute_type="int8_float16",
    num_workers=4,
)

def transcribe_streaming_chunks(audio_chunks: list[np.ndarray]) -> str:
    """오디오 청크 리스트를 순차적으로 변환한다."""
    full_audio = np.concatenate(audio_chunks)
    segments, info = model.transcribe(
        full_audio,
        language="ko",
        beam_size=5,
        vad_filter=True,  # VAD 필터 활성화 - 무음 구간 자동 제거
        vad_parameters=dict(
            min_silence_duration_ms=500,
            speech_pad_ms=400,
        ),
    )

    transcription = ""
    for segment in segments:
        transcription += segment.text

    print(f"감지 언어: {info.language} (확률: {info.language_probability:.2f})")
    return transcription.strip()

from openai import OpenAI

client = OpenAI()

def transcribe_with_streaming(audio_file_path: str) -> str:
    """OpenAI API를 통해 스트리밍 방식으로 음성을 텍스트로 변환한다."""
    with open(audio_file_path, "rb") as audio_file:
        stream = client.audio.transcriptions.create(
            model="gpt-4o-mini-transcribe",
            file=audio_file,
            response_format="text",
            stream=True,
            language="ko",
        )

    full_text = ""
    for event in stream:
        if hasattr(event, "text"):
            full_text += event.text
            # 실시간으로 부분 결과를 UI에 표시할 수 있다
            print(f"[부분 결과] {event.text}", end="", flush=True)

    print()  # 줄바꿈
    return full_text

import httpx
import asyncio
from typing import AsyncGenerator

ELEVENLABS_API_KEY = "your-api-key"
VOICE_ID = "your-voice-id"

async def stream_tts_elevenlabs(
    text: str,
    model_id: str = "eleven_flash_v2_5",
) -> AsyncGenerator[bytes, None]:
    """ElevenLabs 스트리밍 TTS로 텍스트를 음성 청크로 변환한다."""
    url = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID}/stream"

    headers = {
        "xi-api-key": ELEVENLABS_API_KEY,
        "Content-Type": "application/json",
    }

    payload = {
        "text": text,
        "model_id": model_id,
        "voice_settings": {
            "stability": 0.5,
            "similarity_boost": 0.75,
            "style": 0.3,
            "use_speaker_boost": True,
        },
        "output_format": "pcm_24000",  # 24kHz PCM for low latency
    }

    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST", url, json=payload, headers=headers, timeout=30.0
        ) as response:
            response.raise_for_status()
            async for chunk in response.aiter_bytes(chunk_size=4096):
                if chunk:
                    yield chunk


async def synthesize_and_play(text: str):
    """텍스트를 음성으로 합성하고 오디오 버퍼에 축적한다."""
    audio_buffer = bytearray()
    chunk_count = 0

    async for audio_chunk in stream_tts_elevenlabs(text):
        audio_buffer.extend(audio_chunk)
        chunk_count += 1

        # 첫 번째 청크 도착 시 재생 시작 가능
        if chunk_count == 1:
            print(f"첫 번째 오디오 청크 수신 - 재생 시작 가능")

    print(f"총 {chunk_count}개 청크, {len(audio_buffer)} 바이트 수신 완료")
    return bytes(audio_buffer)

from openai import OpenAI
from pathlib import Path

client = OpenAI()

def synthesize_with_emotion(
    text: str,
    emotion_prompt: str = "친근하고 따뜻한 톤으로, 자연스럽게 말하세요.",
    output_path: str = "response.mp3",
) -> Path:
    """감정 프롬프트를 적용한 TTS 합성을 수행한다."""
    response = client.audio.speech.create(
        model="gpt-4o-mini-tts",
        voice="coral",
        input=text,
        instructions=emotion_prompt,
        response_format="mp3",
        speed=1.0,
    )

    output = Path(output_path)
    response.stream_to_file(output)
    return output

# 상황별 감정 프롬프트 사전
EMOTION_PROMPTS = {
    "greeting": "밝고 활기찬 톤으로, 친근하게 인사하세요.",
    "apology": "진심 어린 사과의 톤으로, 차분하고 성실하게 말하세요.",
    "explanation": "명확하고 교육적인 톤으로, 천천히 또박또박 설명하세요.",
    "confirmation": "확신에 찬 톤으로, 간결하고 명확하게 확인해주세요.",
    "empathy": "공감하는 따뜻한 톤으로, 상대방의 감정을 이해한다는 느낌으로 말하세요.",
}

import asyncio
import json
import base64
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from faster_whisper import WhisperModel
import numpy as np

app = FastAPI()

# Whisper 모델을 전역으로 로드 (서버 시작 시 1회)
stt_model = WhisperModel("large-v3-turbo", device="cuda", compute_type="int8_float16")

class VoiceSession:
    """음성 세션 상태를 관리한다."""

    def __init__(self, session_id: str):
        self.session_id = session_id
        self.audio_buffer = bytearray()
        self.is_speaking = False
        self.silence_duration_ms = 0
        self.sample_rate = 16000
        self.chunk_size = 1024

    def add_audio_chunk(self, chunk: bytes) -> None:
        """오디오 청크를 버퍼에 추가한다."""
        self.audio_buffer.extend(chunk)

    def get_audio_as_numpy(self) -> np.ndarray:
        """버퍼의 오디오를 numpy 배열로 변환한다."""
        audio = np.frombuffer(bytes(self.audio_buffer), dtype=np.int16)
        return audio.astype(np.float32) / 32768.0  # 정규화

    def clear_buffer(self) -> None:
        """오디오 버퍼를 초기화한다."""
        self.audio_buffer = bytearray()


@app.websocket("/ws/voice/{session_id}")
async def voice_websocket(websocket: WebSocket, session_id: str):
    """실시간 음성 챗봇 WebSocket 엔드포인트."""
    await websocket.accept()
    session = VoiceSession(session_id)

    try:
        # 세션 초기화 메시지 전송
        await websocket.send_json({
            "type": "session.created",
            "session_id": session_id,
            "config": {
                "sample_rate": session.sample_rate,
                "chunk_size": session.chunk_size,
                "encoding": "pcm_s16le",
            },
        })

        while True:
            # 클라이언트로부터 메시지 수신
            message = await websocket.receive_json()
            msg_type = message.get("type")

            if msg_type == "audio.chunk":
                # Base64 인코딩된 오디오 청크 처리
                audio_data = base64.b64decode(message["data"])
                session.add_audio_chunk(audio_data)

            elif msg_type == "audio.commit":
                # 사용자가 말을 마침 - STT 실행
                audio_np = session.get_audio_as_numpy()

                if len(audio_np) < session.sample_rate * 0.3:
                    # 0.3초 미만의 오디오는 무시
                    await websocket.send_json({
                        "type": "error",
                        "message": "오디오가 너무 짧습니다.",
                    })
                    session.clear_buffer()
                    continue

                # STT 변환
                await websocket.send_json({"type": "stt.started"})
                segments, info = stt_model.transcribe(
                    audio_np, language="ko", beam_size=3
                )
                transcript = "".join(seg.text for seg in segments).strip()

                await websocket.send_json({
                    "type": "stt.completed",
                    "transcript": transcript,
                    "language": info.language,
                })

                if transcript:
                    # LLM 응답 생성 (별도 함수에서 처리)
                    await process_and_respond(websocket, session, transcript)

                session.clear_buffer()

            elif msg_type == "session.close":
                break

    except WebSocketDisconnect:
        print(f"세션 {session_id} 연결 종료")
    except Exception as e:
        await websocket.send_json({
            "type": "error",
            "message": str(e),
        })


async def process_and_respond(
    websocket: WebSocket,
    session: VoiceSession,
    transcript: str,
) -> None:
    """사용자 발화를 처리하고 음성 응답을 스트리밍한다."""
    # 1. LLM 응답 생성 (스트리밍)
    await websocket.send_json({"type": "llm.started"})

    llm_response = await generate_llm_response(transcript)

    await websocket.send_json({
        "type": "llm.completed",
        "text": llm_response,
    })

    # 2. TTS 합성 및 오디오 스트리밍
    await websocket.send_json({"type": "tts.started"})

    chunk_index = 0
    async for audio_chunk in stream_tts_elevenlabs(llm_response):
        await websocket.send_json({
            "type": "audio.delta",
            "data": base64.b64encode(audio_chunk).decode("utf-8"),
            "index": chunk_index,
        })
        chunk_index += 1

    await websocket.send_json({"type": "tts.completed"})

interface VoiceConfig {
  sampleRate: number
  chunkSize: number
  encoding: string
}

interface ServerMessage {
  type: string
  data?: string
  transcript?: string
  text?: string
  message?: string
  index?: number
  config?: VoiceConfig
  session_id?: string
}

class VoiceChatClient {
  private ws: WebSocket | null = null
  private mediaStream: MediaStream | null = null
  private audioContext: AudioContext | null = null
  private processor: ScriptProcessorNode | null = null
  private isRecording = false
  private audioQueue: ArrayBuffer[] = []
  private isPlaying = false

  constructor(
    private serverUrl: string,
    private sessionId: string,
    private onTranscript: (text: string) => void,
    private onResponse: (text: string) => void,
    private onStatusChange: (status: string) => void
  ) {}

  async connect(): Promise<void> {
    this.ws = new WebSocket(`${this.serverUrl}/ws/voice/${this.sessionId}`)

    this.ws.onopen = () => {
      this.onStatusChange('connected')
      console.log('WebSocket 연결 완료')
    }

    this.ws.onmessage = (event: MessageEvent) => {
      const message: ServerMessage = JSON.parse(event.data)
      this.handleServerMessage(message)
    }

    this.ws.onerror = (error: Event) => {
      console.error('WebSocket 오류:', error)
      this.onStatusChange('error')
    }

    this.ws.onclose = () => {
      this.onStatusChange('disconnected')
      this.scheduleReconnect()
    }
  }

  private handleServerMessage(message: ServerMessage): void {
    switch (message.type) {
      case 'session.created':
        console.log(`세션 생성됨: ${message.session_id}`)
        break

      case 'stt.completed':
        if (message.transcript) {
          this.onTranscript(message.transcript)
        }
        break

      case 'llm.completed':
        if (message.text) {
          this.onResponse(message.text)
        }
        break

      case 'audio.delta':
        if (message.data) {
          // Base64 오디오 청크를 디코딩하여 재생 큐에 추가
          const audioData = this.base64ToArrayBuffer(message.data)
          this.audioQueue.push(audioData)
          if (!this.isPlaying) {
            this.playAudioQueue()
          }
        }
        break

      case 'tts.completed':
        this.onStatusChange('ready')
        break

      case 'error':
        console.error('서버 오류:', message.message)
        this.onStatusChange('error')
        break
    }
  }

  async startRecording(): Promise<void> {
    this.mediaStream = await navigator.mediaDevices.getUserMedia({
      audio: {
        sampleRate: 16000,
        channelCount: 1,
        echoCancellation: true,
        noiseSuppression: true,
        autoGainControl: true,
      },
    })

    this.audioContext = new AudioContext({ sampleRate: 16000 })
    const source = this.audioContext.createMediaStreamSource(this.mediaStream)

    // ScriptProcessorNode로 오디오 청크를 캡처
    this.processor = this.audioContext.createScriptProcessor(4096, 1, 1)
    this.processor.onaudioprocess = (event: AudioProcessingEvent) => {
      if (!this.isRecording) return

      const inputData = event.inputBuffer.getChannelData(0)
      // Float32 -> Int16 변환
      const pcmData = new Int16Array(inputData.length)
      for (let i = 0; i < inputData.length; i++) {
        const s = Math.max(-1, Math.min(1, inputData[i]))
        pcmData[i] = s < 0 ? s * 0x8000 : s * 0x7fff
      }

      // Base64 인코딩 후 서버로 전송
      const base64Data = this.arrayBufferToBase64(pcmData.buffer)
      this.ws?.send(
        JSON.stringify({
          type: 'audio.chunk',
          data: base64Data,
        })
      )
    }

    source.connect(this.processor)
    this.processor.connect(this.audioContext.destination)
    this.isRecording = true
    this.onStatusChange('recording')
  }

  stopRecording(): void {
    this.isRecording = false
    // 서버에 오디오 커밋 신호 전송
    this.ws?.send(JSON.stringify({ type: 'audio.commit' }))
    this.onStatusChange('processing')

    // 미디어 스트림 정리
    this.mediaStream?.getTracks().forEach((track) => track.stop())
    this.processor?.disconnect()
  }

  private async playAudioQueue(): Promise<void> {
    this.isPlaying = true
    while (this.audioQueue.length > 0) {
      const chunk = this.audioQueue.shift()!
      await this.playAudioChunk(chunk)
    }
    this.isPlaying = false
  }

  private playAudioChunk(data: ArrayBuffer): Promise<void> {
    return new Promise((resolve) => {
      if (!this.audioContext) {
        resolve()
        return
      }
      // PCM 데이터를 AudioBuffer로 변환하여 재생
      const int16Data = new Int16Array(data)
      const floatData = new Float32Array(int16Data.length)
      for (let i = 0; i < int16Data.length; i++) {
        floatData[i] = int16Data[i] / 32768.0
      }
      const buffer = this.audioContext.createBuffer(1, floatData.length, 24000)
      buffer.copyToChannel(floatData, 0)

      const source = this.audioContext.createBufferSource()
      source.buffer = buffer
      source.connect(this.audioContext.destination)
      source.onended = () => resolve()
      source.start()
    })
  }

  private scheduleReconnect(): void {
    setTimeout(() => {
      console.log('WebSocket 재연결 시도...')
      this.connect()
    }, 3000)
  }

  private base64ToArrayBuffer(base64: string): ArrayBuffer {
    const binary = atob(base64)
    const bytes = new Uint8Array(binary.length)
    for (let i = 0; i < binary.length; i++) {
      bytes[i] = binary.charCodeAt(i)
    }
    return bytes.buffer
  }

  private arrayBufferToBase64(buffer: ArrayBuffer): string {
    const bytes = new Uint8Array(buffer)
    let binary = ''
    for (let i = 0; i < bytes.length; i++) {
      binary += String.fromCharCode(bytes[i])
    }
    return btoa(binary)
  }

  disconnect(): void {
    this.ws?.send(JSON.stringify({ type: 'session.close' }))
    this.ws?.close()
    this.mediaStream?.getTracks().forEach((track) => track.stop())
    this.audioContext?.close()
  }
}

// 사용 예시
const client = new VoiceChatClient(
  'wss://api.example.com',
  'session-001',
  (transcript) => console.log('사용자:', transcript),
  (response) => console.log('챗봇:', response),
  (status) => console.log('상태:', status)
)

await client.connect()

import asyncio
import re
from openai import AsyncOpenAI
from typing import AsyncGenerator

llm_client = AsyncOpenAI()

# 문장 종결 패턴 (한국어 + 영어)
SENTENCE_END_PATTERN = re.compile(r'[.!?。！？]\s*|[다요죠니까네세]\.\s*')

async def generate_llm_response_streaming(
    transcript: str,
    conversation_history: list[dict],
) -> AsyncGenerator[str, None]:
    """LLM 응답을 문장 단위로 스트리밍한다."""

    conversation_history.append({"role": "user", "content": transcript})

    stream = await llm_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "당신은 친절한 음성 비서입니다. "
                    "응답은 2-3문장으로 간결하게 하세요. "
                    "리스트나 마크다운 형식은 사용하지 마세요. "
                    "자연스러운 구어체로 답변하세요."
                ),
            },
            *conversation_history,
        ],
        stream=True,
        max_tokens=300,
        temperature=0.7,
    )

    sentence_buffer = ""
    async for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            sentence_buffer += delta.content

            # 문장 종결 패턴 탐지
            match = SENTENCE_END_PATTERN.search(sentence_buffer)
            if match:
                end_pos = match.end()
                complete_sentence = sentence_buffer[:end_pos].strip()
                sentence_buffer = sentence_buffer[end_pos:]

                if complete_sentence:
                    yield complete_sentence

    # 남은 텍스트 처리
    if sentence_buffer.strip():
        yield sentence_buffer.strip()


async def full_pipeline(
    transcript: str,
    conversation_history: list[dict],
    websocket,
) -> None:
    """STT 결과 -> LLM -> TTS 전체 파이프라인을 실행한다."""

    full_response = ""

    async for sentence in generate_llm_response_streaming(
        transcript, conversation_history
    ):
        full_response += sentence + " "

        # 각 문장을 즉시 TTS로 변환하여 스트리밍
        async for audio_chunk in stream_tts_elevenlabs(sentence):
            await websocket.send_json({
                "type": "audio.delta",
                "data": base64.b64encode(audio_chunk).decode("utf-8"),
            })

    # 대화 이력 업데이트
    conversation_history.append({
        "role": "assistant",
        "content": full_response.strip(),
    })

import torch
import numpy as np

class SileroVAD:
    """Silero VAD 모델을 사용한 음성 활동 감지기."""

    def __init__(
        self,
        threshold: float = 0.5,
        min_silence_ms: int = 500,
        speech_pad_ms: int = 300,
        sample_rate: int = 16000,
    ):
        # Silero VAD 모델 로드
        self.model, self.utils = torch.hub.load(
            repo_or_dir="snakers4/silero-vad",
            model="silero_vad",
            trust_repo=True,
        )
        self.threshold = threshold
        self.min_silence_ms = min_silence_ms
        self.speech_pad_ms = speech_pad_ms
        self.sample_rate = sample_rate

        # 상태 변수
        self.is_speaking = False
        self.silence_start_ms = 0
        self.speech_start_ms = 0
        self.current_ms = 0

    def process_chunk(self, audio_chunk: np.ndarray) -> dict:
        """오디오 청크를 분석하여 발화 상태를 반환한다."""
        # numpy -> torch 텐서 변환
        tensor = torch.from_numpy(audio_chunk).float()

        # VAD 확률 계산
        speech_prob = self.model(tensor, self.sample_rate).item()

        chunk_duration_ms = len(audio_chunk) / self.sample_rate * 1000
        self.current_ms += chunk_duration_ms

        result = {
            "speech_probability": speech_prob,
            "is_speech": speech_prob >= self.threshold,
            "event": None,
        }

        if speech_prob >= self.threshold:
            if not self.is_speaking:
                # 발화 시작 감지
                self.is_speaking = True
                self.speech_start_ms = self.current_ms
                result["event"] = "speech_start"
            self.silence_start_ms = 0

        else:
            if self.is_speaking:
                if self.silence_start_ms == 0:
                    self.silence_start_ms = self.current_ms

                silence_duration = self.current_ms - self.silence_start_ms

                if silence_duration >= self.min_silence_ms:
                    # 발화 종료 감지
                    self.is_speaking = False
                    result["event"] = "speech_end"
                    result["speech_duration_ms"] = (
                        self.current_ms - self.speech_start_ms
                    )

        return result


# VAD를 WebSocket 서버에 통합하는 예시
async def vad_enabled_voice_handler(websocket, session):
    """VAD를 활용하여 발화 종료를 자동 감지하는 핸들러."""
    vad = SileroVAD(
        threshold=0.5,
        min_silence_ms=600,  # 600ms 침묵 시 발화 종료로 판단
    )

    while True:
        message = await websocket.receive_json()

        if message["type"] == "audio.chunk":
            audio_data = base64.b64decode(message["data"])
            audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0

            session.add_audio_chunk(audio_data)
            vad_result = vad.process_chunk(audio_np)

            if vad_result["event"] == "speech_start":
                await websocket.send_json({"type": "vad.speech_start"})

            elif vad_result["event"] == "speech_end":
                await websocket.send_json({
                    "type": "vad.speech_end",
                    "duration_ms": vad_result["speech_duration_ms"],
                })
                # 자동으로 STT 및 응답 파이프라인 트리거
                audio_np_full = session.get_audio_as_numpy()
                segments, info = stt_model.transcribe(audio_np_full, language="ko")
                transcript = "".join(seg.text for seg in segments).strip()

                if transcript:
                    await process_and_respond(websocket, session, transcript)
                session.clear_buffer()

async def robust_stt_pipeline(
    audio_np: np.ndarray,
    websocket,
    max_retries: int = 2,
) -> str:
    """오류 복구 로직이 포함된 STT 파이프라인."""

    # 1단계: 오디오 품질 사전 검증
    rms_energy = np.sqrt(np.mean(audio_np ** 2))
    peak_amplitude = np.max(np.abs(audio_np))

    if rms_energy < 0.001:
        await websocket.send_json({
            "type": "stt.warning",
            "message": "마이크 입력이 감지되지 않습니다. 마이크를 확인해주세요.",
        })
        return ""

    if peak_amplitude > 0.99:
        # 클리핑 감지 - 볼륨 정규화 적용
        audio_np = audio_np * (0.9 / peak_amplitude)

    # 2단계: 1차 STT 시도
    segments, info = stt_model.transcribe(audio_np, language="ko", beam_size=5)
    transcript = "".join(seg.text for seg in segments).strip()

    # 3단계: 신뢰도 검증
    avg_logprob = np.mean([seg.avg_log_prob for seg in segments]) if segments else -10

    if avg_logprob < -1.0 and max_retries > 0:
        # 낮은 신뢰도 - 언어 자동 감지로 재시도
        segments_retry, info_retry = stt_model.transcribe(
            audio_np, language=None, beam_size=5
        )
        transcript_retry = "".join(seg.text for seg in segments_retry).strip()

        avg_logprob_retry = (
            np.mean([seg.avg_log_prob for seg in segments_retry])
            if segments_retry else -10
        )

        if avg_logprob_retry > avg_logprob:
            transcript = transcript_retry
            await websocket.send_json({
                "type": "stt.info",
                "message": f"언어 자동 감지됨: {info_retry.language}",
            })

    # 4단계: 빈 결과 처리
    if not transcript:
        await websocket.send_json({
            "type": "stt.empty",
            "message": "음성을 인식하지 못했습니다. 다시 말씀해주세요.",
        })

    return transcript

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # 정상 상태
    OPEN = "open"          # 차단 상태
    HALF_OPEN = "half_open"  # 시험 상태

class TTSCircuitBreaker:
    """TTS 서비스에 대한 서킷 브레이커."""

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 30,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = 0.0

    async def call(self, primary_fn, fallback_fn, *args, **kwargs):
        """서킷 브레이커를 통해 TTS 호출을 실행한다."""

        if self.state == CircuitState.OPEN:
            # 복구 타임아웃 경과 확인
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                # 폴백 TTS 사용
                return await fallback_fn(*args, **kwargs)

        try:
            result = await primary_fn(*args, **kwargs)

            if self.state == CircuitState.HALF_OPEN:
                # 복구 성공
                self.state = CircuitState.CLOSED
                self.failure_count = 0

            return result

        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()

            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
                print(
                    f"서킷 브레이커 OPEN: {self.failure_count}회 연속 실패. "
                    f"{self.recovery_timeout}초 후 재시도."
                )

            # 폴백 TTS 사용
            return await fallback_fn(*args, **kwargs)