"""검색 그라운딩 형태 참조 — 원본어 검색 지시문 → 웹 이미지 검색 → VLM 1장 선택. 사용자 지시 (2026-07-28): "야외 구조물(건물이나 규모가 있는 모든 복잡한 것들)은 배경 생성시에(씨드 생성시에) 모두 **검색 기반 + VLM** 으로 하게 수정하고 **콘티 관련 부분도**". 설계 = docs/superpowers/specs/ 2026-07-28-search-grounded-seed-and-conti-entity-chain-design.md ## 왜 이 모양인가 (감독 말판 보드 실험 6~13회차 실측) 1. **`web_search` 기본값은 텍스트만 돌려준다.** 그대로 쓰면 검색 근거가 글로 요약되어 이미지 모델에 닿고, 산출의 현지성이 눈에 띄게 약해진다(실측: 표지판이 백색 무지로 비고, 근거를 못 찾아 검색을 8→20회 반복). 사진 자체를 받으려면 `search_content_types:["image","text"]` 를 켜야 한다. 2. **검색을 지휘하는 프롬프트가 시나리오 원본어여야 한다.** 영어 지시문 안에서 "원본어로 검색하라"고 시키는 간접 방식으로는 부족했다 — 검색이 관광지· 타지역으로 샜다. 언어는 **씬 원문에서 판정**한다(특정 언어 하드코딩 금지, 프로젝트 규칙 "작품 고유명사·시나리오 의존 코딩 금지"). 3. **회수분은 비교 선택 1회로 1장만 남긴다.** 지시문에서 관광지를 배제해도 웹이 그쪽을 밀어올린다(실측: 23장 중 다수가 관광 콘텐츠). 장마다 개별 판정하면 "그 중 하나"가 아니라 통과 집합이 되어 지시와 어긋난다. 4. **참조는 형태 전용**이다. 배치·구도·시점·날씨·시간대는 참조에서 가져오지 않는다 — 그쪽 권위는 도면과 CAMERA 계약이 갖는다. ## 격리 - 이 모듈은 **순수 함수 + 호출**만 담는다. DB 쓰기·스텝 오케스트레이션 없음. - 이미지 파일 경로는 호출자가 정한다(프로젝트 루트 안 — DB 경로 CHECK 제약). """ from __future__ import annotations import base64 import json import logging import urllib.request from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Tuple logger = logging.getLogger(__name__) REF_PACK_MODULE = "search_grounded_ref" REF_PACK_VERSION = "10" # 심판 머리말 조립 계약 — **팩 파일이 아니라 코드가 정하는 부분**이라 # `search_contract_sha`(팩 바이트 해시)에 잡히지 않는다. 이것이 바뀌면 같은 # 팩·같은 spec 이어도 심판에게 가는 입력이 달라지므로 그룹 재사용 지문에 # 나란히 싣는다(`_group_fingerprint`). 안 실으면 완료 CP 가 **옛 머리말로 # 고른 사진**을 새 계약인 것처럼 재사용한다 — `search_contract_sha` 주석이 # v9 에서 같은 이유로 남긴 경고와 같은 종류다. # 1 = structure_desc 전문(layout narration + 항목별 위치 서술) # 2 = 항목 이름만 (2026-08-04, prompt diet ⑧) PICK_HEAD_CONTRACT_VERSION = "2" # 팩 버전 매핑 — structure_seed 관례 동형(버전.YYYYMMDDHHmm). 없는 버전은 # 명시 실패시킨다(조용한 fallback 이 사문화된 팩을 만든다). REF_PACK_VERSION_MAP = { # v1 (2026-07-28): 원본어 검색 지시문 저작 / 비교 선택 판정 / 형태 전용 # 계약절. 감독 말판 보드 실험 6~13회차 실측을 그대로 옮긴 초판. "1": "1.202607281800", # v2 (2026-07-29 금월도 2고 1차 실측): 선택 판정이 **장면 구성 완전성**을 # 요구해 10그룹 중 3그룹이 "쓸 만한 후보 0" 으로 죽었다. 판정문 실측 — # "거대한 바위와 중앙 제단이 결합된 형태를 보여주는 사진이 없다" / # "동굴 입구와 나무 기둥 지지대가 함께 온전히 포함된 레퍼런스 없음" / # "순찰정의 전체적인 종단 배치를 종합적으로 파악할 수 있는 이미지 없음". # 장소는 이 작품이 지어낸 것이라 그 구성을 찍은 사진은 웹에 존재할 수 # 없다 — 형태 참조의 목적은 **그 종류 시설 하나**의 구조·재질·비례다. # pick_system 에 ①서술의 요소가 한 장에 다 나올 필요 없음 ②나머지 요소 # 부재·배치 차이는 탈락 사유 아님 ③완전 일치보다 **읽히는 예시** 우선 # ④chosen_index=0 은 "주 시설 종류가 아예 없을 때"로 한정, 을 명문화. # brief_system·form_only_clause 는 v1 과 동일(전체 사본). "2": "2.202607291130", # v3 (2026-07-29 사용자 지적): 검색 대상 집합이 lane plan # `structure_plate` 바인딩이었는데 **그건 구조물의 성질이 아니라 샷의 # 필요성**이다 — 그 그룹의 선택 샷 중 인물 뒤로 구조가 보존돼야 하는 샷이 # 있느냐를 뜻한다. 그래서 편의점·파출소·주유소·도축장·식당 데크·목조 # 창고·스피커탑이 조용히 순수 T2I 로 남았다(실측 7건). 사용자 원문은 # "건물이나 규모가 있는 모든 복잡한 것들은 **모두** 검색 기반"이다. # → v3 팩은 `scope_system`(LLM 대상 판정)을 담았으나 **2026-07-30 에 # 폐기**했다: 판정 입력에 작품 고유명사가 실렸고 validator 가 근거 없는 # NO 를 통과시켜 "애매하면 yes" 가 강제되지 않았다(Codex 리뷰 6a). # 현행 대상 = seed exact parity(판정 없음, 스텝 docstring 참조). # 파일은 관례상 남겨두되 **로드하지 않는다**. 나머지 4파일은 v2 와 동일. "3": "3.202607291430", # v4 (2026-07-30 주유소 1그룹 실측): **이중 판정의 점수 척도가 심판마다 # 달랐다.** pick_system 어디에도 척도 지시가 없어 Gemini 는 0~100 # (100/90/85…)을, GPT 는 사실상 0~10(10/9/8…)을 썼다. 두 점수를 그대로 # 평균하면 Gemini 가 사실상 단독으로 결정한다 — 사용자가 정한 "둘 다 # 보고 평균"이 성립하지 않았다. → pick_system 에 0~100 구간별 기준과 # "다른 심판과 평균되니 네 마음대로 척도를 만들지 말라"를 명문화하고, # 스키마 `score` 필드 description 에도 같은 척도를 박았다. # 나머지 3파일(brief_system·form_only_clause·narrow_retry_hint)은 v3 와 # 바이트 동일. "4": "4.202607301950", # v5 (2026-08-01): 어휘 중립화. 장소를 알리는 표시물의 **통칭 자체**를 # 계약에서 뺀다 — 일반화된 표현은 "장소를 알리는 활자나 표시가 들어가는 # 면" 이고, 그런 면이 있는지·읽힐 크기인지는 VLM 이 판정한다. # ★발행 이유가 어휘만이 아니다: 같은 날 form_only_clause 가 **전 롤에** # 주입되도록 배선이 바뀌면서(BLOCKING 2 수정) 이 절의 도달 범위가 넓어졌다. # 금지 어휘를 그대로 둔 채 도달 범위만 넓히면 규칙 위반이 확대된다. # 요구 자체는 v4 와 같다(그 지역의 표시 관례·서체를 따른다). "5": "5.202608011353", # v6 (2026-08-03 사용자 육안 반려 4건): 관할절이 두 가지를 놓쳤다. # ①자연 지형에도 "그 지역의 표시 관례를 따르라"가 무조건 붙어 이미지 # 모델이 없는 팻말을 만들어 냈다(해안 절벽 실측) → 활자·표식은 사람이 # 만든 것에만 있고 자연물에는 아무것도 붙이지 않는다를 명시. # ②"관례·배치·서체"만 요구해 표식의 **개수·구성·색 분할·부속 표시**가 # 전이되지 않았다. 참조 사진에 그 체계가 다 보이는데 산출은 빈 밴드에 # 글자만이었다(편의점·주유소 실측). "6": "6.202608030348", # v7 (2026-08-03): 관할절도 같은 범주로 — 사람이 표면에 얹은 시각물 # 일체(글자·표장·인쇄물·칠·낙서·색면·판). 개념이 저작과 어긋나면 # 한쪽만 넓어진다. "7": "7.202608030403", # v8 (2026-08-03 사용자 확정 흐름): 두 경로를 하나씩 만들어 결과로 # 고르기 위해 관할절이 두 벌이 된다. # form_only_clause — 사진 한 장이 형태와 표면을 다 맡는 경로 # sketch_transform — 형태만 남긴 선 도해를 그리는 지시 # form_only_clause_sketched — 형태=도해 / 표면=사진 으로 쪼갠 경로 # ★쪼갠 근거는 실측이다. 나쁜 참조는 구조를 그대로 전이시키는데(로우앵글 # 사선 슬래브가 계단을 망친 반려), 선만 남기면 층수·개구부가 또렷해져 # 같은 브리프에서 3층에 머물던 것이 4층으로 나왔다. 다만 그 실측은 # 한 그룹 한 번이라 **어느 경로가 나은지는 미검증**이고, 그래서 미리 # 고르지 않고 둘 다 그려 참조 사진 앞에서 고르게 한다. # ★도해에도 "사람이 얹은 것"의 빈 외곽은 남긴다 — 자리를 비워 두어야 # 최종 생성에서 사진의 체계가 그 자리에 들어온다(글자만 금지). "8": "8.202608031024", # v9 (2026-08-03): 실험에서 통한 다섯 가지를 production 으로 옮긴다. # ①brief_system 이 **사전조사 산출**을 받아 그것이 사진에 나오도록 # 질의를 쓴다 — 조사와 이미지 검색이 서로 모른 채 따로 돌던 고리를 # 잇는다(실측: 조사가 답을 냈는데 그 답이 질의로 이어지지 않았다). # ②brief_system 이 **언어 잠금 문장을 원어로 따로 저작**한다. 질의를 # 원어로 줘도 검색 모델이 다시 쓰면서 영어를 덧붙였다(실측) — # 잠금은 지시문 **맨 앞**에 놓아야 묻히지 않는다. # ③`focus_query_system` (신규) — 대상 하나의 조사 결과를 근거로 그 # 하나만을 위한 이미지 검색 지시문·질의를 원어로 쓴다. # ④`fitting_pick_system` (신규) — 그 대상이 가장 잘 보이는 사진 1장. # ⑤`fitting_ref_clause` (신규) — 참조가 여러 장일 때 무엇이 무엇을 # 정하는지 나눈다. 나누지 않으면 서로의 배치·구도가 섞여 들어온다. # 나머지 파일은 v8 과 바이트 동일. "9": "9.202608031301", # v10 (2026-08-03 사용자 지시 "주 질의도 좁혀"): 주 구조물 검색이 명세 # 항목을 **전부 나열해 뭉쳐** 나가고 있었다(실측: "경찰서 외부 주차장 # 차량 진입로 청사 출입구", "주유소 진입로 주유 공간 주유기 고정식 # 표지판"). 부속은 v9 에서 각자 따로 검색하게 됐으므로 주 질의가 그것들을 # 안고 갈 이유가 사라졌고, 안고 가면 하나뿐인 질의를 그것들이 차지한다. # ★좁힘 문안은 이 팩에 **이미 있었다** — `narrow_retry_hint` 가 그것이다. # 다만 "1차 실패 뒤"에만 걸려 있었다. 기본값으로 올린다. # ★그러면 재시도 문안이 기본값과 같은 말이 되어, 실패할 때마다 같은 # 질의를 돈 내고 두 번 던지게 된다. 그래서 재시도의 몫을 **수식어 # 버리기**로 바꿨다(지역명·시대·재질·색·상태를 질의에서 뺀다. 그것들은 # 결과를 받아들일 때의 기준으로만 남는다). # ★함께 (같은 날 육안 반려 — 해안 절벽에 관람 난간과 안내판이 들어갔다): # `pick_system` 의 관광 배제 조항이 **"이 장소가 관광지인가"**를 묻고 # 있어서, 평범한 대상에 나중에 붙인 방문객 시설(난간·전망 데크·안내판· # 놓인 길)은 그대로 통과했다. 그 사진이 주 참조가 되자 그림이 난간을 # 베꼈고, 판정이 "참조처럼 갖춰진 쪽"을 골라 결함을 키웠다. # → 선택 계약에 "대상 자체는 평범해도 **사람이 와서 보라고 지어 붙인 # 것**이 있으면 탈락"을 더하고, 관할절 두 벌에 "그것은 형태가 아니다 — # 넘어오지 않는다"를 더했다. 자연물 조항이 글자·표식만 막고 있어서 # 난간처럼 글자 없는 구조물은 걸리지 않았다. "10": "10.202608031530", } # 검색이 컨텍스트로 돌려줄 사진 수 / 판정에 태울 최대 후보 수 SEARCH_IMAGE_RESULTS = 6 MAX_PICK_CANDIDATES = 12 # 검색·판정을 부리는 오케스트레이터 (이미지 툴은 쓰지 않는다 — 검색 전용) SEARCH_ORCHESTRATOR = "gpt-5.6" _UA = "TheRoad-SceneLab/1.0" def search_contract_sha(version: str = REF_PACK_VERSION) -> str: """검색·선택 **산출을 지배하는** 팩 파일들의 바이트 해시. 대상 집합 정책 파일(`scope_system` — v3 에서 폐기)은 여기 넣지 않는다. 그것은 "어느 그룹이 검색을 타는가"만 정하고 한 그룹의 검색·선택 결과에는 기여하지 않는다. 그래서 정책 변경이 이미 뽑아 둔 참조를 무효화하지 않는다 (그룹 단위 재사용 근거). """ import hashlib from app.modules.prompt_loader import load_prompt v = resolve_ref_pack_version(version) h = hashlib.sha256() for name in ("brief_system", "pick_system", "narrow_retry_hint", # v9 — 부속 대상의 질의 저작·선정도 이 스텝의 검색 산출을 # 지배한다. 빼 두면 그 계약을 고쳐도 완료 CP 가 옛 결과를 # 그대로 재사용한다. "focus_query_system", "fitting_pick_system"): try: h.update(load_prompt(REF_PACK_MODULE, name, version=v) .encode("utf-8")) except FileNotFoundError: # 구 팩에 없는 파일만 빈 값으로 흡수한다. 그 외 예외(loader 손상· # 권한·DB)는 올린다 — 전부 삼키면 서로 다른 고장이 같은 해시가 돼 # 재실행 판단을 조용히 망친다(Codex 리뷰 6b). h.update(b"") h.update(b"\x00") return h.hexdigest()[:16] def resolve_ref_pack_version(version: str = REF_PACK_VERSION) -> str: """팩 버전 해석 — 미배선 팩 발행 방지를 위해 명시 실패시킨다.""" v = str(version or REF_PACK_VERSION) if v in REF_PACK_VERSION_MAP: return REF_PACK_VERSION_MAP[v] if v in REF_PACK_VERSION_MAP.values(): return v raise ValueError(f"search_grounded_ref 팩 버전 없음: {version}") # ───────────────────────────────────────────────────────────────────── # 1. 검색 지시문 저작 (시나리오 원본어) # ───────────────────────────────────────────────────────────────────── def build_search_brief_schema() -> Dict[str, Any]: return { "type": "object", "properties": { "source_language": { "type": "string", "minLength": 2, "description": "The language the source text is written in.", }, "search_directive_native": { "type": "string", "minLength": 40, "description": ( "The whole search instruction, written entirely in " "the source language. It is handed to the searcher " "as their only briefing, so it must also tell them " "to gather photographs and nothing else, to write " "every query in this same language, and never to " "generate an image."), }, "search_terms_native": { "type": "array", "minItems": 3, "maxItems": 6, "items": {"type": "string", "minLength": 2}, "description": "Short everyday search terms, same language.", }, # ★[v9] 언어 잠금은 **따로 받아 지시문 맨 앞에** 놓는다. 질의를 # 원어로 줘도 검색 모델이 질의를 다시 쓰면서 영어를 덧붙였다 # (실측). 본문 뒤쪽에 적으면 지시가 길어질수록 묻힌다. 문장 자체가 # 원어 산출이라 코드에는 어느 언어도 남지 않는다. "language_lock_native": { "type": "string", "minLength": 10, "description": ( "A standing order, in the source language, forbidding " "the searcher to compose, translate or append a query " "in any other language."), }, }, "required": ["source_language", "search_directive_native", "search_terms_native", "language_lock_native"], "additionalProperties": False, } # 씬 원문 중 `source_language` 판정용으로만 전달할 앞부분 길이. # 언어는 도입부 몇 줄이면 확정되므로 여유를 둔 값이다 — 이 블록은 언어 외 # 어떤 것도 검색 지시문에 기여하면 안 된다(같은 지시문이 명시적으로 금지). _SOURCE_LANGUAGE_SAMPLE_CHARS = 1200 def build_search_brief_user( *, structure_desc: str, world_facts_block: str, source_text: str, researched_facts_block: str = "", ) -> str: """저작 입력 조립. ★ `source_text` 는 **여기서만** 앞부분 발췌로 전달한다 (`_SOURCE_LANGUAGE_SAMPLE_CHARS`). 프로젝트 절대 규칙은 "LLM 에 전달하는 시나리오 텍스트를 자르지 마라" 이지만, 사용자가 2026-08-03 에 조건부 예외를 열었다 — *"단순한 정보를 얻기 위해서 전체를 넣는 것보다 부분만 넣어도 되는 경우에는 예외로 하자(단 무조건 전체가 아니어도 되는 경우만)"*. 이 블록은 그 조건에 정확히 해당한다: - 이 블록의 용도가 **`source_language` 판정 하나뿐**이고(아래 지시문이 "THIS BLOCK OWNS `source_language` AND NOTHING ELSE" 로 못 박는다), 언어는 도입부 몇 줄로 확정된다. - 전체를 싣는 것이 오히려 **위험을 키운다** — 같은 지시문이 인물·사건· 임시 사물·레이아웃·작품 고유명사가 검색 지시문으로 새어 나가는 것을 금지하는데, 원문이 길수록 그 누출 표면이 커진다(실측: 검색 대상이 타지역으로 샌 사례가 이 모듈 docstring 1번에 기록돼 있다). - 실측 기준 이 블록은 콜당 65,758자였다(전체 68,364자 중 96%). 다른 경로의 시나리오 전달은 이 예외의 대상이 아니다 — 분석·추출처럼 원문 전체가 판단 근거인 곳은 그대로 전문을 넘긴다. ★[v9] `researched_facts_block` = 이 그룹의 사전조사 산출. 조사와 이미지 검색은 그동안 서로 모른 채 따로 돌았다 — 조사가 답을 내도 그 답이 질의로 이어지지 않았다(실측). 조사를 먼저 돌리는 이유가 **무엇을 사진으로 찾을지 정하는 것**이므로, 그 산출이 질의 저작의 입력이어야 한다. """ if not (structure_desc or "").strip(): raise ValueError("structure_desc 결손 — 검색 대상 서술 필수") # ★관할 분리: 검색 **의미**는 place spec + world facts + 사전조사에서만 # 나온다. 씬 원문은 `source_language` 판정 전용으로 격리한다 — 격리하지 # 않으면 인물·사건·임시 사물·작품 고유명사가 검색 지시문으로 새어 나간다. facts = (researched_facts_block or "").strip() sample = (source_text or "")[:_SOURCE_LANGUAGE_SAMPLE_CHARS] return ( "SEARCH SEMANTICS COME FROM THE BLOCKS ABOVE THE LINE ONLY.\n\n" "REGION AND ERA (creator-confirmed):\n" + (world_facts_block or "") + "\n\nTHE STRUCTURE that must be built:\n" + structure_desc + (("\n\n" + facts) if facts else "") + "\n\n---\nSOURCE TEXT SAMPLE (opening excerpt) — THIS BLOCK OWNS " "`source_language` AND NOTHING ELSE. It must not contribute people, " "events, temporary objects, layout, or any story-specific name to " "the search instruction you write. Read it only to decide which " "language to write in:\n" + sample ) # ───────────────────────────────────────────────────────────────────── # 2. 웹 이미지 검색 (사진 결과) # ───────────────────────────────────────────────────────────────────── def build_web_search_tool(max_results: int = SEARCH_IMAGE_RESULTS ) -> Dict[str, Any]: """★ 사진 결과를 받는 검색 툴 스펙. `search_content_types` 를 켜지 않으면 텍스트 결과만 돌아온다 — 모듈 docstring 1번 실측. """ return { "type": "web_search", "search_content_types": ["image", "text"], "image_settings": {"max_results": int(max_results), "caption": True}, } # ★검색에 나가는 프롬프트는 무조건 원본어다 (2026-08-03 사용자 재지시). # 이전에는 이 영어 system 이 briefing 이고 원어 지시문은 user 로만 갔다 — # 모델이 받는 지시의 절반이 영어면 영어권 전형이 섞여 돌아온다. 이제 # 원어 지시문 자체가 briefing 이고, 그 안에 "사진만·이 언어로·생성 금지" # 까지 저작이 원어로 적는다(위 스키마 description). def search_reference_images( client: Any, *, directive_native: str, terms_native: Optional[List[str]] = None, language_lock_native: str = "", max_results: int = SEARCH_IMAGE_RESULTS, model: str = SEARCH_ORCHESTRATOR, ) -> Dict[str, Any]: """원본어 지시문으로 검색해 **사진 결과**를 회수한다. ★[v9] `language_lock_native` 는 지시문 **맨 앞**에 놓는다. 검색 모델은 준 질의를 그대로 쓰지 않고 다시 쓰면서 영어 질의를 덧붙였다(실측: 원어 2개 옆에 "convenience store exterior fixed bench …"). 잠금을 본문 뒤쪽에 적으면 지시가 길어질수록 묻힌다. 잠금 적용 후 나간 질의 2개가 전부 원어였다. 문장 자체가 원어 산출이라 코드에는 어느 언어도 없다. 반환 = {"queries": [...], "images": [{image_url, thumbnail_url, source_website_url, caption}], "said": str} """ if not (directive_native or "").strip(): raise ValueError("directive_native 결손 — 검색 지시문 필수") # 지시는 briefing 으로, 검색어는 입력으로 — 둘 다 원본어다. user_text = (" / ".join(str(t) for t in terms_native) if terms_native else directive_native) lock = (language_lock_native or "").strip() instructions = (lock + "\n\n" + directive_native) if lock \ else directive_native # ★검색 호출도 남긴다 (2026-08-07 사용자 지시). `responses.create` 는 # litellm 을 거치지 않아 Opik 자동 추적 밖이다 — 붙이지 않으면 나간 질의도 # 실패도 어디에도 안 남는다. import time as _time from app.modules.llm.image_tracer import ( ambient_call_meta, record_provider_call, resolve_step_name) _meta = ambient_call_meta() _step = resolve_step_name("search_grounded_ref", _meta) _t0 = _time.monotonic() try: resp = client.responses.create( model=model, instructions=instructions, # ★원어 지시문이 곧 briefing input=[{"role": "user", "content": [{"type": "input_text", "text": user_text}]}], tools=[build_web_search_tool(max_results)], include=["web_search_call.results"], ) except Exception as _exc: record_provider_call( step=_step, model=model, prompt=f"{instructions}\n---\n{user_text}", status="error", duration_ms=int((_time.monotonic() - _t0) * 1000), meta=_meta, operation="web_search", provider="openai", error=str(_exc)[:500]) raise record_provider_call( step=_step, model=model, prompt=f"{instructions}\n---\n{user_text}", status="success", duration_ms=int((_time.monotonic() - _t0) * 1000), meta=_meta, operation="web_search", provider="openai", output_text="[web search completed]") queries: List[Any] = [] images: List[Dict[str, Any]] = [] seen: set = set() said = "" for o in resp.output: if o.type == "web_search_call": act = getattr(o, "action", None) q = getattr(act, "queries", None) or getattr(act, "query", None) if q: queries.append(q) for r in (getattr(o, "results", None) or []): d = r if isinstance(r, dict) else r.model_dump() if d.get("type") != "image_result": continue key = d.get("image_url") or d.get("thumbnail_url") if not key or key in seen: continue seen.add(key) images.append({ "image_url": d.get("image_url"), "thumbnail_url": d.get("thumbnail_url"), "source_website_url": d.get("source_website_url"), "caption": d.get("caption"), }) elif o.type == "message": for c in (getattr(o, "content", None) or []): said += getattr(c, "text", "") or "" # ★나간 질의를 그대로 남기고 로그에도 찍는다. 언어 잠금이 통했는지는 # 이것으로만 안다 — 회수 결과만 보면 위반을 못 본다(운 좋게 그 나라 # 사진이 오기도 한다). logger.info("search_grounded_ref: 검색 %d회 / 회수 사진 %d장 · 나간 질의 %s", len(queries), len(images), json.dumps(queries, ensure_ascii=False)[:400]) return {"queries": queries, "images": images, "said": said.strip(), "language_lock_native": lock} # ───────────────────────────────────────────────────────────────────── # 안전 다운로드 정책 — 검색 URL 은 **외부 입력**이다. # # 이 경계가 없으면 production 에 SSRF(내부망·메타데이터 엔드포인트 취득), # 대용량 다운로드, 위장 파일(SVG/HTML) 경로가 열린다. 정책 버전은 # config_hash 에 편입한다(정책이 바뀌면 재실행이 유도되어야 한다). # ───────────────────────────────────────────────────────────────────── SAFE_DOWNLOAD_POLICY_VERSION = "1" _MAX_DOWNLOAD_BYTES = 12 * 1024 * 1024 # 12MB _MAX_PIXELS = 40_000_000 # decompression bomb 방어 _DOWNLOAD_TIMEOUT = 30 _ALLOWED_CONTENT_PREFIX = "image/" _REJECTED_CONTENT = ("image/svg", "text/", "application/") def _is_public_host(host: str) -> bool: """DNS 해석 결과가 전부 공인 주소일 때만 허용 (redirect 후에도 검사).""" import ipaddress import socket if not host: return False try: infos = socket.getaddrinfo(host, None) except OSError: return False if not infos: return False for info in infos: addr = info[4][0] try: ip = ipaddress.ip_address(addr) except ValueError: return False if (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast or ip.is_unspecified): return False return True def _fetch_safe(url: str) -> Optional[bytes]: """https 전용 + 사설망 차단 + 크기·타입 상한. 위반은 None.""" from urllib.parse import urlsplit parts = urlsplit(url) if parts.scheme != "https" or not parts.hostname: logger.info("search_grounded_ref: https 아님 — 거부 %s", url[:80]) return None if not _is_public_host(parts.hostname): logger.warning("search_grounded_ref: 비공인 호스트 거부 %s", parts.hostname) return None class _NoRedirect(urllib.request.HTTPRedirectHandler): """redirect 를 자동 추종하지 않고 **재검사**해서 따라간다.""" def redirect_request(self, req, fp, code, msg, headers, newurl): p = urlsplit(newurl) if p.scheme != "https" or not _is_public_host(p.hostname or ""): logger.warning( "search_grounded_ref: redirect 대상 거부 %s", newurl[:80]) return None return super().redirect_request( req, fp, code, msg, headers, newurl) opener = urllib.request.build_opener(_NoRedirect) req = urllib.request.Request(url, headers={"User-Agent": _UA}) with opener.open(req, timeout=_DOWNLOAD_TIMEOUT) as resp: ctype = (resp.headers.get("Content-Type") or "").lower() if any(ctype.startswith(bad) for bad in _REJECTED_CONTENT): logger.info("search_grounded_ref: content-type 거부 %s", ctype) return None if ctype and not ctype.startswith(_ALLOWED_CONTENT_PREFIX): logger.info("search_grounded_ref: content-type 거부 %s", ctype) return None raw = resp.read(_MAX_DOWNLOAD_BYTES + 1) if len(raw) > _MAX_DOWNLOAD_BYTES: logger.info("search_grounded_ref: 크기 상한 초과 — 거부") return None return raw def download_candidate(url: str, dest: Path, fallback_url: str = "") -> bool: """후보 사진을 **안전 정책 아래** 받아 PNG 로 재인코딩해 저장한다. 검색이 돌려주는 원본 URL 은 410/403 이 흔하다(실측) — 같은 사진의 썸네일이 함께 오므로 그것으로 떨어뜨리되 **같은 정책**을 적용한다. """ import io from PIL import Image for u in [url, fallback_url]: if not u: continue try: raw = _fetch_safe(str(u)) if raw is None: continue img = Image.open(io.BytesIO(raw)) # 실제 decode 성공 + 픽셀 상한 (위장 파일·bomb 방어) img.verify() img = Image.open(io.BytesIO(raw)) w, h = img.size if w * h > _MAX_PIXELS: logger.info("search_grounded_ref: 픽셀 상한 초과 %dx%d", w, h) continue img = img.convert("RGB") img.thumbnail((1024, 1024)) dest.parent.mkdir(parents=True, exist_ok=True) img.save(dest, format="PNG") # 안전한 재인코딩 return True except Exception as exc: # noqa: BLE001 logger.info("search_grounded_ref: 다운로드 실패 %s — %s", str(u)[:80], exc) return False # ───────────────────────────────────────────────────────────────────── # 3. VLM 비교 선택 (여러 장 중 1장) # ───────────────────────────────────────────────────────────────────── def build_pick_schema(max_items: int = MAX_PICK_CANDIDATES ) -> Dict[str, Any]: return { "type": "object", "properties": { "verdicts": { "type": "array", "maxItems": max_items, "items": { "type": "object", "properties": { "index": {"type": "integer", "minimum": 1}, "usable": {"type": "boolean"}, # 이중 판정 합산용 점수 (2026-07-30 사용자 확정 # "둘다 보고 평균 높은 쪽으로"). 0 = 쓸 수 없음. # usable=false 면 반드시 0 이어야 한다. # ★척도를 필드 설명에도 박는다 — 팩 v3 실측에서 # Gemini 는 0~100 을, GPT 는 사실상 0~10 을 써서 # 평균이 한쪽 심판에 지배됐다(주유소 그룹: 100 vs 10). # 척도 본문은 pick_system 팩 v4 가 소유한다. "score": { "type": "integer", "minimum": 0, "maximum": 100, "description": ( "0-100. 90-100 clearly readable main " "structure; 70-89 usable but harder to " "read; 40-69 weak reference; 1-39 barely " "shows it; 0 unusable. Another judge scores " "the same candidates on this same scale and " "the two are averaged — do not invent your " "own range." ), }, "reason_ko": {"type": "string", "minLength": 2}, }, "required": ["index", "usable", "score", "reason_ko"], "additionalProperties": False, }, }, "chosen_index": { "type": "integer", "minimum": 0, "description": "1-based index of the chosen photograph; " "0 when none is usable.", }, "chosen_reason_ko": {"type": "string", "minLength": 2}, }, "required": ["verdicts", "chosen_index", "chosen_reason_ko"], "additionalProperties": False, } def build_pick_user_head(*, structure_desc: str, world_facts_block: str) -> str: """심판(사진 1장 고르기)에게 줄 머리말. ★``structure_desc`` 에는 **무엇이 있는지**만 넣는다(prompt diet ⑧). 이 콜의 지시문(pick_system)이 스스로 두 번 못박는다 — *"a different arrangement of them, is NEVER a reason to reject a candidate"* / *"JUDGE THE STRUCTURE, NOT ITS SURROUNDINGS"*. 그런데도 seed 저작용 서술(layout narration + 항목별 위치)이 콜마다 실려 있었다. 호출부가 ``structure_names_en`` 을 넘긴다. """ return ("THE STRUCTURE the photograph must show:\n" + structure_desc + "\n\nREGION AND ERA it must belong to:\n" + (world_facts_block or "")) # ───────────────────────────────────────────────────────────────────── # 4. 형태 전용 계약절 (씨드·콘티 프롬프트에 덧붙인다) # ───────────────────────────────────────────────────────────────────── def build_form_only_clause(*, prompt_version: str = REF_PACK_VERSION) -> str: from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "form_only_clause", version=resolve_ref_pack_version(prompt_version) ).strip() def build_sketch_transform_clause( *, prompt_version: str = REF_PACK_VERSION ) -> str: """형태만 남긴 선 도해를 그리게 하는 지시 (v8~). 뒤에 브리프를 이어 붙여 쓴다. 참조 사진 1장을 함께 물린다 — 사용자 확정 흐름의 "자연에서 나온 것이 아니면 스케치부터, 앞 이미지 참조". """ from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "sketch_transform", version=resolve_ref_pack_version(prompt_version) ).strip() def build_form_only_clause_sketched( *, prompt_version: str = REF_PACK_VERSION ) -> str: """형태=도해 / 표면=사진 으로 관할을 쪼갠 절 (v8~). 참조 2장(도해·사진)이 붙는 롤에만 쓴다. 한 장짜리 경로는 `build_form_only_clause` 그대로다. """ from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "form_only_clause_sketched", version=resolve_ref_pack_version(prompt_version) ).strip() def load_ref_pick_system( *, label_list: str, prompt_version: str = REF_PACK_VERSION ) -> str: """참조 사진 대비 **상대 비교** 판정 계약 (v8~). ★절대 문턱을 두지 않는다. 같은 참조·같은 산출로 문턱만 바꿔 두 번 시험 했더니 엄한 쪽은 회차가 갈수록 미세 결함까지 잡다 매체 자체를 잃었고 (3회차에 사진이 선화로), 느슨한 쪽은 이미 육안 반려된 원본을 통과시켰다. "둘 중 어느 쪽"만 물으니 문턱이 필요 없어지고 개선일 때만 교체돼 뒤로 가지 않았다(6그룹 중 4그룹 1회 개선, 2회차 전부 자동 정지). """ from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "ref_pick_system", label_list=label_list, version=resolve_ref_pack_version(prompt_version) ).strip() def build_ref_pick_schema(labels: Sequence[str]) -> Dict[str, Any]: """판정 출력 — 승자와 사유 한 줄. 점수·통과 여부를 두지 않는다.""" return { "type": "object", "properties": { "winner": {"type": "string", "enum": list(labels)}, "reason_ko": {"type": "string"}, }, "required": ["winner", "reason_ko"], } def load_brief_system(*, prompt_version: str = REF_PACK_VERSION) -> str: from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "brief_system", version=resolve_ref_pack_version(prompt_version) ).strip() def load_narrow_retry_hint(*, prompt_version: str = REF_PACK_VERSION) -> str: """[v2] 선택 0 일 때 **주 구조물 하나로 좁혀** 재검색하는 힌트. 실측(금월도 2고): `structure_desc` 가 복합 서술이라 검색어가 하위 요소로 분산돼 엉뚱한 주제를 끌어왔다 — "동굴 입구 + 나무 기둥 + 목재 받침" 이 고고학 유구를, "염소 우리 + 가축 출입문" 이 소 외양간·농장 대문을 회수. 판정을 느슨하게 푸는 게 아니라 **질의를 좁히는 것**이 옳은 처방이다. """ from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "narrow_retry_hint", version=resolve_ref_pack_version(prompt_version) ).strip() def load_pick_system(*, prompt_version: str = REF_PACK_VERSION) -> str: from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "pick_system", version=resolve_ref_pack_version(prompt_version) ).strip() # ───────────────────────────────────────────────────────────────────── # 4.4 부속 대상 — 조사 결과를 근거로 그 하나만 따로 검색한다 (v9) # # ★왜 따로 검색하는가. 주 구조물 질의에 부속을 섞으면 회수는 지배적인 쪽, # 즉 건물 전경만 온다(실측: 12질의가 전부 "외관 출입문 간판 보도 벤치"로 # 뭉쳐 나갔고 그 물건의 사진은 한 장도 오지 않았다). 대상 하나로 좁혀 # 따로 물으니 회수 6장이 전부 그 물건이었고 그대로 그려졌다. # ───────────────────────────────────────────────────────────────────── def load_focus_query_system(*, prompt_version: str = REF_PACK_VERSION) -> str: """조사 결과 → 그 대상 하나의 이미지 검색 지시문·질의 (원어).""" from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "focus_query_system", version=resolve_ref_pack_version(prompt_version) ).strip() def build_focus_query_schema() -> Dict[str, Any]: """대상 하나짜리 질의 저작 산출. 잠금 문장이 첫 칸이다.""" return { "type": "object", "properties": { "language_lock_native": {"type": "string", "minLength": 10}, "directive_native": {"type": "string", "minLength": 40}, "queries_native": { "type": "array", "minItems": 1, "maxItems": 3, "items": {"type": "string", "minLength": 2}, }, }, "required": ["language_lock_native", "directive_native", "queries_native"], "additionalProperties": False, } def load_fitting_pick_system(*, prompt_version: str = REF_PACK_VERSION) -> str: """회수분 중 그 대상이 가장 잘 보이는 1장 — 판정은 VLM 이 한다.""" from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "fitting_pick_system", version=resolve_ref_pack_version(prompt_version) ).strip() def build_fitting_pick_schema() -> Dict[str, Any]: """승자 하나와 사유 한 줄. 점수·통과 여부를 두지 않는다.""" return { "type": "object", "properties": { "index": {"type": "integer", "minimum": 0}, "why_ko": {"type": "string"}, }, "required": ["index", "why_ko"], } def build_fitting_ref_clause( *, head: int, count: int, prompt_version: str = REF_PACK_VERSION ) -> str: """참조가 여러 장일 때의 관할절 — 앞 `head` 장 뒤의 `count` 장이 부속. ★라벨이 아니라 **순서**로 가리킨다. 그림 모델 경로에 따라 라벨이 실제로 전달되지 않기도 한다(gpt-image 편집 호출은 파일만 보낸다) — 전달되지 않는 것을 근거로 쓴 지시문은 그 경로에서 조용히 헛돈다. 앞장 수를 함께 주는 이유는 경로마다 앞이 한 장(사진)이거나 두 장(도해+사진)이기 때문이다. """ from app.modules.prompt_loader import load_prompt return load_prompt(REF_PACK_MODULE, "fitting_ref_clause", head=int(head), count=int(count), version=resolve_ref_pack_version(prompt_version) ).strip() # ───────────────────────────────────────────────────────────────────── # 4.5 검색 대상 판정 (v3) — "사람이 지은 것이 주 피사체인가" # ───────────────────────────────────────────────────────────────────── # 판정 정책 버전 — 기본값 방향(애매하면 검색)·합집합 계약이 바뀌면 올린다. # v3 (2026-07-30): 대상 = seed exact parity — LLM 판정 없음. # 판정을 없앤 이유는 스텝 모듈 docstring 참조(두 번의 실패 기록). TARGET_POLICY_VERSION = "3-seed-parity" # 배경 검색 후보 선택 = 이중 판정 (사용자 확정 2026-07-30 "VLM 판단은 # Gemini + GPT 판정으로 하자 둘다 보고 평균 높은 쪽으로"). # 순서는 감사 기록의 안정성을 위해 고정한다. PICK_JUDGES: tuple = ("gemini-pro", "gpt") def _validate_judge_verdicts( verdicts: Any, candidate_count: int, ) -> Tuple[Optional[Dict[int, Dict[str, Any]]], str]: """한 심판의 응답을 **원자 단위**로 검증한다. (판정, 사유) 를 돌려준다. [2026-08-01 A6] 이전에는 잘못된 항목만 버리고 나머지를 합산에 넣었다. 그러면 후보마다 심판 수가 달라져 **분모가 다른 평균**을 비교하게 된다 — 어떤 후보는 2명 평균, 어떤 후보는 1명 평균이다. 게다가 index 를 틀리게 낸 심판의 나머지 판정을 신뢰할 근거도 없다. 계약: ``index`` 집합이 **1..N 정확한 순열**이어야 한다. 중복 0, 누락 0, 범위 밖 0, 비정수 0. 하나라도 어긋나면 그 심판을 통째로 제외한다. """ if not isinstance(verdicts, list) or not verdicts: return None, "verdicts 가 비었거나 목록이 아니다" seen: Dict[int, Dict[str, Any]] = {} for item in verdicts: if not isinstance(item, dict): return None, "verdict 항목이 dict 가 아니다" raw = item.get("index") if isinstance(raw, bool) or not isinstance(raw, int): # 문자열 숫자도 받지 않는다 — 스키마가 정수를 요구한다. return None, f"index 가 정수가 아니다: {raw!r}" if not (1 <= raw <= candidate_count): return None, f"index 가 후보 범위 밖이다: {raw}" if raw in seen: return None, f"index 가 중복이다: {raw}" score = 0.0 if not item.get("usable") else float(item.get("score") or 0) seen[raw] = { "score": max(0.0, min(100.0, score)), "usable": bool(item.get("usable")), "reason_ko": item.get("reason_ko"), } missing = sorted(set(range(1, candidate_count + 1)) - set(seen)) if missing: return None, f"빠뜨린 후보가 있다: {missing}" return seen, "" def combine_pick_verdicts( per_judge: Dict[str, Dict[str, Any]], candidate_count: int, ) -> Dict[str, Any]: """심판별 판정을 **후보 index 별 평균 점수**로 합산해 1장을 고른다. 계약: - 각 심판은 **모든 후보**에 0~100 점을 준다. ``usable=false`` 는 0 점으로 강제한다(점수를 후하게 주고 usable 만 내리는 경우를 막는다). - **심판 응답은 원자 단위다** — index 가 1..N 정확한 순열이 아니면 그 심판을 통째로 제외한다. 개별 항목만 버리면 후보마다 분모가 달라진다. - 살아남은 심판은 **모든 후보에 동일한 분모**로 기여한다. - 평균이 가장 높은 후보를 고른다. 동점이면 낮은 index(=검색 상위). - **평균 0 이면 선택 없음**(chosen_index=0) — 쓸 만한 후보가 없다는 뜻. - 심판이 하나만 살아남아도 진행하되 ``single_judge`` 로 드러낸다. 조용한 단독 판정 금지 — 감사에서 이중/단독을 구분할 수 있어야 한다. - **valid 0명이면 fail-closed** — 고르지 않는다(조용히 1번을 집지 않는다). - 제외한 심판과 사유를 ``rejected_judges`` 에 남긴다. 조용한 제외 금지. per_judge = {judge_name: } (죽은 심판은 아예 없는 키) """ valid: Dict[str, Dict[int, Dict[str, Any]]] = {} rejected: Dict[str, str] = {} for judge in sorted(per_judge): parsed, why = _validate_judge_verdicts( (per_judge.get(judge) or {}).get("verdicts"), candidate_count) if parsed is None: rejected[judge] = why logger.warning( "combine_pick_verdicts: 심판 %s 제외 — %s", judge, why) continue valid[judge] = parsed detail: Dict[int, Dict[str, Any]] = {} averages: Dict[int, float] = {} if valid: for idx in range(1, candidate_count + 1): detail[idx] = {j: valid[j][idx] for j in sorted(valid)} averages[idx] = sum( valid[j][idx]["score"] for j in valid) / len(valid) chosen = 0 if averages: best = max(averages.values()) if best > 0: chosen = min(i for i, a in averages.items() if a == best) return { "chosen_index": chosen, "judges_used": sorted(valid), "rejected_judges": rejected, "judge_count_per_candidate": len(valid), "single_judge": len(valid) == 1, "averages": {str(i): round(a, 2) for i, a in sorted(averages.items())}, "per_candidate": {str(i): detail[i] for i in sorted(detail)}, } # ───────────────────────────────────────────────────────────────────── # 5. 감사 기록 조립 (스텝이 체크포인트에 싣는다) # ───────────────────────────────────────────────────────────────────── def build_audit_record( *, target_key: str, brief: Dict[str, Any], search: Dict[str, Any], candidates: List[Dict[str, Any]], verdict: Dict[str, Any], chosen: Optional[Dict[str, Any]], ) -> Dict[str, Any]: """판정 원문을 **요약하지 않고** 그대로 남긴다 (실패 귀속의 근거).""" return { "target_key": target_key, "source_language": brief.get("source_language"), "search_directive_native": brief.get("search_directive_native"), "search_terms_native": brief.get("search_terms_native"), "queries": search.get("queries"), "found": len(search.get("images") or []), "candidates": candidates, "verdict": verdict, "chosen": chosen, "pack_version": resolve_ref_pack_version(), } def summarize_for_log(rec: Dict[str, Any]) -> str: ch = (rec.get("chosen") or {}).get("index") return (f"{rec.get('target_key')}: 언어={rec.get('source_language')} " f"회수={rec.get('found')} 후보={len(rec.get('candidates') or [])} " f"선택={ch or '없음'}")