Skip to content

feat(ai): trace_context 구현 — 문서에만 있던 함수를 실제로 만든다 #263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
i3months merged 1 commit into from
Sep 28, 2026
Merged

feat(ai): trace_context 구현 — 문서에만 있던 함수를 실제로 만든다 #263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions ai/src/ai_server/CLAUDE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,14 @@ async def consume(message: AbstractIncomingMessage) -> None:
await mark_processed(envelope.message_id)
```

> **`trace_context` 는 2026-09-28 까지 이 문서에만 있었고 구현이 없었다.** 이 패턴을 따라
> 쓰면 ImportError 가 났고, 실제 컨슈머들은 traceId 를 로그에 전혀 싣지 않았다 — 운영 AI
> 로그에 trace 가 하나도 없던 이유다. 지금은 `observability/trace.py` 에 있고
> `consume_with_failure_signal`(생성·분석 컨슈머 공통) + `voice`/`tts` 컨슈머가 쓴다.
> structlog 기본 프로세서 첫 번째가 `merge_contextvars` 라 한 번 묶으면 그 구간의 **모든
> 로그 줄에 자동으로 실린다**. contextvars 는 asyncio 태스크별로 격리돼 동시 처리 메시지끼리
> 섞이지 않는다(`tests/test_trace_context.py` 가 고정).

### `analyzer/`
- 분석 use case 단위 (`resume_analyzer.py`, `repository_analyzer.py`, `web_resume_analyzer.py`)
- 소스 추출 추상화는 `analyzer/sources/` (PDF/GitHub/웹/텍스트), 임베딩 인제스트는 `_embedding_step.py`
Expand Down
124 changes: 63 additions & 61 deletions ai/src/ai_server/messaging/consumers/failure_signal.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from aio_pika.abc import AbstractIncomingMessage
from pydantic import BaseModel

from ai_server.observability.trace import trace_context
from ai_server.messaging.idempotency import LruIdempotencyStore
from ai_server.messaging.publisher import CallbackPublisher
from ai_server.model.envelope import Envelope
Expand Down Expand Up @@ -118,68 +119,69 @@ async def consume_with_failure_signal(
)
raise

if idempotency.is_seen_then_mark(envelope.message_id):
log.info(
f"{domain}.idempotent.skip",
message_id=envelope.message_id,
trace_id=envelope.trace_id,
)
return

async def _publish(payload: BaseModel) -> None:
await publisher.publish(
routing_key=routing_key,
message_type=message_type,
payload=payload,
trace_id=envelope.trace_id,
correlation_id=envelope.message_id,
context=envelope.context,
)
with trace_context(envelope.trace_id):
if idempotency.is_seen_then_mark(envelope.message_id):
log.info(
f"{domain}.idempotent.skip",
message_id=envelope.message_id,
trace_id=envelope.trace_id,
)
return

async def _publish(payload: BaseModel) -> None:
await publisher.publish(
routing_key=routing_key,
message_type=message_type,
payload=payload,
trace_id=envelope.trace_id,
correlation_id=envelope.message_id,
context=envelope.context,
)

log_ids: dict[str, Any] = {
"message_id": envelope.message_id,
"trace_id": envelope.trace_id,
}
session_id = getattr(envelope.payload, "session_id", None)
if session_id is not None:
log_ids["session_id"] = session_id
log_ids: dict[str, Any] = {
"message_id": envelope.message_id,
"trace_id": envelope.trace_id,
}
session_id = getattr(envelope.payload, "session_id", None)
if session_id is not None:
log_ids["session_id"] = session_id

try:
payload = await process(envelope)
except Exception as exc: # noqa: BLE001
if expected_errors and isinstance(exc, expected_errors):
# 예상된 도메인 실패(빈 PDF·404 URL 등 일상 입력)는 traceback 없는 warning —
# ERROR 레벨 스택트레이스로 일상 실패가 알람을 울리지 않게 한다.
log.warning(
f"{domain}.{action}.failed",
error_code=getattr(exc, "code", None),
retriable=getattr(exc, "retriable", None),
error=str(exc),
**log_ids,
)
else:
log.exception(f"{domain}.{action}.failed", **log_ids)
try:
# 팩토리 자체가 죽어도(검증 오류 등) 같은 안전망을 태운다 —
# try 밖이면 unmark 없이 DLQ 로 가서 재주입이 duplicate skip 으로 삼켜진다.
fallback = failed_payload(envelope.payload, exc)
await _publish(fallback)
except Exception: # noqa: BLE001
log.exception(
f"{domain}.failed_callback.publish_failed",
**log_ids,
)
idempotency.unmark(envelope.message_id)
raise exc
return
payload = await process(envelope)
except Exception as exc: # noqa: BLE001
if expected_errors and isinstance(exc, expected_errors):
# 예상된 도메인 실패(빈 PDF·404 URL 등 일상 입력)는 traceback 없는 warning —
# ERROR 레벨 스택트레이스로 일상 실패가 알람을 울리지 않게 한다.
log.warning(
f"{domain}.{action}.failed",
error_code=getattr(exc, "code", None),
retriable=getattr(exc, "retriable", None),
error=str(exc),
**log_ids,
)
else:
log.exception(f"{domain}.{action}.failed", **log_ids)
try:
# 팩토리 자체가 죽어도(검증 오류 등) 같은 안전망을 태운다 —
# try 밖이면 unmark 없이 DLQ 로 가서 재주입이 duplicate skip 으로 삼켜진다.
fallback = failed_payload(envelope.payload, exc)
await _publish(fallback)
except Exception: # noqa: BLE001
log.exception(
f"{domain}.failed_callback.publish_failed",
**log_ids,
)
idempotency.unmark(envelope.message_id)
raise exc
return

try:
await _publish(payload)
except Exception:
idempotency.unmark(envelope.message_id)
raise
log.info(
f"{domain}.{action}.done",
**log_ids,
**(done_fields(payload) if done_fields is not None else {}),
)
try:
await _publish(payload)
except Exception:
idempotency.unmark(envelope.message_id)
raise
log.info(
f"{domain}.{action}.done",
**log_ids,
**(done_fields(payload) if done_fields is not None else {}),
)
96 changes: 49 additions & 47 deletions ai/src/ai_server/messaging/consumers/tts_consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import structlog
from aio_pika.abc import AbstractIncomingMessage

from ai_server.observability.trace import trace_context
from ai_server.chain.sentence_split import next_sentences
from ai_server.messaging.consumers.failure_signal import unmark_on_error
from ai_server.messaging.idempotency import LruIdempotencyStore
Expand Down Expand Up @@ -81,65 +82,66 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
)
raise

if self._idempotency.is_seen_then_mark(envelope.message_id):
log.info("tts.idempotent.skip", message_id=envelope.message_id)
return
with trace_context(envelope.trace_id):
if self._idempotency.is_seen_then_mark(envelope.message_id):
log.info("tts.idempotent.skip", message_id=envelope.message_id)
return

# 마킹 이후 어떤 예외든 unmark — 콜백 0건 DLQ 재주입 삼킴 방지 (F6).
async with unmark_on_error(self._idempotency, envelope.message_id):
req = envelope.payload
log.info(
"tts.synthesize.start",
message_id=envelope.message_id,
session_id=req.session_id,
interview_message_id=req.message_id,
streaming=self._notifier is not None,
trace_id=envelope.trace_id,
)
# 마킹 이후 어떤 예외든 unmark — 콜백 0건 DLQ 재주입 삼킴 방지 (F6).
async with unmark_on_error(self._idempotency, envelope.message_id):
req = envelope.payload
log.info(
"tts.synthesize.start",
message_id=envelope.message_id,
session_id=req.session_id,
interview_message_id=req.message_id,
streaming=self._notifier is not None,
trace_id=envelope.trace_id,
)

if self._notifier is not None:
result = await self._synthesize_segmented(envelope, req)
else:
result = await self._synthesize_once(envelope, req)
if result is None:
return # 실패 — 콜백은 내부에서 이미 발행
if self._notifier is not None:
result = await self._synthesize_segmented(envelope, req)
else:
result = await self._synthesize_once(envelope, req)
if result is None:
return # 실패 — 콜백은 내부에서 이미 발행

key = self._build_key(
req.session_id, req.message_id, result.content_type
)
try:
await self._storage.put_bytes(
key, result.audio_bytes, content_type=result.content_type
key = self._build_key(
req.session_id, req.message_id, result.content_type
)
except Exception as exc:
log.error("tts.storage.failed", error=str(exc), key=key)
try:
await self._storage.put_bytes(
key, result.audio_bytes, content_type=result.content_type
)
except Exception as exc:
log.error("tts.storage.failed", error=str(exc), key=key)
await self._publish(
envelope,
TtsCallbackPayload(
session_id=req.session_id,
message_id=req.message_id,
status="FAILED",
error_code="TTS_STORAGE_FAILED",
),
)
return

await self._publish(
envelope,
TtsCallbackPayload(
session_id=req.session_id,
message_id=req.message_id,
status="FAILED",
error_code="TTS_STORAGE_FAILED",
status="SUCCEEDED",
audio_key=key,
duration_sec=result.duration_sec,
),
)
return

await self._publish(
envelope,
TtsCallbackPayload(
log.info(
"tts.synthesize.done",
session_id=req.session_id,
message_id=req.message_id,
status="SUCCEEDED",
audio_key=key,
duration_sec=result.duration_sec,
),
)
log.info(
"tts.synthesize.done",
session_id=req.session_id,
interview_message_id=req.message_id,
key=key,
)
interview_message_id=req.message_id,
key=key,
)

async def _synthesize_once(
self, envelope: Envelope[GenerateTtsRequest], req: GenerateTtsRequest
Expand Down
Loading
Loading