"""still_variants — 스틸 프롬프트 변형 2종 저작·조립 (2026-07-17). 사용자 확정 스펙(동일 프롬프트 반복 롤 → LLM 저작 변형 2종): Sol 이 샷별 "시각 접근(카메라·프레이밍·구도)" 변형 2종을 저작하고, 코드가 base 스틸 프롬프트(불변 안전계약, build_still_prompt 산출) + VISUAL APPROACH 절 + hierarchy tail(base 우선 계약, Codex 설계 리뷰 BLOCKING-1) 순으로 조립한다. base 전문은 그대로 보존되므로 안전절 소실이 구조적으로 불가능하다. A/B 4택1 라벨 계약(ab_label_map): A=변형0+콘티 / B=변형1+콘티 / C=변형0+무콘티 / D=변형1+무콘티 — 프롬프트는 A==C, B==D 이고 콘티 유무는 roll_refs(참조) 관할. 판정자에게 매핑은 미노출(블라인드). 저작 캐시·재시도 계약은 seed 2R(_author_seed_variants) 패턴 정합: author_fp(입력 전체 지문) 일치+재검증 통과=0콜 재사용, 구조 검증 실패= 이전 위반 병기 1회 재시도 → 소진=ValueError(호출측 샷 단위 fail-closed). """ from __future__ import annotations import logging from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence logger = logging.getLogger(__name__) VARIANTS_PACK_MODULE = "still_prompt_variants" VARIANTS_PACK_VERSION_MAP = { "1": "1.202607172227", } AUTHOR_MODEL = "gpt" # 저작=Sol (seed 2R 관례), 판정=gemini-pro (multiroll) AUTHOR_STEP_TAG = "still_variant_author" VARIANT_COUNT = 2 # 라벨 매핑·flip·조립 계약 버전 — config_hash 스탬프 대상 (R7) STILL_VARIANTS_CONTRACT_VERSION = 1 APPROACH_HEAD = ( "VISUAL APPROACH — this take (camera, framing, composition):" ) def resolve_variants_pack_version(version: str) -> str: if version not in VARIANTS_PACK_VERSION_MAP: raise ValueError(f"still_prompt_variants 팩 버전 없음: {version}") return VARIANTS_PACK_VERSION_MAP[version] def _load(name: str, version: str) -> str: from app.modules.prompt_loader import load_prompt resolved = resolve_variants_pack_version(version) return load_prompt(VARIANTS_PACK_MODULE, name, version=resolved) def load_author_sys(version: str = "1") -> str: return _load("author_sys", version) def load_hierarchy_tail(version: str = "1") -> str: """BLOCKING-1: approach 가 프롬프트 말미 위치로 base SOT 를 상쇄하는 경로 차단 — 항상 approach **뒤** 최후미에 부착되는 base 우선 계약.""" return _load("hierarchy_tail", version) def load_variant_judge_header(version: str = "1") -> str: """HIGH-6: 변형 모드 judge header — 기본 header('all candidates were generated from this')는 변형 롤에서 거짓. variant/conti 매핑 미노출.""" return _load("judge_header", version) def build_author_schema() -> Dict[str, Any]: return { "type": "object", "additionalProperties": False, "properties": { "variants": { "type": "array", "minItems": VARIANT_COUNT, "maxItems": VARIANT_COUNT, "items": { "type": "object", "additionalProperties": False, "properties": { "approach_ko": {"type": "string"}, "prompt_en": {"type": "string"}, }, "required": ["approach_ko", "prompt_en"], }, }, }, "required": ["variants"], } def _norm(text: str) -> str: """비교용 정규화 — whitespace collapse + casefold (HIGH-5).""" return " ".join(str(text).split()).casefold() def validate_still_variants(data: Any) -> List[str]: """구조 검증 — violations 목록 반환(빈 목록=통과). seed 관례 정합. HIGH-5: approach_ko·prompt_en 각각 타입·non-empty, 정규화 후 두 변형이 **두 필드 모두** 상이해야 한다(approach 만 다르고 prompt_en 동일 통과 차단). """ violations: List[str] = [] variants = (data or {}).get("variants") if isinstance(data, dict) else None if not isinstance(variants, list) or len(variants) != VARIANT_COUNT: return [ f"variants 는 정확히 {VARIANT_COUNT}개 배열이어야 함: " f"{type(variants).__name__}" + (f"({len(variants)})" if isinstance(variants, list) else "") ] for i, item in enumerate(variants): if not isinstance(item, dict): violations.append(f"variants[{i}] 가 object 아님") continue for key in ("approach_ko", "prompt_en"): val = item.get(key) if not isinstance(val, str) or not val.strip(): violations.append(f"variants[{i}].{key} 비어있음/비문자열") if violations: return violations if _norm(variants[0]["prompt_en"]) == _norm(variants[1]["prompt_en"]): violations.append("두 변형의 prompt_en 이 정규화 후 동일 — 실질 상이 필요") if _norm(variants[0]["approach_ko"]) == _norm(variants[1]["approach_ko"]): violations.append("두 변형의 approach_ko 가 정규화 후 동일") return violations def ab_label_map() -> Dict[str, Dict[str, Any]]: """A/B 4택1 라벨 계약 (고정 상수 — record 감사·lineage 판정 SOT).""" return { "A": {"variant": 0, "conti": True}, "B": {"variant": 1, "conti": True}, "C": {"variant": 0, "conti": False}, "D": {"variant": 1, "conti": False}, } def assemble_still_roll_prompts( *, base_prompt: str, variants: Sequence[Mapping[str, Any]], labels: Sequence[str], pack_version: str = "1", ) -> Dict[str, str]: """라벨→생성 프롬프트 — base 전문 1회 + approach 절 + hierarchy tail. labels 2개(비A/B)=[변형0, 변형1], 4개(A/B 4택1)=[0,1,0,1] cycling — ab_label_map 과 정합(A/C=변형0, B/D=변형1). 그 외 개수=ValueError. """ if len(variants) != VARIANT_COUNT: raise ValueError(f"variants 수 {len(variants)} != {VARIANT_COUNT}") if len(labels) not in (VARIANT_COUNT, VARIANT_COUNT * 2): raise ValueError( f"labels 수 {len(labels)} — {VARIANT_COUNT} 또는 " f"{VARIANT_COUNT * 2} 만 허용" ) tail = load_hierarchy_tail(pack_version).strip() out: Dict[str, str] = {} for i, lab in enumerate(labels): body = str(variants[i % VARIANT_COUNT]["prompt_en"]).strip() out[lab] = "\n\n".join( [base_prompt, f"{APPROACH_HEAD}\n{body}", tail] ) return out def build_author_user_content( *, base_prompt: str, shot_text: str, prior_violations: Optional[Sequence[str]] = None, ) -> str: """저작 입력 — base·샷 원문 전체 무절단 전달 (CLAUDE.md 절대 규칙). 재시도 시 이전 위반 목록만 병기하고 본문은 그대로 유지 (HIGH-5). """ parts = [ "STILL PROMPT (immutable base contract — the approaches you author" " will be appended to this):", base_prompt, "SHOT TEXT (authoritative, original language):", shot_text, ] if prior_violations: parts.append( "PRIOR ATTEMPT VIOLATIONS (your previous output failed" " validation — fix all of these):\n" + "\n".join(f"- {v}" for v in prior_violations) ) return "\n\n".join(parts) def author_still_variants( *, base_prompt: str, shot_text: str, cached_entry: Optional[Dict[str, Any]], project_config: Optional[Dict[str, Any]], call_structured_fn: Optional[Callable[..., Dict[str, Any]]] = None, opik_metadata: Optional[Dict[str, Any]] = None, pack_version: str = "1", ) -> tuple[Dict[str, Any], str, bool]: """Sol 저작 → (저작본, author_fp, 캐시 재사용 여부). author_fp = author_sys+입력 전문+스키마+물리 모델+계약 버전 지문 — 일치+재검증 통과=0콜 재사용(비결정 저작이 하류 지문을 흔드는 창 제거), 캐시 오염(검증 실패)=재저작. 검증 실패=위반 병기 1회 재시도 → 소진=ValueError (호출측 샷 단위 fail-closed — Codex 합의 4). """ import hashlib import json as _json from app.core.config import settings if call_structured_fn is None: from app.modules.llm.llm_client import call_structured call_structured_fn = call_structured author_sys = load_author_sys(pack_version) schema = build_author_schema() user_content = build_author_user_content( base_prompt=base_prompt, shot_text=shot_text, ) physical_model = str(getattr(settings, "openai_model", "")) h = hashlib.sha256() for part in ( author_sys, user_content, _json.dumps(schema, sort_keys=True), physical_model, str(STILL_VARIANTS_CONTRACT_VERSION), ): h.update(part.encode("utf-8")) h.update(b"\x00") author_fp = h.hexdigest()[:16] if ( cached_entry and cached_entry.get("author_fp") == author_fp and not validate_still_variants(cached_entry.get("author")) ): return dict(cached_entry["author"]), author_fp, True pc = { **(project_config or {}), AUTHOR_STEP_TAG: {"model": AUTHOR_MODEL}, } violations: List[str] = [] for attempt in range(2): content = ( user_content if not violations else build_author_user_content( base_prompt=base_prompt, shot_text=shot_text, prior_violations=violations, ) ) data = call_structured_fn( AUTHOR_STEP_TAG, author_sys, content, schema, project_config=pc, schema_name=AUTHOR_STEP_TAG, opik_metadata=opik_metadata, ) violations = validate_still_variants(data) if not violations: return data, author_fp, False logger.warning( "still_variant_author: 저작 검증 위반(attempt %d): %s", attempt + 1, violations[:6], ) raise ValueError( f"스틸 변형 저작 검증 실패(재시도 소진): {violations[:6]}" ) def build_ab_roll_refs( refs_a: Sequence[Any], refs_b: Sequence[Any], ) -> Dict[str, List[Any]]: """A/B 4택1 roll_refs — ab_label_map 의 conti 플래그로 브랜치 매핑. refs_a=콘티 포함(build_ab_branch_refs 첫 반환), refs_b=무콘티. 라벨별 독립 복사본 — 호출측 변형이 다른 라벨을 오염시키지 않는다. """ return { lab: list(refs_a if meta["conti"] else refs_b) for lab, meta in ab_label_map().items() } def winner_uses_conti(selected: str) -> bool: """승자 라벨 → 콘티 사용 여부 (conti lineage 부착 판정 SOT). 계약 밖 라벨=ValueError — 조용한 미부착(거짓 lineage 부재) 금지. """ m = ab_label_map() if selected not in m: raise ValueError(f"A/B 4택1 계약 밖 승자 라벨: {selected!r}") return bool(m[selected]["conti"])