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>
57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
"""Audit — 4B 가 자체 답변한 경우 금지 패턴 검출.
|
|
|
|
escalate_to_26b=False 인 이벤트에만 호출. 위반 검출 시 policy_violation=true 로
|
|
analyze_events 에 기록되고 야간 sweep 에서 under_escalation 후보로 포획된다.
|
|
|
|
detection_patterns 는 Python re.search() 로 평가 (Postgres regex 아님).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from functools import lru_cache
|
|
from typing import Iterable
|
|
|
|
from policy.loader import load_policy
|
|
from policy.schema import DomainPolicy, ForbiddenRule
|
|
|
|
|
|
@lru_cache(maxsize=256)
|
|
def _compiled_patterns(pattern_tuple: tuple[str, ...]) -> tuple[re.Pattern[str], ...]:
|
|
return tuple(re.compile(p) for p in pattern_tuple)
|
|
|
|
|
|
def _rules_for_subject(
|
|
policy: DomainPolicy, subject_domain: str
|
|
) -> Iterable[ForbiddenRule]:
|
|
for rule in policy.forbidden_for_4b:
|
|
if subject_domain in rule.applies_when_subject_in:
|
|
yield rule
|
|
|
|
|
|
def check_4b_output_violations(
|
|
output_text: str,
|
|
subject_domain: str,
|
|
*,
|
|
policy: DomainPolicy | None = None,
|
|
) -> list[str]:
|
|
"""Return list of violated forbidden-rule IDs (빈 리스트면 위반 없음).
|
|
|
|
Parameters
|
|
----------
|
|
output_text: 4B 가 생성한 자체 답변 텍스트.
|
|
subject_domain: routing 에서 결정된 도메인 이름. fallback 도메인은 `generic`.
|
|
policy: 주입용 (테스트). None 이면 load_policy().
|
|
"""
|
|
if not output_text:
|
|
return []
|
|
if policy is None:
|
|
policy = load_policy()
|
|
|
|
violations: list[str] = []
|
|
for rule in _rules_for_subject(policy, subject_domain):
|
|
patterns = _compiled_patterns(tuple(rule.detection_patterns))
|
|
if any(p.search(output_text) for p in patterns):
|
|
violations.append(rule.id)
|
|
return violations
|