Hyungi Ahn
99672292d3
프로덕션 컨테이너는 /app 을 cwd 로 실행하고 import 는 `from api...`, `from core...`, `from workers...` 처럼 무접두 스타일을 사용한다. PR-A 내부 import 가 `from app.policy...`, `from app.ai.envelope` 로 되어 있어서 컨테이너에서 ModuleNotFoundError 발생. 변경: - app/policy/*.py: `from app.policy.X` → `from policy.X` - app/services/prompt_versions.py: lazy import 도 `from policy.prompt_render` - app/ai/envelope.py: 영향 없음 (내부 import 없음) - tests/policy/*.py: 모두 `from policy.X` / `from ai.envelope` 로 통일 - tests/policy/conftest.py: 로컬 pytest 용 sys.path.insert(app/) 추가 (MacBook 에서 repo-root 기준 실행 시 app/ 를 package root 로 취급) CI: pytest tests/policy/ -q → 98 passed (로컬, 동일 결과) 프로덕션: docker exec fastapi python -c "from policy.loader import load_policy" → OK Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
"""ShadowLogger — Protocol + in-memory implementation.
|
|
|
|
Live 전환 전 1주 shadow 기간에 "만약 이 정책이면 어디로 라우팅했을지" 를 기록.
|
|
실제 DB writer (DBShadowLogger) 는 PR-B 의 책임. PR-A 는:
|
|
1. Protocol 로 인터페이스 확정.
|
|
2. InMemoryShadowLogger 로 테스트 가능한 fake 제공.
|
|
|
|
PR-B 가 Protocol 시그니처를 변경하지 않는 것이 불변식.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Protocol, runtime_checkable
|
|
|
|
from policy.routing import RoutingDecision
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ShadowRecord:
|
|
"""단일 shadow 이벤트 — InMemoryShadowLogger 가 dict 로 보관."""
|
|
|
|
doc_id: str
|
|
decision: RoutingDecision
|
|
actual_model_used: str
|
|
prompt_version: str
|
|
policy_version: str
|
|
recorded_at: datetime
|
|
extra: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@runtime_checkable
|
|
class ShadowLogger(Protocol):
|
|
"""PR-A 가 정의하는 shadow 기록 인터페이스.
|
|
|
|
PR-B 가 DBShadowLogger(ShadowLogger) 로 구현할 때 이 시그니처를 그대로 준수.
|
|
"""
|
|
|
|
async def record_would_route(
|
|
self,
|
|
*,
|
|
doc_id: str,
|
|
decision: RoutingDecision,
|
|
actual_model_used: str,
|
|
prompt_version: str,
|
|
policy_version: str,
|
|
extra: dict[str, Any] | None = None,
|
|
) -> None:
|
|
...
|
|
|
|
|
|
class InMemoryShadowLogger:
|
|
"""테스트 전용 구현. PR-B 의 DBShadowLogger 와 시그니처 호환."""
|
|
|
|
def __init__(self) -> None:
|
|
self._records: list[ShadowRecord] = []
|
|
|
|
async def record_would_route(
|
|
self,
|
|
*,
|
|
doc_id: str,
|
|
decision: RoutingDecision,
|
|
actual_model_used: str,
|
|
prompt_version: str,
|
|
policy_version: str,
|
|
extra: dict[str, Any] | None = None,
|
|
) -> None:
|
|
self._records.append(
|
|
ShadowRecord(
|
|
doc_id=doc_id,
|
|
decision=decision,
|
|
actual_model_used=actual_model_used,
|
|
prompt_version=prompt_version,
|
|
policy_version=policy_version,
|
|
recorded_at=datetime.now(timezone.utc),
|
|
extra=dict(extra or {}),
|
|
)
|
|
)
|
|
|
|
# --- Inspection helpers (테스트 전용) ----------------------------------
|
|
@property
|
|
def records(self) -> tuple[ShadowRecord, ...]:
|
|
return tuple(self._records)
|
|
|
|
def clear(self) -> None:
|
|
self._records.clear()
|
|
|
|
def count(self) -> int:
|
|
return len(self._records)
|