feat(ai): 체인 밖 외부 호출(TTS·임베딩·라이브 STT) 호출 로깅

ai/CLAUDE.md

@@ -383,6 +383,20 @@ docker run --env-file .env -p 8000:8000 stackup-ai
 - **스토리지 추상화** (`storage/`): `S3Storage`(기본) / `LocalFilesystemStorage`. `STORAGE_BACKEND` 토글.
 - **LLM 호출 로깅 본 구현** (`observability/llm_logging_callback.py`, US-30):
   LangChain `AsyncCallbackHandler` 가 토큰/latency 측정 → Core `/api/internal/ai-logs` POST.
+- **체인 밖 외부 호출 로깅 본 구현** (`observability/ai_call_log.py`): 위 콜백은 LangChain 경로만
+  잡아서 TTS·임베딩·라이브 STT 가 `ai_request_logs` 에 **한 줄도 남지 않았다** — 운영 TTS 요청
+  245건 중 43건이 실패했는데 원인을 사후에 알 수 없었고, "지금 정상"이 아니라 "안 보임"이었다.
+  공용 `record_ai_call`/`measure_ai_call` 로 세 경로를 계측한다:
+  - **TTS** — `voice/tts/logging_provider.py: LoggingTtsProvider` 데코레이터(`request_type=tts.synthesize`).
+    consumer 가 아니라 provider 를 감싼 이유는 TTS 가 `generate.tts` consumer 말고도 followup
+    consumer 의 문장 단위 인라인 합성(Part B)에서 호출되기 때문 — consumer 쪽만 계측하면 라이브
+    경로가 통째로 빠진다. mock 은 외부 호출이 아니라 감싸지 않는다.
+  - **임베딩** — `rag/embedder.py` 가 **시도마다** 한 행(`request_type=embedding.embed`,
+    `error_message` 앞에 `[n/N]`). 429 재시도로 사용자에겐 실패가 안 보였지만 몇 번 만에
+    통과하는지 아무도 몰랐다 — batch_size·백오프를 근거 있게 조정하려면 이 숫자가 필요하다.
+  - **라이브 STT** — `api/voice_stream.py` 가 스트림 1건당 한 행(`request_type=stt.live`).
+    빈 전사는 FAILED 로 남겨 무음과 상류 무응답을 구분할 단서를 만든다.
+  `MessagingRuntime.core_client` 를 노출해 WS 핸들러(`app.state.core_client`)도 같은 클라이언트를 쓴다.
 - **질문 TTS consumer 본 구현** (`messaging/consumers/tts_consumer.py`, `voice/tts/`):
   `generate.tts` 수신 → TTS 합성(`GatewayTtsProvider`/`GeminiTtsProvider`, mock fallback) → S3 PUT(`interview/tts/{sessionId}/{messageId}.*`, 확장자는 content_type 기준) → `callback.tts` 발행.
 - **실시간 스트리밍 음성 답변 본 구현** (RT3, `api/voice_stream.py`, `voice/stt/{live,mock_live,deepgram_live,live_factory}.py`):

ai/src/ai_server/CLAUDE.md

@@ -130,6 +130,11 @@ async def consume(message: AbstractIncomingMessage) -> None:
 ### `observability/`
 - `llm_logging_callback.py` — LangChain `AsyncCallbackHandler`. 토큰/latency 측정 후
   `core/client.py: record_ai_log` 로 Core `POST /api/internal/ai-logs` (fire-and-forget)
+- `ai_call_log.py` — **체인을 거치지 않는 외부 호출**(STT/TTS/임베딩)용 공용 기록기.
+  `record_ai_call(...)`(fire-and-forget) + `measure_ai_call(...)`(지연 측정 + 성공/실패 자동 기록).
+  콜백은 LangChain 경로만 잡으므로 이쪽은 호출부가 직접 남겨야 한다 — 실제로 TTS·임베딩·라이브
+  STT 는 기록이 아예 없어서 운영 TTS 실패 43건의 원인을 사후에 알 수 없었다. 관측이 본 작업을
+  죽이면 안 되므로 태스크로 떼어 보내고 예외는 삼킨다.
 
 ---
 

ai/src/ai_server/api/voice_stream.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import time
 
 import structlog
 from fastapi import APIRouter, Query, WebSocket, WebSocketDisconnect
@@ -10,6 +11,8 @@
 from ai_server.model.envelope import MessageContext
 from ai_server.model.messages.voice import VoiceCallbackPayload
 from ai_server.voice.analysis.metrics import analyze
+from ai_server.core.client import CoreClient
+from ai_server.observability.ai_call_log import record_ai_call
 from ai_server.voice.stt.live import LiveSttProvider
 
 log = structlog.get_logger(__name__)
@@ -25,6 +28,10 @@ def _publisher(ws: WebSocket) -> CallbackPublisher:
     return ws.app.state.callback_publisher
 
 
+def _core_client(ws: WebSocket) -> CoreClient | None:
+    return getattr(ws.app.state, "core_client", None)
+
+
 def _settings(ws: WebSocket) -> Settings:
     return ws.app.state.settings
 
@@ -49,6 +56,7 @@ async def voice_stream(
     provider = _provider(ws)
     publisher = _publisher(ws)
     session = provider.open_session(content_type=content_type, language=None)
+    started = time.perf_counter()
     await session.start()
 
     async def pump_transcripts() -> None:
@@ -93,6 +101,19 @@ async def pump_transcripts() -> None:
             )
         result = await session.result()
         await session.close()
+        # 라이브 STT 도 과금되는 외부 호출인데 기록이 없어 지연·실패가 보이지 않았다.
+        # 전사가 비어 있으면 무음일 수도, 상류가 조용히 죽었을 수도 있다 — 구분이 되게 남긴다.
+        record_ai_call(
+            _core_client(ws),
+            request_type="stt.live",
+            model_name=getattr(provider, "model_name", None),
+            latency_ms=int((time.perf_counter() - started) * 1000),
+            status="SUCCESS" if result.text.strip() else "FAILED",
+            error_message=(
+                None if result.text.strip() else "빈 전사(무음 또는 상류 무응답)"
+            ),
+            session_id=session_id,
+        )
 
         if result.text.strip():
             metrics = analyze(result, filler_pattern=settings.voice_filler_pattern)

ai/src/ai_server/main.py

@@ -23,6 +23,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
             app.state.settings = settings
             app.state.live_stt_provider = build_live_stt_provider(settings)
             app.state.callback_publisher = runtime.publisher
+            app.state.core_client = runtime.core_client
             yield
         finally:
             await runtime.stop()

ai/src/ai_server/messaging/consumers/voice_consumer.py

@@ -11,6 +11,7 @@
 from ai_server.messaging.idempotency import LruIdempotencyStore
 from ai_server.messaging.publisher import CallbackPublisher
 from ai_server.model.envelope import Envelope
+from ai_server.observability.ai_call_log import record_ai_call
 from ai_server.model.messages.voice import (
     AnalyzeVoiceRequest,
     VoiceCallbackPayload,
@@ -242,28 +243,16 @@ def _record_stt_log(
         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())
+        record_ai_call(
+            self._core_client,
+            request_type="stt.transcribe",
+            model_name=_provider_model_name(self._stt),
+            latency_ms=latency_ms,
+            status=status,
+            error_message=error_message,
+            user_id=envelope.context.user_id,
+            session_id=req.session_id,
+        )
 
 
 def _elapsed_ms(started: float | None) -> int | None:

ai/src/ai_server/messaging/runner.py

@@ -38,7 +38,7 @@
     build_question_generation_chain,
 )
 from ai_server.config.settings import Settings
-from ai_server.core.client import HttpCoreClient
+from ai_server.core.client import CoreClient, HttpCoreClient
 from ai_server.messaging.connection import RabbitConnection
 from ai_server.rag.chunker import MarkdownChunker
 from ai_server.rag.embedder import build_embedding_provider
@@ -91,7 +91,7 @@ def __init__(self, settings: Settings) -> None:
         )
 
         storage = build_storage(settings)
-        core_client = HttpCoreClient(
+        core_client = self._core_client = HttpCoreClient(
             base_url=settings.core_internal_base_url,
             api_key=settings.core_internal_api_key,
             timeout_sec=settings.core_internal_timeout_sec,
@@ -111,6 +111,7 @@ def __init__(self, settings: Settings) -> None:
             batch_size=settings.embedding_batch_size,
             max_retries=settings.embedding_max_retries,
             retry_base_delay_sec=settings.embedding_retry_base_delay_sec,
+            core_client=core_client,
         )
         vision_pdf_reader = build_vision_pdf_reader(settings, core_client=core_client)
 
@@ -227,7 +228,7 @@ def __init__(self, settings: Settings) -> None:
             settings, core_client=core_client
         )
         # TTS provider 는 꼬리질문 인라인 세그먼트 합성과 질문 TTS 양쪽에서 재사용한다.
-        tts = build_tts_provider(settings)
+        tts = build_tts_provider(settings, core_client=core_client)
         self._followup_consumer = FollowupConsumer(
             generator=followup_generator,
             publisher=self._publisher,
@@ -316,6 +317,10 @@ def __init__(self, settings: Settings) -> None:
     def publisher(self) -> CallbackPublisher:
         return self._publisher
 
+    @property
+    def core_client(self) -> CoreClient:
+        return self._core_client
+
     async def start(self) -> None:
         await self._connection.open()
         await self._publisher.open()

ai/src/ai_server/observability/ai_call_log.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from contextlib import asynccontextmanager
+from typing import AsyncIterator
+
+import structlog
+
+from ai_server.core.client import CoreClient
+
+log = structlog.get_logger(__name__)
+
+# error_message 컬럼 상한(1000)보다 넉넉히 짧게 — 원문 전체가 아니라 분류가 목적이다.
+_MAX_ERROR_LEN = 500
+
+
+def record_ai_call(
+    core_client: CoreClient | None,
+    *,
+    request_type: str,
+    model_name: str | None,
+    latency_ms: int | None,
+    status: str,
+    error_message: str | None = None,
+    user_id: int | None = None,
+    session_id: int | None = None,
+    input_tokens: int | None = None,
+    output_tokens: int | None = None,
+) -> None:
+    """`ai_request_logs` 에 fire-and-forget 기록.
+
+    LangChain 경로는 `CoreAiLogCallback` 이 콜백으로 잡지만 STT/TTS/임베딩처럼 체인을
+    거치지 않는 외부 호출은 각자 호출해야 한다. 관측은 부가 기능이므로 기록 실패가
+    본 작업을 죽이지 않는다 — 태스크로 떼어 보내고 예외는 삼킨다.
+    """
+    if core_client is None:
+        return
+
+    async def _do() -> None:
+        try:
+            await core_client.record_ai_log(
+                request_type=request_type,
+                model_name=model_name,
+                input_tokens=input_tokens,
+                output_tokens=output_tokens,
+                latency_ms=latency_ms,
+                status=status,
+                user_id=user_id,
+                session_id=session_id,
+                error_message=(
+                    error_message[:_MAX_ERROR_LEN] if error_message else None
+                ),
+            )
+        except Exception as exc:
+            log.warn("ai_log.failed", error=str(exc), request_type=request_type)
+
+    asyncio.create_task(_do())
+
+
+@asynccontextmanager
+async def measure_ai_call(
+    core_client: CoreClient | None,
+    *,
+    request_type: str,
+    model_name: str | None,
+    user_id: int | None = None,
+    session_id: int | None = None,
+) -> AsyncIterator[None]:
+    """블록의 지연을 재서 성공/실패를 기록한다.
+
+    수동으로 perf_counter 를 감싸면 실패 경로에서 기록을 빠뜨리기 쉽다(실제로 TTS·임베딩은
+    아예 기록이 없었고, 그래서 운영 TTS 실패 43건의 원인을 사후에 알 수 없었다).
+    """
+    started = time.perf_counter()
+    try:
+        yield
+    except Exception as exc:
+        record_ai_call(
+            core_client,
+            request_type=request_type,
+            model_name=model_name,
+            latency_ms=_elapsed_ms(started),
+            status="FAILED",
+            error_message=f"{type(exc).__name__}: {exc}",
+            user_id=user_id,
+            session_id=session_id,
+        )
+        raise
+    record_ai_call(
+        core_client,
+        request_type=request_type,
+        model_name=model_name,
+        latency_ms=_elapsed_ms(started),
+        status="SUCCESS",
+        user_id=user_id,
+        session_id=session_id,
+    )
+
+
+def _elapsed_ms(started: float) -> int:
+    return int((time.perf_counter() - started) * 1000)

ai/src/ai_server/rag/embedder.py

@@ -4,10 +4,14 @@
 import hashlib
 import random
 import struct
+import time
 from typing import Protocol
 
 import structlog
 
+from ai_server.core.client import CoreClient
+from ai_server.observability.ai_call_log import record_ai_call
+
 log = structlog.get_logger(__name__)
 
 
@@ -90,6 +94,7 @@ def __init__(
         batch_size: int = 32,
         max_retries: int = 5,
         retry_base_delay_sec: float = 2.0,
+        core_client: CoreClient | None = None,
     ) -> None:
         if not api_key:
             raise ValueError("GEMINI_API_KEY 누락 — provider=gemini 사용 불가")
@@ -103,6 +108,7 @@ def __init__(
         self._batch_size = max(1, batch_size)
         self._max_retries = max(0, max_retries)
         self._retry_base_delay = max(0.0, retry_base_delay_sec)
+        self._core_client = core_client
 
     @property
     def dim(self) -> int:
@@ -136,13 +142,26 @@ async def embed(
     async def _embed_batch_with_retry(self, batch: list[str], config: object) -> object:
         attempt = 0
         while True:
+            started = time.perf_counter()
             try:
-                return await self._client.aio.models.embed_content(
+                resp = await self._client.aio.models.embed_content(
                     model=self._model,
                     contents=batch,
                     config=config,
                 )
             except Exception as exc:
+                # 시도마다 한 행 — STT 와 같은 규약(`[n/N]` 접두)으로 재시도율을 추적한다.
+                record_ai_call(
+                    self._core_client,
+                    request_type="embedding.embed",
+                    model_name=self._model,
+                    latency_ms=int((time.perf_counter() - started) * 1000),
+                    status="FAILED",
+                    error_message=(
+                        f"[{attempt + 1}/{self._max_retries + 1}] "
+                        f"{type(exc).__name__}: {exc}"
+                    ),
+                )
                 rate_limited = _is_rate_limited(exc)
                 if rate_limited and attempt < self._max_retries:
                     delay = min(
@@ -164,6 +183,17 @@ async def _embed_batch_with_retry(self, batch: list[str], config: object) -> obj
                     retriable=True,
                 ) from exc
 
+            record_ai_call(
+                self._core_client,
+                request_type="embedding.embed",
+                model_name=self._model,
+                latency_ms=int((time.perf_counter() - started) * 1000),
+                status="SUCCESS",
+                input_tokens=None,
+                output_tokens=None,
+            )
+            return resp
+
 
 def build_embedding_provider(
     *,
@@ -174,6 +204,7 @@ def build_embedding_provider(
     batch_size: int = 32,
     max_retries: int = 5,
     retry_base_delay_sec: float = 2.0,
+    core_client: CoreClient | None = None,
 ) -> EmbeddingProvider:
     if provider == "mock":
         return MockEmbeddingProvider(dim=dim, model=model)
@@ -185,6 +216,7 @@ def build_embedding_provider(
             batch_size=batch_size,
             max_retries=max_retries,
             retry_base_delay_sec=retry_base_delay_sec,
+            core_client=core_client,
         )
     if provider == "openai":
         raise NotImplementedError("openai embedding provider 미구현 — 후속 PR에서 추가")