diff --git a/ai/CLAUDE.md b/ai/CLAUDE.md index a3f6fb29..d48c6cc1 100644 --- a/ai/CLAUDE.md +++ b/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`): diff --git a/ai/src/ai_server/CLAUDE.md b/ai/src/ai_server/CLAUDE.md index 4cb0f1f0..6a554d58 100644 --- a/ai/src/ai_server/CLAUDE.md +++ b/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건의 원인을 사후에 알 수 없었다. 관측이 본 작업을 + 죽이면 안 되므로 태스크로 떼어 보내고 예외는 삼킨다. --- diff --git a/ai/src/ai_server/api/voice_stream.py b/ai/src/ai_server/api/voice_stream.py index 0e8e4656..d7b7b41f 100644 --- a/ai/src/ai_server/api/voice_stream.py +++ b/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) diff --git a/ai/src/ai_server/main.py b/ai/src/ai_server/main.py index ef464ed9..55a7b0f2 100644 --- a/ai/src/ai_server/main.py +++ b/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() diff --git a/ai/src/ai_server/messaging/consumers/voice_consumer.py b/ai/src/ai_server/messaging/consumers/voice_consumer.py index 7cfe4861..ef462c34 100644 --- a/ai/src/ai_server/messaging/consumers/voice_consumer.py +++ b/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: diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index 3e86a749..d33df619 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/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() diff --git a/ai/src/ai_server/observability/ai_call_log.py b/ai/src/ai_server/observability/ai_call_log.py new file mode 100644 index 00000000..23e516fa --- /dev/null +++ b/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) diff --git a/ai/src/ai_server/rag/embedder.py b/ai/src/ai_server/rag/embedder.py index 9f2fc44e..61b4e696 100644 --- a/ai/src/ai_server/rag/embedder.py +++ b/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에서 추가") diff --git a/ai/src/ai_server/voice/tts/factory.py b/ai/src/ai_server/voice/tts/factory.py index 1de921f1..be3941b5 100644 --- a/ai/src/ai_server/voice/tts/factory.py +++ b/ai/src/ai_server/voice/tts/factory.py @@ -3,19 +3,34 @@ import structlog from ai_server.config.settings import Settings +from ai_server.core.client import CoreClient from ai_server.voice.tts.base import TtsProvider from ai_server.voice.tts.gateway import GatewayTtsProvider from ai_server.voice.tts.gemini import GeminiTtsProvider +from ai_server.voice.tts.logging_provider import LoggingTtsProvider from ai_server.voice.tts.mock import MockTtsProvider log = structlog.get_logger(__name__) -def build_tts_provider(settings: Settings) -> TtsProvider: +def build_tts_provider( + settings: Settings, core_client: CoreClient | None = None +) -> TtsProvider: """TTS 공급자 선택. auto → gateway(충남대 키) > gemini(직접 키), 없으면 mock. gateway 를 우선해 직접 GEMINI_API_KEY 의 429 부하를 던다(게이트웨이가 Gemini TTS 로 라우팅). + core_client 가 있으면 호출 로깅 데코레이터로 감싼다(mock 은 외부 호출이 아니라 제외). """ + return _wrap(_select(settings), core_client) + + +def _wrap(provider: TtsProvider, core_client: CoreClient | None) -> TtsProvider: + if core_client is None or isinstance(provider, MockTtsProvider): + return provider + return LoggingTtsProvider(provider, core_client) + + +def _select(settings: Settings) -> TtsProvider: provider = (settings.tts_provider or "auto").lower() if provider == "auto": diff --git a/ai/src/ai_server/voice/tts/logging_provider.py b/ai/src/ai_server/voice/tts/logging_provider.py new file mode 100644 index 00000000..a1a9fcbb --- /dev/null +++ b/ai/src/ai_server/voice/tts/logging_provider.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from ai_server.core.client import CoreClient +from ai_server.observability.ai_call_log import measure_ai_call +from ai_server.voice.tts.base import TtsProvider, TtsResult + + +class LoggingTtsProvider(TtsProvider): + """TTS 합성 호출을 `ai_request_logs` 에 남기는 데코레이터. + + consumer 가 아니라 provider 를 감싸는 이유: TTS 는 `generate.tts` consumer 말고도 + followup consumer 의 문장 단위 인라인 합성(Part B)에서 호출된다. consumer 쪽에만 + 계측을 넣으면 라이브 경로가 통째로 빠진다. + + 세션/사용자 맥락은 싣지 않는다 — `synthesize(text, *, voice)` 시그니처를 호출부마다 + 바꿔야 해서 비용 대비 이득이 적다. 실패율·지연·모델별 사용량은 이것만으로 보인다. + """ + + def __init__(self, inner: TtsProvider, core_client: CoreClient | None) -> None: + self._inner = inner + self._core_client = core_client + self.model_name = inner.model_name + + async def synthesize(self, text: str, *, voice: str) -> TtsResult: + async with measure_ai_call( + self._core_client, + request_type="tts.synthesize", + model_name=self._inner.model_name, + ): + return await self._inner.synthesize(text, voice=voice) diff --git a/ai/tests/test_ai_call_log.py b/ai/tests/test_ai_call_log.py new file mode 100644 index 00000000..fbd7ae2a --- /dev/null +++ b/ai/tests/test_ai_call_log.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import asyncio +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from ai_server.observability.ai_call_log import measure_ai_call, record_ai_call +from ai_server.voice.tts.base import TtsResult +from ai_server.voice.tts.logging_provider import LoggingTtsProvider + + +def _core() -> MagicMock: + core = MagicMock() + core.record_ai_log = AsyncMock() + return core + + +@pytest.mark.asyncio +async def test_record_is_fire_and_forget_and_swallows_failure(): + """관측이 본 작업을 죽이면 안 된다 — Core 가 죽어도 호출부는 영향을 받지 않는다.""" + core = _core() + core.record_ai_log.side_effect = RuntimeError("core down") + + record_ai_call( + core, + request_type="tts.synthesize", + model_name="m", + latency_ms=10, + status="SUCCESS", + ) + await asyncio.sleep(0) + await asyncio.sleep(0) + + core.record_ai_log.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_record_without_core_client_is_noop(): + record_ai_call(None, request_type="x", model_name=None, latency_ms=1, status="OK") + await asyncio.sleep(0) + + +@pytest.mark.asyncio +async def test_record_truncates_long_error_message(): + core = _core() + record_ai_call( + core, + request_type="x", + model_name=None, + latency_ms=1, + status="FAILED", + error_message="e" * 5000, + ) + await asyncio.sleep(0) + assert len(core.record_ai_log.await_args.kwargs["error_message"]) == 500 + + +@pytest.mark.asyncio +async def test_measure_records_success_with_latency(): + core = _core() + async with measure_ai_call(core, request_type="embedding.embed", model_name="g"): + await asyncio.sleep(0.01) + await asyncio.sleep(0) + + kwargs = core.record_ai_log.await_args.kwargs + assert kwargs["status"] == "SUCCESS" + assert kwargs["request_type"] == "embedding.embed" + assert kwargs["latency_ms"] >= 5 + + +@pytest.mark.asyncio +async def test_measure_records_failure_and_reraises(): + """실패 경로 기록을 빠뜨리면 '왜 실패했는지'가 영원히 남지 않는다.""" + core = _core() + with pytest.raises(ValueError): + async with measure_ai_call(core, request_type="tts.synthesize", model_name="m"): + raise ValueError("boom") + await asyncio.sleep(0) + + kwargs = core.record_ai_log.await_args.kwargs + assert kwargs["status"] == "FAILED" + assert kwargs["error_message"] == "ValueError: boom" + + +class _StubTts: + model_name = "gemini-tts" + + def __init__(self, exc: Exception | None = None) -> None: + self.exc = exc + self.calls = 0 + + async def synthesize(self, text: str, *, voice: str) -> TtsResult: + self.calls += 1 + if self.exc: + raise self.exc + return TtsResult(audio_bytes=b"x", duration_sec=1.0) + + +@pytest.mark.asyncio +async def test_logging_tts_provider_records_success_and_delegates(): + core = _core() + inner = _StubTts() + provider = LoggingTtsProvider(inner, core) + + result = await provider.synthesize("안녕하세요", voice="Kore") + await asyncio.sleep(0) + + assert result.audio_bytes == b"x" + assert inner.calls == 1 + kwargs = core.record_ai_log.await_args.kwargs + assert kwargs["request_type"] == "tts.synthesize" + assert kwargs["model_name"] == "gemini-tts" + assert kwargs["status"] == "SUCCESS" + + +@pytest.mark.asyncio +async def test_logging_tts_provider_records_failure(): + # 운영 TTS 실패 43건의 원인을 사후에 알 수 없었던 이유가 이 기록의 부재였다. + core = _core() + provider = LoggingTtsProvider(_StubTts(exc=RuntimeError("gateway 400")), core) + + with pytest.raises(RuntimeError): + await provider.synthesize("안녕하세요", voice="Kore") + await asyncio.sleep(0) + + kwargs = core.record_ai_log.await_args.kwargs + assert kwargs["status"] == "FAILED" + assert "gateway 400" in kwargs["error_message"] diff --git a/ai/tests/test_embedder.py b/ai/tests/test_embedder.py index fb6f99ed..cae05e04 100644 --- a/ai/tests/test_embedder.py +++ b/ai/tests/test_embedder.py @@ -285,3 +285,49 @@ class _RateLimited(Exception): assert exc_info.value.retriable is True # 최초 1 + 재시도 2 = 3회. assert fake_aio.models.embed_content.await_count == 3 + + +@pytest.mark.asyncio +async def test_gemini_embed_records_ai_log_per_attempt() -> None: + """재시도율이 보여야 batch_size·백오프를 근거 있게 조정할 수 있다. + + 운영에서 임베딩은 429 재시도로 사용자에게 실패를 보이지 않았지만, 호출 로그가 없어 + '몇 번 만에 통과했는지'를 아무도 몰랐다. STT 와 같은 `[n/N]` 규약으로 남긴다. + """ + import asyncio + from types import SimpleNamespace + from unittest.mock import AsyncMock, MagicMock, patch + + from ai_server.rag.embedder import GeminiEmbeddingProvider + + class _RateLimited(Exception): + code = 429 + status = "RESOURCE_EXHAUSTED" + + core = MagicMock() + core.record_ai_log = AsyncMock() + ok = SimpleNamespace(embeddings=[SimpleNamespace(values=[0.1, 0.2, 0.3, 0.4])]) + + fake_aio = MagicMock() + fake_aio.models.embed_content = AsyncMock(side_effect=[_RateLimited("429"), ok]) + fake_client = MagicMock() + fake_client.aio = fake_aio + + with patch("google.genai.Client", return_value=fake_client): + emb = GeminiEmbeddingProvider( + api_key="fake", + model="gemini-embedding-001", + dim=4, + max_retries=1, + retry_base_delay_sec=0.0, + core_client=core, + ) + + await emb.embed(["안녕하세요"]) + await asyncio.sleep(0) + + assert fake_aio.models.embed_content.await_count == 2 + calls = core.record_ai_log.await_args_list + assert [c.kwargs["status"] for c in calls] == ["FAILED", "SUCCESS"] + assert "[1/2]" in calls[0].kwargs["error_message"] + assert all(c.kwargs["request_type"] == "embedding.embed" for c in calls) diff --git a/docs/observability.md b/docs/observability.md index 0319d963..92c714e6 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -118,6 +118,25 @@ error_message - 실패율 알림 (5분 윈도우 5% 초과 시) - 사용자별 토큰 사용량 (남용 감지) +### 3.1 기록 경로 — 두 가지 + +| 경로 | 구현 | 대상 | +|------|------|------| +| LangChain 콜백 | `observability/llm_logging_callback.py` | 체인을 타는 LLM 호출(질문·꼬리질문·피드백·문서 분석) | +| 직접 기록 | `observability/ai_call_log.py` | 체인 밖 외부 호출 — STT(배치·라이브)·TTS·임베딩 | + +콜백은 체인 경로만 잡는다. 그래서 한동안 TTS·임베딩·라이브 STT 는 `ai_request_logs` 에 +**한 줄도 남지 않았고**, 운영 TTS 요청 245건 중 43건이 실패했는데 원인을 사후에 알 수 없었다. +외부 API 를 새로 붙일 때는 이 표에 어느 경로로 기록되는지 함께 적는다. + +**현재 `request_type` 목록**: `generate.questions` · `generate.followup.stream` · +`generate.feedback.{panel,synthesis,coaching,self_intro}` · `analyze.document` · +`stt.transcribe` · `stt.live` · `tts.synthesize` · `embedding.embed` + +재시도가 있는 경로(STT·임베딩)는 **시도마다 한 행**을 남기고 `error_message` 앞에 `[n/N]` 을 +붙인다. 실패율을 셀 때 분모가 '요청'이 아니라 '시도'가 되므로, 사용자에게 보인 실패만 세려면 +`[N/N]` 으로 끝난 행만 센다. 이 규약 덕분에 재시도율 자체를 관측할 수 있다. + ```sql SELECT model_name, AVG(latency_ms), COUNT(*) AS reqs FROM ai_request_logs