feat(voice): STT 환각(방송 클로징 등) 자동 제거

ai/CLAUDE.md

@@ -193,6 +193,7 @@ chain = prompt | llm | PydanticOutputParser(pydantic_object=...)
 - **TTS: Gemini TTS 기본(한국어)** (`voice/tts/`) — 질문(INTERVIEWER) 메시지 음성화. Deepgram/OpenAI TTS 는 한국어 미지원이라 `GeminiTtsProvider`(`gemini-2.5-flash-preview-tts`, voice=Kore)가 기본. Gemini 는 raw PCM(L16/24kHz)을 반환하므로 WAV 로 감싸 `audio/wav` 로 저장. `TtsProvider` 추상화 + `GatewayTtsProvider`(Mindlogic 게이트웨이 `/audio/speech`, LLM_API_KEY, raw PCM→WAV)/`GeminiTtsProvider`(직접 GEMINI_API_KEY)/`OpenAiTtsProvider`(`gpt-4o-mini-tts`, mp3)/`MockTtsProvider`, `build_tts_provider` factory(`TTS_PROVIDER=auto`면 **LLM_API_KEY(gateway) > GEMINI_API_KEY > OPENAI_API_KEY** 순 — 게이트웨이 우선으로 직접 키 429 부하 분산). gateway/gemini 는 `gemini_tts_model`/`gemini_tts_voice` 공유. `generate.tts` consumer 가 합성 → S3 PUT(확장자는 content_type 기준) → `callback.tts` 발행. 재생은 Core 오디오 프록시(`GET /api/sessions/{sid}/messages/{mid}/audio`) 경유(MinIO presigned URL 이 내부 호스트라 브라우저 직접 접근 불가).
 - **스트리밍 STT (실시간 음성 답변, RT3): Deepgram Live** (`voice/stt/deepgram_live.py`) — `websockets`로 Deepgram WS(`wss://api.deepgram.com/v1/listen`, nova-2)에 연결, interim/final 자막을 실시간 반환. `voice/stt/live.py`(`LiveSttProvider`/`LiveSttSession` 추상) + `voice/stt/mock_live.py`(키 없을 때 fallback) + `voice/stt/live_factory.py`(`LIVE_STT_PROVIDER=auto`면 DEEPGRAM_API_KEY 보유 시 deepgram_live).
   - FastAPI WS 엔드포인트 `/internal/voice/stream`(`api/voice_stream.py`): RealTime이 프록시한 오디오를 받아 부분/최종 자막을 다운 프레임(`transcript.partial`/`transcript.final`)으로 보내고, 발화 종료(`stop` 또는 UtteranceEnd) 시 메트릭 계산 후 `callback.voice` 발행 → 기존 followup 파이프라인 재사용.
+- **STT 환각 제거** (`voice/stt/sanitize.py`): Whisper/Deepgram 이 발화 끝 무음·잡음에서 학습데이터(방송/유튜브) 정형 문구를 환각으로 덧붙이는 문제(예: "MBC 뉴스 OOO입니다", "시청해주셔서 감사합니다", "구독과 좋아요", 영어 "thanks for watching")를 보수적으로 제거. 배치(`deepgram`/`openai_whisper`)는 `transcribe` 반환 시, 라이브(`deepgram_live`)는 **최종 자막마다** + `result()` 백스톱에서 적용 → 저장 전사·메트릭·실시간 표시 모두 정화. 환각만 남은 segment 는 제거해 무음/발음 메트릭이 실제 무음 반영. "감사합니다"·"뉴스 앱" 등 정상 표현은 보존.
 - 추상화 계층 두기: `voice/stt/base.py` (interface), `voice/stt/whisper_api.py`, `voice/tts/base.py` + `voice/tts/{provider}.py`
 - 분석:
   - WPM = words / minutes

ai/src/ai_server/voice/stt/deepgram.py

@@ -8,6 +8,7 @@
     TranscriptionResult,
     TranscriptionSegment,
 )
+from ai_server.voice.stt.sanitize import sanitize_transcription
 
 log = structlog.get_logger(__name__)
 
@@ -149,11 +150,13 @@ async def transcribe(
                 )
             )
 
-        return TranscriptionResult(
-            text=text,
-            language=self._language,
-            duration_sec=(float(duration) if duration is not None else None),
-            segments=segments,
+        return sanitize_transcription(
+            TranscriptionResult(
+                text=text,
+                language=self._language,
+                duration_sec=(float(duration) if duration is not None else None),
+                segments=segments,
+            )
         )
 
 

ai/src/ai_server/voice/stt/deepgram_live.py

@@ -11,6 +11,10 @@
 
 from ai_server.voice.stt.base import TranscriptionResult, TranscriptionSegment
 from ai_server.voice.stt.live import LiveSttProvider, LiveSttSession, LiveTranscriptEvent
+from ai_server.voice.stt.sanitize import (
+    sanitize_transcription,
+    strip_stt_hallucinations,
+)
 
 log = structlog.get_logger(__name__)
 
@@ -63,6 +67,9 @@ async def _recv_loop(self) -> None:
                     text = str(alt.get("transcript") or "")
                     is_final = bool(msg.get("is_final"))
                     speech_final = bool(msg.get("speech_final"))
+                    # 최종 자막에서만 환각 제거(부분 자막은 변동성이 커 손대지 않음).
+                    if is_final and text:
+                        text = strip_stt_hallucinations(text)
                     if text:
                         await self._queue.put(
                             LiveTranscriptEvent(
@@ -109,11 +116,13 @@ async def events(self) -> AsyncIterator[LiveTranscriptEvent]:
     async def result(self) -> TranscriptionResult:
         text = " ".join(self._finals).strip()
         dur = self._segments[-1].end_sec if self._segments else None
-        return TranscriptionResult(
-            text=text,
-            language=self._language,
-            duration_sec=dur,
-            segments=list(self._segments),
+        return sanitize_transcription(
+            TranscriptionResult(
+                text=text,
+                language=self._language,
+                duration_sec=dur,
+                segments=list(self._segments),
+            )
         )
 
     async def close(self) -> None:

ai/src/ai_server/voice/stt/openai_whisper.py

@@ -8,6 +8,7 @@
     TranscriptionResult,
     TranscriptionSegment,
 )
+from ai_server.voice.stt.sanitize import sanitize_transcription
 
 log = structlog.get_logger(__name__)
 
@@ -119,13 +120,15 @@ async def transcribe(
                     ),
                 )
             )
-        return TranscriptionResult(
-            text=str(data_resp.get("text", "")),
-            language=data_resp.get("language"),
-            duration_sec=(
-                float(data_resp["duration"]) if "duration" in data_resp else None
-            ),
-            segments=segments,
+        return sanitize_transcription(
+            TranscriptionResult(
+                text=str(data_resp.get("text", "")),
+                language=data_resp.get("language"),
+                duration_sec=(
+                    float(data_resp["duration"]) if "duration" in data_resp else None
+                ),
+                segments=segments,
+            )
         )
 
 

ai/src/ai_server/voice/stt/sanitize.py

@@ -0,0 +1,98 @@
+"""STT 환각(hallucination) 제거.
+
+Whisper/Deepgram 등 STT 모델은 발화 끝 무음·잡음 구간에서, 학습 데이터(유튜브/방송)
+에 흔한 정형 문구를 환각으로 덧붙인다. 대표적으로 방송 클로징("MBC 뉴스 OOO입니다"),
+구독 유도("구독과 좋아요 부탁드립니다"), 시청 인사("시청해주셔서 감사합니다"),
+자막 크레딧 등이다. 이런 문구는 IT 면접 답변에 정상적으로 등장하지 않으므로,
+고유 패턴만 보수적으로 제거한다.
+
+보수성 원칙: 면접 답변에 정상적으로 나올 수 있는 표현(단독 "감사합니다",
+"뉴스 앱" 같은 일반 명사구)은 건드리지 않는다. 방송사명·시청/구독 유도 등
+환각 고유 신호가 있을 때만 제거한다.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import replace
+
+from ai_server.voice.stt.base import TranscriptionResult
+
+# 방송사 클로징: "<방송사> 뉴스 <이름>입니다" (예: "MBC 뉴스 김재경입니다").
+_BROADCASTERS = (
+    r"MBC|KBS|SBS|YTN|JTBC|MBN|TBS|OBS|EBS|채널\s*A|TV\s*조선|연합뉴스(?:TV)?|"
+    r"뉴스데스크|뉴스룸"
+)
+
+_HALLUCINATION_PATTERNS = [
+    re.compile(p, re.IGNORECASE)
+    for p in (
+        # 방송 뉴스 클로징
+        rf"(?:{_BROADCASTERS})\s*뉴스\s*[가-힣]{{2,5}}\s*입니다",
+        rf"이상\s*(?:{_BROADCASTERS})?\s*뉴스(?:였|입니다|였습니다)",
+        # 시청 인사
+        r"시청\s*해\s*주(?:셔서|시고)\s*감사합니다",
+        r"시청해주셔서\s*감사합니다",
+        r"오늘도\s*시청해\s*주셔서[^.\n]*",
+        r"영상\s*을?\s*시청해\s*주셔서[^.\n]*",
+        # 구독/좋아요 유도
+        r"구독\s*(?:과|,|및|\s)*\s*좋아요[^.\n]*(?:부탁\s*드립니다|눌러\s*주세요|부탁해요)?",
+        r"좋아요\s*(?:와|,|및|\s)*\s*구독[^.\n]*(?:부탁\s*드립니다|눌러\s*주세요)?",
+        r"구독\s*(?:과|,)?\s*알림\s*설정[^.\n]*",
+        # 다음 영상/편 인사
+        r"다음\s*(?:영상|시간|편|시간에)\s*(?:에서?)?\s*(?:뵙겠습니다|만나요|만나뵙겠습니다|봬요|뵐게요)",
+        # 자막 크레딧
+        r"(?:한글\s*)?자막\s*(?:제공|by|:)[^.\n]*",
+        # 영어 환각
+        r"thank\s+you\s+(?:so\s+much\s+|very\s+much\s+)?for\s+watching",
+        r"thanks\s+for\s+watching",
+        r"please\s+(?:like\s+and\s+|don'?t\s+forget\s+to\s+)?subscribe",
+        r"like\s+and\s+subscribe",
+    )
+]
+
+# 실제 내용(글자/숫자) 보유 여부 — 환각 제거 후 문장부호만 남은 segment 판별.
+_HAS_CONTENT = re.compile(r"[0-9A-Za-z가-힣]")
+
+# 환각 제거 후 정돈: 중복 공백, 문장부호 앞 공백, 연속 문장부호.
+_CLEANUP_SPACE = re.compile(r"[ \t]{2,}")
+_CLEANUP_PUNCT_SPACE = re.compile(r"\s+([.,!?])")
+_CLEANUP_DUP_PUNCT = re.compile(r"([.?!])(?:\s*[.?!])+")
+
+
+def strip_stt_hallucinations(text: str) -> str:
+    """텍스트에서 알려진 STT 환각 문구를 제거하고 잔여 공백/문장부호를 정돈한다."""
+    if not text:
+        return text
+    cleaned = text
+    for pattern in _HALLUCINATION_PATTERNS:
+        cleaned = pattern.sub(" ", cleaned)
+    if cleaned == text:
+        return text
+    cleaned = _CLEANUP_SPACE.sub(" ", cleaned)
+    cleaned = _CLEANUP_PUNCT_SPACE.sub(r"\1", cleaned)
+    cleaned = _CLEANUP_DUP_PUNCT.sub(r"\1", cleaned)
+    return cleaned.strip()
+
+
+def sanitize_transcription(result: TranscriptionResult) -> TranscriptionResult:
+    """TranscriptionResult 의 본문과 segment 에서 환각을 제거한다.
+
+    - 본문 text: 환각 문구 제거.
+    - segment: 환각만으로 이루어진 segment 는 제거(무음 구간 blip), 일부만 환각인
+      segment 는 정화된 텍스트로 교체. segment 제거 시 무음(silence)·발음 메트릭이
+      실제 무음을 반영하도록 자연스럽게 보정된다.
+    """
+    cleaned_text = strip_stt_hallucinations(result.text)
+    cleaned_segments = []
+    for seg in result.segments:
+        seg_text = strip_stt_hallucinations(seg.text)
+        # 환각만으로 이루어져 문장부호/공백만 남은 segment 는 제거.
+        if not _HAS_CONTENT.search(seg_text):
+            continue
+        cleaned_segments.append(
+            seg if seg_text == seg.text else replace(seg, text=seg_text)
+        )
+    if cleaned_text == result.text and len(cleaned_segments) == len(result.segments):
+        return result
+    return replace(result, text=cleaned_text, segments=cleaned_segments)

ai/tests/test_stt_sanitize.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import pytest
+
+from ai_server.voice.stt.base import TranscriptionResult, TranscriptionSegment
+from ai_server.voice.stt.sanitize import (
+    sanitize_transcription,
+    strip_stt_hallucinations,
+)
+
+
+@pytest.mark.parametrize(
+    "raw, expected",
+    [
+        # 실제 세션 82 케이스: 답변 끝 무음에서 방송 클로징 환각.
+        (
+            "그런 식으로 타임아웃을 잡았던 것이 생각납니다. MBC 뉴스 김재경입니다.",
+            "그런 식으로 타임아웃을 잡았던 것이 생각납니다.",
+        ),
+        ("정리하겠습니다. KBS 뉴스 홍길동입니다", "정리하겠습니다."),
+        ("답변 마치겠습니다. 시청해주셔서 감사합니다.", "답변 마치겠습니다."),
+        (
+            "이상입니다. 구독과 좋아요 부탁드립니다.",
+            "이상입니다.",
+        ),
+        ("좋은 경험이었습니다. 다음 영상에서 만나요", "좋은 경험이었습니다."),
+        ("That is my answer. Thanks for watching!", "That is my answer."),
+    ],
+)
+def test_strips_known_hallucinations(raw: str, expected: str) -> None:
+    assert strip_stt_hallucinations(raw) == expected
+
+
+@pytest.mark.parametrize(
+    "text",
+    [
+        "감사합니다.",  # 단독 인사는 정상 답변 — 건드리지 않는다.
+        "뉴스 피드를 보여주는 앱을 만들었습니다.",  # '뉴스' 일반 명사
+        "구독자 수를 늘리는 추천 알고리즘을 구현했습니다.",  # '구독' 일반 명사
+        "다음 영상 처리 파이프라인을 설계했습니다.",  # '다음 영상' 일반 명사구
+        "Explain 으로 실행 계획을 분석했습니다.",
+    ],
+)
+def test_keeps_legitimate_answers(text: str) -> None:
+    assert strip_stt_hallucinations(text) == text
+
+
+def test_empty_and_none_safe() -> None:
+    assert strip_stt_hallucinations("") == ""
+    assert strip_stt_hallucinations("   ") == "   "
+
+
+def test_sanitize_transcription_drops_hallucination_segment() -> None:
+    result = TranscriptionResult(
+        text="타임아웃을 잡았습니다. MBC 뉴스 김재경입니다.",
+        language="ko",
+        duration_sec=30.0,
+        segments=[
+            TranscriptionSegment(start_sec=0.0, end_sec=4.0, text="타임아웃을 잡았습니다."),
+            TranscriptionSegment(
+                start_sec=29.0, end_sec=30.0, text="MBC 뉴스 김재경입니다."
+            ),
+        ],
+    )
+
+    cleaned = sanitize_transcription(result)
+
+    assert cleaned.text == "타임아웃을 잡았습니다."
+    assert len(cleaned.segments) == 1
+    assert cleaned.segments[0].text == "타임아웃을 잡았습니다."
+    # 환각이 없는 결과는 동일 객체를 그대로 반환(불필요한 재할당 방지).
+    clean_in = TranscriptionResult(text="정상 답변입니다.", language="ko", duration_sec=2.0, segments=[])
+    assert sanitize_transcription(clean_in) is clean_in