"""GROUNDING-V2 §2-3.5 — A0 후보 수집. **거르기 전에 건진다.** 설계: 계획 §1.8. ★왜 여기인가 — 뒤 단계의 제외 규칙들이 **여러 겹으로** 고증 대상을 걸러 낸다: entity_character_list 「여러 번 출현하는 경우만」 entity_all/prop 착용물 · 고정 설비 · 「텍스트만 다른 종이류」 entity_extract_v4/prop 같은 부류를 **또** entity_filter 저빈도 강등 entity_extractor_v2 상세 단계에서 **또** 그 규칙들은 **이미지 일관성**을 위한 것이라 그 자체로는 옳지만, 고증 대상을 정확히 같이 걸러 낸다. 그래서 **제외보다 먼저** 여기서 건진다. ★``entity_all`` schema 를 고치는 안은 기각됐다 — ``additionalProperties=false`` (``name`` + ``shot_count`` 뿐)이고 ``entity_extractor_v4.py:69`` 가 ``e["name"]`` 만 텍스트로 내려 provenance 를 버린다. 그래서 **전용 sidecar CP** 다. ★★그리고 sidecar 로만 두면 **결속할 엔티티가 아예 없다** — ``entity_all`` 은 **shot description 만** 읽고(``entity_lister.py:170``) shot schema 는 ``characters`` 가 필수가 아니다. 한 번만 크게 나오는 대상은 샷 구조에 없으면 ``entity_all`` 이 볼 기회조차 없다. 그래서 A0 는 관찰용이 아니라 **후보 승격 통로**다 — 산출이 다음 단계 입력에 합류한다. """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional, Sequence from app.modules.prompt_loader import resolve_effective from app.modules.pipeline.grounding_subject import build_subject logger = logging.getLogger(__name__) A0_CONTRACT_VERSION = 1 _MODULE = "grounding_a0" #: ★명시 pin — latest 자동 선택에 기대지 않는다 (계획 §4-6). PROMPT_PACK_VERSION = "2.202608301300" SYSTEM_STEM = "system" SCHEMA_STEM = "a0_schema" STEP_NAME = "grounding_a0" #: A0 산출이 앉는 자리. production step id 와 안 겹친다. A0_STEP_ID = "grounding_a0_candidates" def _sha16(text: str) -> str: import hashlib return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16] def _call_structured(**kwargs) -> Dict[str, Any]: """``llm_client`` 로 가는 유일한 문 (grounding_classifier 와 같은 이유).""" from app.modules.llm.llm_client import call_structured return call_structured(**kwargs) def load_pack(*, db=None, version: Optional[str] = None) -> Dict[str, Any]: ver = version or PROMPT_PACK_VERSION resolved = { SYSTEM_STEM: resolve_effective(_MODULE, SYSTEM_STEM, kind="prompt", version=ver, db=db), SCHEMA_STEM: resolve_effective(_MODULE, SCHEMA_STEM, kind="schema", version=ver, db=db), } return { "module": _MODULE, "version": ver, "stems": resolved, "pack_manifest_hash": _sha16("|".join( f"{s}:{r['source']}:{r['version']}:{r['raw_content_hash']}" for s, r in sorted(resolved.items()))), } def build_user_prompt( fulltext: str, *, era: str, region: str, shot_lines: Sequence[str] = (), ) -> str: """★원문 **전체**를 넣는다. 자르지 않는다. shot description 만 보면 안 되는 이유는 모듈 머리글에 있다 — 샷 프롬프트가 복장을 금지하고 중요 물체만 남기므로 **fulltext 가 필수**다. """ parts = [ "## 목표 맥락", f"- 시대: {era}", f"- 지역: {region}", "", ] if shot_lines: parts += ["## 계획된 샷 (참고 — 여기 없는 것도 건집니다)", *shot_lines, ""] parts += ["## 원문", fulltext] return "\n".join(parts) def requested_model(project_config: Optional[Dict] = None) -> str: """★**부탁할 모델 별칭.** 호출 **전에** 무료로 안다. A0 는 판정자를 명시로 안 받고 manifest 기본값을 탄다. 그 기본값이 바뀌면 **다른 모델이 건진 후보**인데, 신원에 안 접으면 옛 후보를 새 판에 재생한다 (검색은 `model`, 분류는 `judges` 를 접는 것과 같은 이유다 — Codex). ★물리 모델은 호출 전에 못 정한다(tier fallback). 그것은 산 뒤 `fingerprint.judge_physical_model` 로 남는다. """ from app.modules.llm.llm_client import _resolve_model return _resolve_model(STEP_NAME, project_config) def payload_identity(system: str, user: str, schema: Any) -> str: """★이 호출의 신원. **한 곳에서만** 만든다 — 두 곳에 적으면 한쪽만 고쳐진다. `grounding_classifier` · `grounding_claims_search` 와 같은 자리다. """ return _sha16("\x1e".join((system, user, str(sorted(schema.items()))))) def planned_payload(fulltext: str, *, era: str, region: str, shot_lines: Sequence[str] = (), db=None, pack_version: Optional[str] = None) -> str: """**보내기 전에** 이 호출의 신원을 낸다. ★무료. ★`collect_candidates` 와 **같은 함수들**로 만든다 — `load_pack` · `build_user_prompt` · `payload_identity`. 팩 **버전**만 접으면 같은 버전 안에서 지문 bytes 가 바뀐 것을 못 잡고, 옛 후보를 새 판에 재생한다. """ pack = load_pack(db=db, version=pack_version) return payload_identity( pack["stems"][SYSTEM_STEM]["content"], build_user_prompt(fulltext, era=era, region=region, shot_lines=shot_lines), pack["stems"][SCHEMA_STEM]["content"]) def collect_candidates( fulltext: str, *, project_id: str, episode_id: str, era: str, region: str, shot_lines: Sequence[str] = (), project_config: Optional[Dict] = None, opik_metadata: Optional[Dict] = None, db=None, pack_version: Optional[str] = None, strict_single_attempt: bool = False, ) -> Dict[str, Any]: """원문에서 고증 후보를 건진다. ★**검색을 하지 않는다.** Raises: ValueError: 원문이 비었거나 시대/지역이 없을 때 — 빈 값으로 부르면 「후보 0」이 「깨끗함」으로 읽힌다. """ if not (fulltext or "").strip(): raise ValueError("원문이 비었다 — 후보 0을 「깨끗함」으로 읽게 된다") if not (era or "").strip() or not (region or "").strip(): raise ValueError( f"시대/지역이 없다 (era={era!r}, region={region!r}) — " "맥락 없이 건지면 시대 대상을 못 알아본다") if not (project_id or "").strip() or not (episode_id or "").strip(): raise ValueError( f"범위가 없다 (project_id={project_id!r}, episode_id={episode_id!r}) — " "subject id 를 발급할 수 없다") pack = load_pack(db=db, version=pack_version) system = pack["stems"][SYSTEM_STEM]["content"] schema = pack["stems"][SCHEMA_STEM]["content"] user = build_user_prompt(fulltext, era=era, region=region, shot_lines=shot_lines) # ★신원은 **한 함수**가 만든다 — 여기서 따로 조립하면 재생 gate 가 # 잠그는 것과 실제로 나가는 것이 갈린다. payload_hash = payload_identity(system, user, schema) usage: Dict[str, Any] = {} result = _call_structured( step=STEP_NAME, system_prompt=system, user_prompt=user, response_schema=schema, project_config=project_config, schema_name="grounding_a0_candidates", opik_metadata=opik_metadata, usage_sink=usage, **({"enable_fallback": False, "num_retries": 0} if strict_single_attempt else {}), ) raw = (result or {}).get("candidates") or [] kept: List[Dict[str, Any]] = [] dropped: List[Dict[str, Any]] = [] for c in raw: quote = (c.get("source_quote") or "").strip() # ★원문에 없는 인용은 후보가 아니다 — 지어낸 것이다. # 글자로 뜻을 판단하는 것이 아니라 **인용이 실재하는지**만 본다. if not (quote and quote in fulltext): dropped.append({**c, "drop_reason": "source_quote 가 원문에 없다"}) continue # ★subject id 를 **여기서** 발급한다. 뒤에서 다시 발급하면 # 같은 대상에 다른 id 가 붙어 A0 가 건진 근거(원문 인용)가 끊긴다. anchor = (c.get("source_anchor") or "").strip() surface = (c.get("surface_form") or "").strip() owner = (c.get("owner_type") or "").strip() if not (anchor and surface and owner): dropped.append({**c, "drop_reason": "anchor/표면형/owner 가 비었다"}) continue kept.append({ **c, **build_subject( project_id=project_id, episode_id=episode_id, source_anchor=anchor, surface_form=surface, owner_type=owner, provenance={ "a0_pack_version": pack["version"], "a0_pack_hash": pack["pack_manifest_hash"], "a0_model_alias": usage.get("alias"), "a0_physical_model": usage.get("physical_model"), "source_step": STEP_NAME, }, ), # build_subject 가 덮어쓴 값을 원문 그대로 되돌린다. "source_quote": quote, }) if dropped: logger.warning("grounding_a0: 원문에 없는 인용 %d건 버림", len(dropped)) return { "contract_version": A0_CONTRACT_VERSION, "candidates": kept, "hallucinated": dropped, "fingerprint": { "prompt_module": _MODULE, "prompt_version": pack["version"], "pack_manifest_hash": pack["pack_manifest_hash"], "payload_hash": payload_hash, # ★**부탁한** 것과 **응답이 말한** 것을 따로 남긴다 — tier # fallback 이 있으면 둘이 갈리고, 갈린 것을 못 보면 「어느 # 모델이 건졌나」를 영영 모른다. "requested_model_alias": requested_model(project_config), "judge_model_alias": usage.get("alias"), "judge_physical_model": usage.get("physical_model"), "era": era, "region": region, }, "search_calls": 0, }