diff --git a/app/ai/envelope.py b/app/ai/envelope.py new file mode 100644 index 0000000..c0c6330 --- /dev/null +++ b/app/ai/envelope.py @@ -0,0 +1,97 @@ +"""EscalationEnvelope — 4B → 26B 핸드오프 계약. + +4B 가 "자신이 처리 못한다" 고 판단했을 때 26B 에게 전달하는 구조화 메시지. +26B 는 distilled_context 로 방향을 잡고 original_pointers 로 필요한 원문만 재조회. + +PR-A 는 dataclass 계약만 정의. 실제 생성/소비는 PR-B 의 escalation_service 가 담당. +""" + +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass, field +from typing import Any + + +ValidFromStage = { + "triage", + "classify", + "summarize_short", + "advice_trigger", + "night_sweep", + "ask_pre", + "unknown", # 호환성용 +} + + +@dataclass(frozen=True) +class EscalationEnvelope: + from_stage: str + escalation_reasons: tuple[str, ...] + risk_flags: tuple[str, ...] + distilled_context: str + original_pointers: dict[str, Any] = field(default_factory=dict) + synthesis_directives: tuple[str, ...] = () + user_intent: str | None = None + draft_hint: str | None = None + + def __post_init__(self) -> None: + if self.from_stage not in ValidFromStage: + raise ValueError( + f"from_stage '{self.from_stage}' not in {ValidFromStage}" + ) + if not isinstance(self.escalation_reasons, tuple): + raise TypeError("escalation_reasons must be tuple (for hashability)") + if not isinstance(self.risk_flags, tuple): + raise TypeError("risk_flags must be tuple (for hashability)") + if not isinstance(self.synthesis_directives, tuple): + raise TypeError("synthesis_directives must be tuple (for hashability)") + + # -- 26B system prompt 주입용 텍스트 ----------------------------------- + def to_system_injection(self) -> str: + lines = [ + "=== ESCALATION ENVELOPE (from 4B) ===", + f"from_stage: {self.from_stage}", + f"reasons: {', '.join(self.escalation_reasons) or '(none)'}", + f"risk_flags: {', '.join(self.risk_flags) or '(none)'}", + ] + if self.user_intent: + lines.append(f"user_intent: {self.user_intent}") + if self.draft_hint: + lines.append(f"draft_hint: {self.draft_hint}") + + if self.synthesis_directives: + lines.append("") + lines.append("synthesis_directives (각 risk_flag 별 지시사항, 반드시 준수):") + for d in self.synthesis_directives: + lines.append(f" - {d}") + + if self.distilled_context: + lines.append("") + lines.append("distilled_context (4B 가 압축한 요지 — 참고용, 숫자·인용은 원문 재확인 필수):") + lines.append(self.distilled_context) + + if self.original_pointers: + lines.append("") + lines.append("original_pointers (필요 시 재조회):") + lines.append(json.dumps(self.original_pointers, ensure_ascii=False, indent=2)) + + return "\n".join(lines) + + # -- JSON round-trip --------------------------------------------------- + def to_json(self) -> str: + return json.dumps(asdict(self), ensure_ascii=False) + + @classmethod + def from_json(cls, s: str) -> EscalationEnvelope: + raw = json.loads(s) + return cls( + from_stage=raw["from_stage"], + escalation_reasons=tuple(raw.get("escalation_reasons", ())), + risk_flags=tuple(raw.get("risk_flags", ())), + distilled_context=raw.get("distilled_context", ""), + original_pointers=raw.get("original_pointers", {}) or {}, + synthesis_directives=tuple(raw.get("synthesis_directives", ())), + user_intent=raw.get("user_intent"), + draft_hint=raw.get("draft_hint"), + )