Merge pull request #53 from Team-StackUp/feature/ai-voice-stt-log-flow-v2

pswaao88 · web-flow · commit b019d19bb6f9 · 2026-06-02T02:35:07.000+09:00
feat: AI 서버 STT 로그 연동
diff --git 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,20 +41,28 @@ def __init__(
         idempotency: LruIdempotencyStore,
         callback_routing_key: str,
         filler_pattern: str,
+        core_client: CoreClient | None = None,
     ) -> None:
         self._stt = stt
         self._storage = storage
         self._publisher = publisher
         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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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,
         )
 
diff --git a/ai/tests/test_voice_consumer.py b/ai/tests/test_voice_consumer.py

Original file line number	Diff line number	Diff line change
`@@ -194,6 +194,7 @@ def __init__(self, settings: Settings) -> None:`
`194`	`194`	`idempotency=self._idempotency,`
`195`	`195`	`callback_routing_key=settings.ai_callback_routing_voice,`
`196`	`196`	`filler_pattern=settings.voice_filler_pattern,`
	`197`	`+ core_client=core_client,`
`197`	`198`	`)`
`198`	`199`
`199`	`200`	`self._consumers: list[tuple[AbstractRobustQueue, str]] = []`