From 8839a9864916b2574cdc0c5da894e4b466d40892 Mon Sep 17 00:00:00 2001 From: pswaao88 Date: Tue, 2 Jun 2026 02:23:31 +0900 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20STT=20=EC=9A=94=EC=B2=AD=20AI=20?= =?UTF-8?q?=EB=A1=9C=EA=B7=B8=20=EC=97=B0=EB=8F=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../messaging/consumers/voice_consumer.py | 92 ++++++++++++++++++- ai/src/ai_server/messaging/runner.py | 1 + ai/src/ai_server/voice/stt/base.py | 3 + ai/src/ai_server/voice/stt/deepgram.py | 40 ++++++-- ai/src/ai_server/voice/stt/mock.py | 4 + ai/src/ai_server/voice/stt/openai_whisper.py | 44 +++++++-- 6 files changed, 165 insertions(+), 19 deletions(-) diff --git a/ai/src/ai_server/messaging/consumers/voice_consumer.py b/ai/src/ai_server/messaging/consumers/voice_consumer.py index b4adfb4f..d90408ff 100644 --- a/ai/src/ai_server/messaging/consumers/voice_consumer.py +++ b/ai/src/ai_server/messaging/consumers/voice_consumer.py @@ -1,8 +1,12 @@ from __future__ import annotations +import asyncio +import time + import structlog from aio_pika.abc import AbstractIncomingMessage +from ai_server.core.client import CoreClient from ai_server.messaging.idempotency import LruIdempotencyStore from ai_server.messaging.publisher import CallbackPublisher from ai_server.model.envelope import Envelope @@ -37,6 +41,7 @@ def __init__( idempotency: LruIdempotencyStore, callback_routing_key: str, filler_pattern: str, + core_client: CoreClient | None = None, ) -> None: self._stt = stt self._storage = storage @@ -44,13 +49,20 @@ def __init__( self._idempotency = idempotency self._callback_routing_key = callback_routing_key self._filler_pattern = filler_pattern + self._core_client = core_client async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): try: - envelope = Envelope[AnalyzeVoiceRequest].model_validate_json(message.body) + envelope = Envelope[AnalyzeVoiceRequest].model_validate_json( + message.body + ) except Exception as exc: - log.error("voice.parse.failed", error=str(exc), delivery_tag=message.delivery_tag) + log.error( + "voice.parse.failed", + error=str(exc), + delivery_tag=message.delivery_tag, + ) raise if self._idempotency.is_seen_then_mark(envelope.message_id): @@ -74,13 +86,29 @@ async def handle(self, message: AbstractIncomingMessage) -> None: await self._publish_failed(envelope, req, code="AUDIO_FETCH_FAILED") return + started = None try: + started = time.perf_counter() result = await self._stt.transcribe( audio_bytes=audio_bytes, content_type=req.content_type, hint=req.previous_question_text, ) + self._record_stt_log( + req, + envelope, + latency_ms=_elapsed_ms(started), + status="SUCCESS", + error_message=None, + ) except SttError as exc: + self._record_stt_log( + req, + envelope, + latency_ms=_elapsed_ms(started), + status="FAILED", + error_message=exc.message, + ) log.error( "voice.stt.failed", error=str(exc), @@ -90,7 +118,16 @@ async def handle(self, message: AbstractIncomingMessage) -> None: await self._publish_failed(envelope, req, code=exc.code) return except Exception as exc: - log.error("voice.stt.unexpected", error=str(exc), session_id=req.session_id) + self._record_stt_log( + req, + envelope, + latency_ms=_elapsed_ms(started), + status="FAILED", + error_message=str(exc), + ) + log.error( + "voice.stt.unexpected", error=str(exc), session_id=req.session_id + ) await self._publish_failed(envelope, req, code="TRANSCRIPTION_FAILED") return @@ -123,7 +160,9 @@ async def handle(self, message: AbstractIncomingMessage) -> None: trace_id=envelope.trace_id, ) - async def _publish_failed(self, envelope, req: AnalyzeVoiceRequest, *, code: str) -> None: + async def _publish_failed( + self, envelope, req: AnalyzeVoiceRequest, *, code: str + ) -> None: payload = VoiceCallbackPayload( session_id=req.session_id, interview_message_id=req.message_id, @@ -142,3 +181,48 @@ async def _publish_failed(self, envelope, req: AnalyzeVoiceRequest, *, code: str correlation_id=envelope.message_id, context=envelope.context, ) + + def _record_stt_log( + self, + req: AnalyzeVoiceRequest, + envelope, + *, + latency_ms: int | None, + status: str, + error_message: str | None, + ) -> None: + if self._core_client is None: + return + + async def _do() -> None: + try: + await self._core_client.record_ai_log( + request_type="stt.transcribe", + model_name=_provider_model_name(self._stt), + input_tokens=None, + output_tokens=None, + latency_ms=latency_ms, + status=status, + user_id=envelope.context.user_id, + session_id=req.session_id, + error_message=(error_message[:1000] if error_message else None), + ) + except Exception as exc: + log.warn( + "voice.ai_log.failed", error=str(exc), session_id=req.session_id + ) + + asyncio.create_task(_do()) + + +def _elapsed_ms(started: float | None) -> int | None: + if started is None: + return None + return int((time.perf_counter() - started) * 1000) + + +def _provider_model_name(provider) -> str | None: + value = getattr(provider, "model_name", None) + if isinstance(value, str): + return value + return None diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index 4edcc3ca..eb953790 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/ai/src/ai_server/messaging/runner.py @@ -194,6 +194,7 @@ def __init__(self, settings: Settings) -> None: idempotency=self._idempotency, callback_routing_key=settings.ai_callback_routing_voice, filler_pattern=settings.voice_filler_pattern, + core_client=core_client, ) self._consumers: list[tuple[AbstractRobustQueue, str]] = [] diff --git a/ai/src/ai_server/voice/stt/base.py b/ai/src/ai_server/voice/stt/base.py index c380d52a..2c7a8281 100644 --- a/ai/src/ai_server/voice/stt/base.py +++ b/ai/src/ai_server/voice/stt/base.py @@ -21,6 +21,9 @@ class TranscriptionResult: class SttProvider(Protocol): + @property + def model_name(self) -> str | None: ... + async def transcribe( self, *, diff --git a/ai/src/ai_server/voice/stt/deepgram.py b/ai/src/ai_server/voice/stt/deepgram.py index 44896550..cf649363 100644 --- a/ai/src/ai_server/voice/stt/deepgram.py +++ b/ai/src/ai_server/voice/stt/deepgram.py @@ -41,6 +41,10 @@ def __init__( self._timeout_sec = timeout_sec self._client = client + @property + def model_name(self) -> str | None: + return self._model + async def transcribe( self, *, @@ -69,17 +73,33 @@ async def transcribe( try: if self._client is not None: - resp = await self._client.post(url, params=params, headers=headers, content=audio_bytes) + resp = await self._client.post( + url, params=params, headers=headers, content=audio_bytes + ) else: async with httpx.AsyncClient(timeout=self._timeout_sec) as client: - resp = await client.post(url, params=params, headers=headers, content=audio_bytes) + resp = await client.post( + url, params=params, headers=headers, content=audio_bytes + ) except httpx.HTTPError as exc: - raise SttError(code="STT_UNAVAILABLE", message=f"Deepgram 호출 실패: {exc}", retriable=True) from exc + raise SttError( + code="STT_UNAVAILABLE", + message=f"Deepgram 호출 실패: {exc}", + retriable=True, + ) from exc if resp.status_code in (401, 403): - raise SttError(code="STT_AUTH_FAILED", message=f"Deepgram 인증 실패: {resp.status_code}", retriable=False) + raise SttError( + code="STT_AUTH_FAILED", + message=f"Deepgram 인증 실패: {resp.status_code}", + retriable=False, + ) if resp.status_code >= 500: - raise SttError(code="STT_UNAVAILABLE", message=f"Deepgram 5xx: {resp.status_code}", retriable=True) + raise SttError( + code="STT_UNAVAILABLE", + message=f"Deepgram 5xx: {resp.status_code}", + retriable=True, + ) if resp.status_code >= 400: raise SttError( code="STT_BAD_REQUEST", @@ -90,13 +110,19 @@ async def transcribe( try: data = resp.json() except ValueError as exc: - raise SttError(code="STT_BAD_RESPONSE", message=f"JSON 파싱 실패: {exc}", retriable=True) from exc + raise SttError( + code="STT_BAD_RESPONSE", + message=f"JSON 파싱 실패: {exc}", + retriable=True, + ) from exc metadata = data.get("metadata") or {} results = data.get("results") or {} channels = results.get("channels") or [] if not channels: - return TranscriptionResult(text="", language=self._language, duration_sec=None, segments=[]) + return TranscriptionResult( + text="", language=self._language, duration_sec=None, segments=[] + ) alt = (channels[0].get("alternatives") or [{}])[0] text = str(alt.get("transcript") or "") diff --git a/ai/src/ai_server/voice/stt/mock.py b/ai/src/ai_server/voice/stt/mock.py index b6c13655..e2d3c445 100644 --- a/ai/src/ai_server/voice/stt/mock.py +++ b/ai/src/ai_server/voice/stt/mock.py @@ -13,6 +13,10 @@ class MockSttProvider: def __init__(self, *, default_text: str = "(mock transcript)") -> None: self._default_text = default_text + @property + def model_name(self) -> str | None: + return "mock-stt" + async def transcribe( self, *, diff --git a/ai/src/ai_server/voice/stt/openai_whisper.py b/ai/src/ai_server/voice/stt/openai_whisper.py index 7f95d1e2..54b81c62 100644 --- a/ai/src/ai_server/voice/stt/openai_whisper.py +++ b/ai/src/ai_server/voice/stt/openai_whisper.py @@ -38,6 +38,10 @@ def __init__( self._timeout_sec = timeout_sec self._client = client + @property + def model_name(self) -> str | None: + return self._model + async def transcribe( self, *, @@ -60,17 +64,33 @@ async def transcribe( try: if self._client is not None: - resp = await self._client.post(url, headers=headers, files=files, data=data) + resp = await self._client.post( + url, headers=headers, files=files, data=data + ) else: async with httpx.AsyncClient(timeout=self._timeout_sec) as client: - resp = await client.post(url, headers=headers, files=files, data=data) + resp = await client.post( + url, headers=headers, files=files, data=data + ) except httpx.HTTPError as exc: - raise SttError(code="STT_UNAVAILABLE", message=f"OpenAI 호출 실패: {exc}", retriable=True) from exc + raise SttError( + code="STT_UNAVAILABLE", + message=f"OpenAI 호출 실패: {exc}", + retriable=True, + ) from exc if resp.status_code in (401, 403): - raise SttError(code="STT_AUTH_FAILED", message=f"OpenAI 인증 실패: {resp.status_code}", retriable=False) + raise SttError( + code="STT_AUTH_FAILED", + message=f"OpenAI 인증 실패: {resp.status_code}", + retriable=False, + ) if resp.status_code >= 500: - raise SttError(code="STT_UNAVAILABLE", message=f"OpenAI 5xx: {resp.status_code}", retriable=True) + raise SttError( + code="STT_UNAVAILABLE", + message=f"OpenAI 5xx: {resp.status_code}", + retriable=True, + ) if resp.status_code >= 400: raise SttError( code="STT_BAD_REQUEST", @@ -81,7 +101,11 @@ async def transcribe( try: data_resp = resp.json() except ValueError as exc: - raise SttError(code="STT_BAD_RESPONSE", message=f"JSON 파싱 실패: {exc}", retriable=True) from exc + raise SttError( + code="STT_BAD_RESPONSE", + message=f"JSON 파싱 실패: {exc}", + retriable=True, + ) from exc segments = [] for seg in data_resp.get("segments") or []: @@ -90,13 +114,17 @@ async def transcribe( start_sec=float(seg.get("start", 0.0)), end_sec=float(seg.get("end", 0.0)), text=str(seg.get("text", "")), - avg_logprob=(float(seg["avg_logprob"]) if "avg_logprob" in seg else None), + avg_logprob=( + float(seg["avg_logprob"]) if "avg_logprob" in seg else None + ), ) ) 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), + duration_sec=( + float(data_resp["duration"]) if "duration" in data_resp else None + ), segments=segments, ) From 91da13c6fe2aee424d2f48e0075e43d4749db130 Mon Sep 17 00:00:00 2001 From: pswaao88 Date: Tue, 2 Jun 2026 02:25:45 +0900 Subject: [PATCH 2/2] =?UTF-8?q?test:=20STT=20=EB=A1=9C=EA=B7=B8=20?= =?UTF-8?q?=EC=B2=98=EB=A6=AC=20=ED=85=8C=EC=8A=A4=ED=8A=B8=20=EB=B3=B4?= =?UTF-8?q?=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ai/tests/test_voice_consumer.py | 69 +++++++++++++++++++++++++++++++-- 1 file changed, 66 insertions(+), 3 deletions(-) diff --git a/ai/tests/test_voice_consumer.py b/ai/tests/test_voice_consumer.py index c9e8dd15..0df1a2dc 100644 --- a/ai/tests/test_voice_consumer.py +++ b/ai/tests/test_voice_consumer.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import asyncio from unittest.mock import AsyncMock, MagicMock import pytest @@ -63,9 +64,12 @@ def _stt_ok(): language="ko", duration_sec=30.0, segments=[ - TranscriptionSegment(start_sec=0.0, end_sec=30.0, - text="ACID는 원자성 일관성 격리성 영속성 입니다", - avg_logprob=-0.3) + TranscriptionSegment( + start_sec=0.0, + end_sec=30.0, + text="ACID는 원자성 일관성 격리성 영속성 입니다", + avg_logprob=-0.3, + ) ], ) ) @@ -126,6 +130,65 @@ async def test_consumer_publishes_error_code_on_stt_failure(): assert payload.transcript is None +@pytest.mark.asyncio +async def test_consumer_records_stt_ai_log_success_fire_and_forget(): + stt = _stt_ok() + stt.model_name = "whisper-test" + publisher = MagicMock() + publisher.publish = AsyncMock() + core_client = MagicMock() + core_client.record_ai_log = AsyncMock(side_effect=RuntimeError("core down")) + + consumer = VoiceConsumer( + stt=stt, + storage=_storage_ok(), + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.voice", + filler_pattern=r"(?:??|??|洹?|??)", + core_client=core_client, + ) + await consumer.handle(_StubMessage(_envelope())) + await asyncio.sleep(0) + + publisher.publish.assert_awaited_once() + core_client.record_ai_log.assert_awaited_once() + kwargs = core_client.record_ai_log.await_args.kwargs + assert kwargs["request_type"] == "stt.transcribe" + assert kwargs["model_name"] == "whisper-test" + assert kwargs["status"] == "SUCCESS" + assert kwargs["session_id"] == 99 + + +@pytest.mark.asyncio +async def test_consumer_records_stt_ai_log_failure_without_blocking_callback(): + stt = MagicMock() + stt.model_name = "whisper-test" + stt.transcribe = AsyncMock( + side_effect=SttError(code="STT_AUTH_FAILED", message="bad key", retriable=False) + ) + publisher = MagicMock() + publisher.publish = AsyncMock() + core_client = MagicMock() + core_client.record_ai_log = AsyncMock() + + consumer = VoiceConsumer( + stt=stt, + storage=_storage_ok(), + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.voice", + filler_pattern=r"(?:??|??|洹?|??)", + core_client=core_client, + ) + await consumer.handle(_StubMessage(_envelope())) + await asyncio.sleep(0) + + payload: VoiceCallbackPayload = publisher.publish.await_args.kwargs["payload"] + assert payload.error_code == "STT_AUTH_FAILED" + assert core_client.record_ai_log.await_args.kwargs["status"] == "FAILED" + + @pytest.mark.asyncio async def test_consumer_publishes_error_code_on_storage_failure(): storage = MagicMock()