Hyungi Ahn
b401085518
frozen dataclass with from_stage / escalation_reasons / risk_flags / distilled_context / original_pointers / synthesis_directives / user_intent / draft_hint. JSON round-trip (to_json/from_json). to_system_injection() 으로 26B system prompt 에 주입할 텍스트 블록 생성 (risk_flags + directives + distilled_context 순). from_stage 는 whitelist 검증 (triage/classify/summarize_short/advice_trigger/ night_sweep/ask_pre/unknown). tuple 타입 강제 (mutability 방지). PR-B 의 escalation_service 가 이 계약을 사용. PR-A 는 계약만 정의. plan: ~/.claude/plans/wise-gliding-hippo.md Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
98 lines
3.6 KiB
Python
98 lines
3.6 KiB
Python
"""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"),
|
|
)
|