From 21945da3ab12742b33310fdfa9ab801e5b1edf40 Mon Sep 17 00:00:00 2001 From: jmj Date: Mon, 28 Sep 2026 19:56:26 +0900 Subject: [PATCH] =?UTF-8?q?fix(ai):=20ai=5Frequest=5Flogs=20=EC=97=90=20?= =?UTF-8?q?=EC=84=B8=EC=85=98=20=EA=B7=80=EC=86=8D=20=E2=80=94=20=EB=B9=84?= =?UTF-8?q?=EC=9A=A9=EC=9D=84=20=EB=A9=B4=EC=A0=91=20=EB=8B=A8=EC=9C=84?= =?UTF-8?q?=EB=A1=9C=20=EB=B3=BC=20=EC=88=98=20=EC=9E=88=EA=B2=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 테스트 면접(세션 102)을 끝까지 돌려 확인하다 나왔다. AI 호출 39건이 기록됐는데 session_id 가 **전부 NULL** 이었다. 회귀가 아니라 오래된 구멍이다. request_type 별 채워진 비율(전체 기간): stt.transcribe 7 / 7 ← 유일 generate.feedback.panel 0 / 45 generate.feedback.synthesis 0 / 15 generate.feedback.coaching 0 / 13 generate.feedback.self_intro 0 / 13 embedding.embed 0 / 12 tts.synthesize 0 / 17 generate.questions 0 / 8 generate.followup.stream 0 / 9 analyze.document 0 / 5 ← 세션 없음(정상) record_ai_call 과 CoreAiLogCallback 은 둘 다 session_id 를 **받을 수 있게** 만들어져 있었다. 넘기는 호출부가 voice consumer 하나뿐이었을 뿐이다. 그래서 "이 면접이 얼마나 들었나", "어느 세션이 게이트웨이 크레딧을 태웠나" 를 답할 수 없었다 — 2026-09-17 에 크레딧이 소진돼 운영 AI 가 402 로 죽었을 때 정확히 이 질문을 못 했다. ## 방법 체인 빌더마다 인자로 넘기면 LangChain 콜백 생성 지점 6곳과 체인 팩토리 시그니처가 전부 오염된다. 관측값은 횡단 관심사이므로 traceId 와 같은 방식으로 컨텍스트에 싣는다. trace_context(trace_id, session_id=..., user_id=...) 컨슈머 진입점이 payload 에서 꺼내 묶고(getattr 로 — 문서 분석 payload 에는 session_id 가 없다), record_ai_call 과 CoreAiLogCallback 이 명시 인자가 없을 때 거기서 읽는다. 명시 인자가 있으면 그쪽이 이긴다(voice consumer 가 그렇게 쓴다). session_id 는 로그 줄에도 실린다 — 로그만 봐도 어느 면접의 호출인지 보인다. ## 테스트 여기서도 "넘겼다"가 아니라 **"기록부에 도착한다"** 를 본다. 가짜 CoreClient 로 실제 record_ai_log 호출 인자를 확인한다: record_ai_call 이 컨텍스트에서 session_id/user_id 를 집는다 명시 인자가 컨텍스트를 이긴다 세션 없는 흐름은 NULL 로 기록되고 죽지 않는다 LangChain 콜백도 컨텍스트에서 집는다 (체인 빌더는 세션을 모른다) session_id 가 로그 줄에도 실린다 AI 전체 427건 통과. --- .../messaging/consumers/failure_signal.py | 6 +- .../messaging/consumers/tts_consumer.py | 6 +- .../messaging/consumers/voice_consumer.py | 6 +- ai/src/ai_server/observability/ai_call_log.py | 6 + .../observability/llm_logging_callback.py | 7 + ai/src/ai_server/observability/trace.py | 46 ++++++- ai/tests/test_trace_context.py | 128 ++++++++++++++++++ docs/observability.md | 20 +++ 8 files changed, 220 insertions(+), 5 deletions(-) diff --git a/ai/src/ai_server/messaging/consumers/failure_signal.py b/ai/src/ai_server/messaging/consumers/failure_signal.py index e448081d..c4207146 100644 --- a/ai/src/ai_server/messaging/consumers/failure_signal.py +++ b/ai/src/ai_server/messaging/consumers/failure_signal.py @@ -119,7 +119,11 @@ async def consume_with_failure_signal( ) raise - with trace_context(envelope.trace_id): + with trace_context( + envelope.trace_id, + session_id=getattr(envelope.payload, "session_id", None), + user_id=getattr(envelope.payload, "user_id", None), + ): if idempotency.is_seen_then_mark(envelope.message_id): log.info( f"{domain}.idempotent.skip", diff --git a/ai/src/ai_server/messaging/consumers/tts_consumer.py b/ai/src/ai_server/messaging/consumers/tts_consumer.py index 16aae488..2ea035d3 100644 --- a/ai/src/ai_server/messaging/consumers/tts_consumer.py +++ b/ai/src/ai_server/messaging/consumers/tts_consumer.py @@ -82,7 +82,11 @@ async def handle(self, message: AbstractIncomingMessage) -> None: ) raise - with trace_context(envelope.trace_id): + with trace_context( + envelope.trace_id, + session_id=getattr(envelope.payload, "session_id", None), + user_id=getattr(envelope.payload, "user_id", None), + ): if self._idempotency.is_seen_then_mark(envelope.message_id): log.info("tts.idempotent.skip", message_id=envelope.message_id) return diff --git a/ai/src/ai_server/messaging/consumers/voice_consumer.py b/ai/src/ai_server/messaging/consumers/voice_consumer.py index 15225537..95e3dfb9 100644 --- a/ai/src/ai_server/messaging/consumers/voice_consumer.py +++ b/ai/src/ai_server/messaging/consumers/voice_consumer.py @@ -72,7 +72,11 @@ async def handle(self, message: AbstractIncomingMessage) -> None: ) raise - with trace_context(envelope.trace_id): + with trace_context( + envelope.trace_id, + session_id=getattr(envelope.payload, "session_id", None), + user_id=getattr(envelope.payload, "user_id", None), + ): if self._idempotency.is_seen_then_mark(envelope.message_id): log.info("voice.idempotent.skip", message_id=envelope.message_id) return diff --git a/ai/src/ai_server/observability/ai_call_log.py b/ai/src/ai_server/observability/ai_call_log.py index 23e516fa..8e665d2e 100644 --- a/ai/src/ai_server/observability/ai_call_log.py +++ b/ai/src/ai_server/observability/ai_call_log.py @@ -8,6 +8,7 @@ import structlog from ai_server.core.client import CoreClient +from ai_server.observability.trace import current_session_id, current_user_id log = structlog.get_logger(__name__) @@ -37,6 +38,11 @@ def record_ai_call( if core_client is None: return + # 호출부가 안 넘기면 컨텍스트에서 가져온다 — 체인 빌더 6곳의 시그니처를 오염시키지 + # 않으면서 "어느 면접의 호출인가" 를 채운다. + session_id = session_id if session_id is not None else current_session_id() + user_id = user_id if user_id is not None else current_user_id() + async def _do() -> None: try: await core_client.record_ai_log( diff --git a/ai/src/ai_server/observability/llm_logging_callback.py b/ai/src/ai_server/observability/llm_logging_callback.py index 4ae28975..edf034b5 100644 --- a/ai/src/ai_server/observability/llm_logging_callback.py +++ b/ai/src/ai_server/observability/llm_logging_callback.py @@ -10,6 +10,7 @@ from langchain_core.outputs import LLMResult from ai_server.core.client import CoreClient +from ai_server.observability.trace import current_session_id, current_user_id log = structlog.get_logger(__name__) @@ -71,6 +72,12 @@ def _latency_ms(self, run_id: UUID) -> int | None: return int((time.perf_counter() - started) * 1000) async def _fire(self, **kwargs: Any) -> None: + # 체인 빌더는 세션을 모른다(팩토리가 요청 밖에서 불릴 수 있다). 기록 시점에 + # 컨텍스트에서 읽어 채운다 — 이게 없으면 generate.* / feedback.* 로그의 + # session_id 가 전부 NULL 로 남는다(2026-08-06 이래 실제로 그랬다). + kwargs.setdefault("session_id", current_session_id()) + kwargs.setdefault("user_id", current_user_id()) + # fire-and-forget: 본 핸들러 실패가 LLM 응답 흐름을 막지 않도록. async def _do() -> None: try: diff --git a/ai/src/ai_server/observability/trace.py b/ai/src/ai_server/observability/trace.py index c4d61d13..ed3216ff 100644 --- a/ai/src/ai_server/observability/trace.py +++ b/ai/src/ai_server/observability/trace.py @@ -18,22 +18,64 @@ from __future__ import annotations import contextlib +import contextvars import uuid from collections.abc import Iterator import structlog +# 로그 바인딩과 별개로, `ai_request_logs` 기록부가 읽어 가는 값. +# +# `record_ai_call` 과 `CoreAiLogCallback` 은 둘 다 session_id 를 **받을 수** 있게 만들어 +# 놨는데 실제로 넘기는 호출부가 stt.transcribe 하나뿐이었다. 그래서 2026-08-06 이래 +# ai_request_logs 의 session_id 는 그 한 종류 말고 **전부 NULL** 이다 — "이 면접이 얼마나 +# 들었나", "어느 세션이 게이트웨이 크레딧을 태웠나" 를 답할 수 없다(2026-09-17 에 실제로 +# 크레딧이 소진돼 운영이 402 로 죽었을 때 정확히 이 질문을 못 했다). +# +# 체인 빌더마다 인자로 넘기면 LangChain 콜백 생성 지점 6곳 + 체인 팩토리 시그니처가 +# 전부 오염된다. 관측값은 횡단 관심사이므로 traceId 와 같은 방식으로 컨텍스트에 싣는다. +_session_id: contextvars.ContextVar[int | None] = contextvars.ContextVar( + "ai_log_session_id", default=None +) +_user_id: contextvars.ContextVar[int | None] = contextvars.ContextVar( + "ai_log_user_id", default=None +) + + +def current_session_id() -> int | None: + """기록부가 명시적 인자 없이 읽어 가는 세션 id.""" + return _session_id.get() + + +def current_user_id() -> int | None: + return _user_id.get() + @contextlib.contextmanager -def trace_context(trace_id: str | None) -> Iterator[str]: +def trace_context( + trace_id: str | None, + *, + session_id: int | None = None, + user_id: int | None = None, +) -> Iterator[str]: """`trace_id` 를 이 구간의 모든 로그에 바인딩한다. 실제 쓰인 값을 돌려준다. + `session_id`/`user_id` 를 주면 같은 구간의 `ai_request_logs` 기록에도 자동으로 붙는다 + (로그 줄에도 실린다 — 어느 면접의 호출인지 로그만 봐도 알 수 있게). + 비어 있으면 새로 만든다 — 없는 것보다 낫고, 적어도 그 처리 한 건은 묶인다. 끝나면 반드시 되돌린다: 워커 태스크가 재사용되면 남은 값이 다음 메시지 로그에 묻는다. """ effective = trace_id if trace_id else str(uuid.uuid4()) - tokens = structlog.contextvars.bind_contextvars(trace_id=effective) + bound: dict[str, object] = {"trace_id": effective} + if session_id is not None: + bound["session_id"] = session_id + tokens = structlog.contextvars.bind_contextvars(**bound) + sid_token = _session_id.set(session_id) + uid_token = _user_id.set(user_id) try: yield effective finally: + _user_id.reset(uid_token) + _session_id.reset(sid_token) structlog.contextvars.reset_contextvars(**tokens) diff --git a/ai/tests/test_trace_context.py b/ai/tests/test_trace_context.py index 6d9a273e..8a863c9e 100644 --- a/ai/tests/test_trace_context.py +++ b/ai/tests/test_trace_context.py @@ -86,3 +86,131 @@ async def main() -> None: seen = {e["which"]: e["trace_id"] for e in cap.entries} assert seen == {"A": "A", "B": "B", "C": "C"} + + +# ── ai_request_logs 세션 귀속 ──────────────────────────────────────────────── +# +# record_ai_call 과 CoreAiLogCallback 은 둘 다 session_id 를 받을 수 있게 만들어 놨는데 +# 실제로 넘기는 호출부가 stt.transcribe 하나뿐이어서, 2026-08-06 이래 ai_request_logs 의 +# session_id 는 그 한 종류 말고 전부 NULL 이었다. "이 면접이 얼마나 들었나", "어느 세션이 +# 게이트웨이 크레딧을 태웠나" 를 답할 수 없다. + + +class _RecordingCore: + def __init__(self) -> None: + self.calls: list[dict] = [] + + async def record_ai_log(self, **kwargs) -> None: + self.calls.append(kwargs) + + +def test_record_ai_call_picks_up_session_from_context(): + from ai_server.observability.ai_call_log import record_ai_call + + core = _RecordingCore() + + async def main() -> None: + with trace_context("t", session_id=102, user_id=7): + record_ai_call( + core, + request_type="tts.synthesize", + model_name="m", + latency_ms=10, + status="SUCCESS", + ) + # fire-and-forget 이라 태스크가 돌 틈을 준다 + await asyncio.sleep(0) + await asyncio.sleep(0) + + asyncio.run(main()) + + assert core.calls, "기록이 아예 안 됐다" + assert core.calls[0]["session_id"] == 102 + assert core.calls[0]["user_id"] == 7 + + +def test_explicit_session_wins_over_context(): + from ai_server.observability.ai_call_log import record_ai_call + + core = _RecordingCore() + + async def main() -> None: + with trace_context("t", session_id=102): + # 호출부가 명시하면 그게 이긴다(voice_consumer 가 그렇게 쓴다). + record_ai_call( + core, + request_type="stt.transcribe", + model_name="m", + latency_ms=10, + status="SUCCESS", + session_id=999, + ) + await asyncio.sleep(0) + await asyncio.sleep(0) + + asyncio.run(main()) + + assert core.calls[0]["session_id"] == 999 + + +def test_no_session_context_records_none_not_crash(): + from ai_server.observability.ai_call_log import record_ai_call + + core = _RecordingCore() + + async def main() -> None: + # 문서 분석처럼 세션이 없는 흐름 — 기록은 되고 session_id 만 비어야 한다. + with trace_context("t"): + record_ai_call( + core, + request_type="analyze.document", + model_name="m", + latency_ms=10, + status="SUCCESS", + ) + await asyncio.sleep(0) + await asyncio.sleep(0) + + asyncio.run(main()) + + assert core.calls[0]["session_id"] is None + + +def test_langchain_callback_picks_up_session_from_context(): + from ai_server.observability.llm_logging_callback import CoreAiLogCallback + + core = _RecordingCore() + cb = CoreAiLogCallback( + core_client=core, request_type="generate.feedback.panel", default_model="m" + ) + + async def main() -> None: + with trace_context("t", session_id=102, user_id=7): + await cb._fire( + request_type="generate.feedback.panel", + model_name="m", + input_tokens=1, + output_tokens=2, + latency_ms=3, + status="SUCCESS", + error_message=None, + ) + await asyncio.sleep(0) + await asyncio.sleep(0) + + asyncio.run(main()) + + # 체인 빌더는 세션을 모른다 — 기록 시점에 컨텍스트에서 채워야 한다. + assert core.calls[0]["session_id"] == 102 + assert core.calls[0]["user_id"] == 7 + + +def test_session_id_also_appears_in_logs(): + cap = _capture() + log = structlog.get_logger("t") + + with trace_context("t", session_id=102): + log.info("inside") + + # 로그만 봐도 어느 면접의 호출인지 알 수 있어야 한다. + assert cap.entries[0]["session_id"] == 102 diff --git a/docs/observability.md b/docs/observability.md index 4e8a9827..fea04b67 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -4,6 +4,26 @@ --- +## 0. AI 호출 비용 귀속 (`ai_request_logs.session_id`) + +**2026-09-28 이전에는 `stt.transcribe` 말고 전부 NULL 이었다.** 2026-08-06 이래 쌓인 +`generate.*`·`feedback.*`·`tts.synthesize`·`embedding.embed` 로그 전부가 세션을 모른 채 +기록됐다. `record_ai_call` 과 `CoreAiLogCallback` 은 둘 다 `session_id` 를 **받을 수 있게** +만들어져 있었는데 실제로 넘기는 호출부가 voice consumer 하나뿐이었다. + +그래서 "이 면접이 얼마나 들었나", "어느 세션이 게이트웨이 크레딧을 태웠나" 를 답할 수 없었다 — +2026-09-17 에 실제로 크레딧이 소진돼 운영 AI 가 402 로 죽었을 때 정확히 이 질문을 못 했다. + +체인 빌더마다 인자로 넘기면 LangChain 콜백 생성 지점 6곳과 체인 팩토리 시그니처가 전부 +오염된다. 관측값은 횡단 관심사이므로 **traceId 와 같은 방식으로 컨텍스트에 싣는다** — +`trace_context(trace_id, session_id=…, user_id=…)` 가 컨슈머 진입점에서 묶고, +`record_ai_call` 과 `CoreAiLogCallback` 이 명시 인자가 없을 때 거기서 읽는다. +명시 인자가 있으면 그쪽이 이긴다. + +세션이 없는 흐름(문서 분석)은 그대로 NULL 이다 — 억지로 채우지 않는다. + +--- + ## 1. 분산 추적 (X-Trace-Id) ### 1.1 ID 생성