"""★거친 종류 선택 — **VLM 은 「우리가 찾던 종류가 맞나」만 본다.** 사용자 확정(2026-08-30): 「우리가 찾던 게 맞냐는 해당 오브젝트 종류만 보는 것뿐이야 상세히는 인간도 몰라 전문가가 아니면. 자동차인지, 화폐인지 등등 만」 그래서 이 자리는 옛 `search_grounded_ref` 의 pick 과 **다르다**: ============ ================================ ============================== 옛 pick (야외 구조물) 여기 (모든 엔티티 공용) ============ ================================ ============================== 묻는 것 0~100 품질 점수 · 시대/국적 판정 `object_type_match` · `visible` · 평범함 · 관광지/개조 여부 고르는 법 심판 평균 점수 최고 **eligibility 합의 → 결정적 index** ============ ================================ ============================== ★평균을 안 쓰는 이유는 실측이다 — 팩 v3 에서 Gemini 는 0~100 을, GPT 는 사실상 0~10 을 써서 평균이 한쪽 심판에 지배됐다(주유소 그룹 100 대 10). ★**VLM 은 상세 정확성을 인증하지 않는다.** 고증이 맞는지·나아졌는지는 §2-5·§2-6 에서 **사람**이 본다. """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional, Sequence logger = logging.getLogger(__name__) COARSE_PICK_CONTRACT_VERSION = 2 PROMPT_MODULE = "coarse_type_pick" #: ★v2 (2026-09-03 · 사용자 5단계): 심판이 CRITERIA(조사가 낸 「어찌 생겼는가」) 일치와 #: **닮은 정도(similarity 0~100)** 를 같이 낸다 — 마지막엔 반드시 한 장을 고르기 위해서다. PROMPT_PACK_VERSION = "2.202609030300" CRITERIA_MATCH = ("yes", "no", "unsure") SYSTEM_STEM = "system" SCHEMA_STEM = "pick_schema" #: 종류가 맞는지 — 3값. `unsure` 는 **고르지 않는다**(맞다고 못 박을 근거가 없다). TYPE_MATCH = ("yes", "no", "unsure") def load_pack(*, db=None, version: Optional[str] = None) -> Dict[str, Any]: """팩을 통째로 읽는다. stem 하나라도 빠지면 선다.""" from app.modules.prompt_loader import resolve_effective ver = version or PROMPT_PACK_VERSION stems = { SYSTEM_STEM: resolve_effective(PROMPT_MODULE, SYSTEM_STEM, kind="prompt", version=ver), SCHEMA_STEM: resolve_effective(PROMPT_MODULE, SCHEMA_STEM, kind="schema", version=ver), } missing = [k for k, v in stems.items() if not v or not v.get("content")] if missing: raise ValueError(f"coarse_type_pick 팩 {ver} 에 stem 이 없다: {missing}") return {"version": ver, "stems": stems} def _parse_one(payload: Any, candidate_count: int ) -> tuple[Optional[Dict[int, Dict[str, Any]]], str]: """심판 하나의 답. ★**원자 단위**다 — index 가 1..N 순열이 아니면 통째로 뺀다. 개별 항목만 버리면 후보마다 분모가 달라지고, 그러면 「몇 명이 봤나」가 후보마다 다른데 같은 규칙으로 합치게 된다. """ rows = (payload or {}).get("verdicts") if not isinstance(rows, list): return None, "verdicts 가 목록이 아니다" seen: Dict[int, Dict[str, Any]] = {} for r in rows: if not isinstance(r, dict): return None, "verdict 가 dict 가 아니다" idx = r.get("index") if not isinstance(idx, int) or isinstance(idx, bool): return None, f"index 가 정수가 아니다: {idx!r}" if not 1 <= idx <= candidate_count: return None, f"index 가 범위 밖이다: {idx}" if idx in seen: return None, f"index 가 겹친다: {idx}" m = r.get("object_type_match") if m not in TYPE_MATCH: return None, f"object_type_match 가 enum 밖이다: {m!r}" v = r.get("visible") if not isinstance(v, bool): return None, f"visible 이 bool 이 아니다: {v!r}" one: Dict[str, Any] = {"object_type_match": m, "visible": v} # ★v2 칸 — 없으면(v1 심판·옛 판정) 「모른다」로 접는다. 있는데 틀리면 그 심판을 뺀다. if "criteria_match" in r: c = r.get("criteria_match") if c not in CRITERIA_MATCH: return None, f"criteria_match 가 enum 밖이다: {c!r}" one["criteria_match"] = c if "similarity" in r: sim = r.get("similarity") if not isinstance(sim, int) or isinstance(sim, bool) or not 0 <= sim <= 100: return None, f"similarity 가 0~100 정수가 아니다: {sim!r}" one["similarity"] = sim seen[idx] = one if len(seen) != candidate_count: return None, f"후보 {candidate_count}개 중 {len(seen)}개만 답했다" return seen, "" def combine_coarse_verdicts( per_judge: Dict[str, Dict[str, Any]], candidate_count: int, ) -> Dict[str, Any]: """심판들의 답 → **고른 index 하나**. ★평균을 안 낸다. 계약: - 후보가 **eligible** 하려면 살아남은 **모든** 심판이 `object_type_match == "yes"` **그리고** `visible is True` 라고 해야 한다. ★보수적으로 합의한다 — 한 명이라도 아니라면 안 고른다. `unsure` 는 `yes` 가 아니다. - eligible 이 여럿이면 **가장 낮은 index**(=검색 상위)를 고른다. 결정적이다. - eligible 이 없으면 `chosen_index=0` — **안 고르는 것이 나쁜 것을 고르는 것보다 낫다**. - 심판 응답은 원자 단위. 뺀 심판과 사유를 남긴다(조용한 제외 금지). - valid 심판이 0명이면 **fail-closed** — 조용히 1번을 집지 않는다. """ valid: Dict[str, Dict[int, Dict[str, Any]]] = {} rejected: Dict[str, str] = {} for judge in sorted(per_judge): parsed, why = _parse_one(per_judge.get(judge), candidate_count) if parsed is None: rejected[judge] = why logger.warning("coarse_type_pick: 심판 %s 제외 — %s", judge, why) continue valid[judge] = parsed out: Dict[str, Any] = { "contract_version": COARSE_PICK_CONTRACT_VERSION, "candidate_count": candidate_count, "judges": sorted(valid), "rejected_judges": rejected, # ★단독 판정을 **드러낸다**. 감사에서 이중/단독을 구분할 수 있어야 한다. "single_judge": len(valid) == 1, "eligible": [], "chosen_index": 0, "reason": "", } if not valid: out["reason"] = "판정한 심판이 없다 — fail-closed" return out def _criteria_ok(v: Dict[str, Any]) -> bool: # ★CRITERIA 답이 있으면 `yes` 여야 한다. 없으면(v1) 종류·보임만으로 판단한다. return v.get("criteria_match", "yes") == "yes" eligible = [ i for i in range(1, candidate_count + 1) if all(v[i]["object_type_match"] == "yes" and v[i]["visible"] and _criteria_ok(v[i]) for v in valid.values()) ] out["eligible"] = eligible # ★★closest — 「마지막엔 반드시 한 장」(사용자 2026-09-03). 심판들의 similarity 평균이 가장 # 높은 후보. 종류가 `no` 로 합의된 것은 뒤로 미루되(있는 것 중 고른다), 후보가 있으면 반드시 하나. def _sim(i: int) -> float: vals = [v[i].get("similarity") for v in valid.values() if isinstance(v[i].get("similarity"), int)] return (sum(vals) / len(vals)) if vals else -1.0 def _not_wrong_kind(i: int) -> bool: return not all(v[i]["object_type_match"] == "no" for v in valid.values()) pool = [i for i in range(1, candidate_count + 1) if _not_wrong_kind(i)] \ or list(range(1, candidate_count + 1)) ranked = sorted(pool, key=lambda i: (-_sim(i), i)) out["closest_index"] = ranked[0] if ranked else 0 out["closest_similarity"] = _sim(ranked[0]) if ranked else None if not eligible: out["reason"] = "종류·보임·기준을 다 만족하는 후보가 없다 — 이 라운드에선 안 고른다" return out out["chosen_index"] = eligible[0] out["reason"] = (f"eligible {len(eligible)}개 중 검색 상위(index " f"{eligible[0]})를 고른다 — 순위는 검색이 정한다") return out def eligibility_table(per_judge: Dict[str, Dict[str, Any]], candidate_count: int) -> List[Dict[str, Any]]: """후보별로 **심판이 무엇이라 했는지** 그대로. 감사·사람 판정용.""" rows: List[Dict[str, Any]] = [] parsed = {j: _parse_one(p, candidate_count)[0] for j, p in sorted(per_judge.items())} for i in range(1, candidate_count + 1): rows.append({ "index": i, "by_judge": {j: (v[i] if v else None) for j, v in parsed.items()}, }) return rows # ───────────────────────────────────────────────────────────────────── # 라운드 계약 — 「없으면 질의를 좁혀 **한 번 더**」 (사용자 확정 + Codex) # ───────────────────────────────────────────────────────────────────── #: ★한 라운드에 VLM 에게 보이는 후보 상한. #: #: ★★**기존 `era_research.MAX_CANDIDATES` 를 그대로 쓴다** (Codex 2026-08-30). #: 사용자가 **3~5** 로 확정했고 기존 값 **4** 가 그 안이므로 **새 상수를 만들 #: 이유가 없다.** 처음에 5 로 뒀던 것을 되돌린다 — 같은 뜻의 수가 둘이면 #: 한쪽만 고쳐진다. #: #: ★기존 값의 근거도 코드에 있다: 「후보가 많아도 선택 품질이 늘지 않는 #: 실측 관례」(`era_research.py:82`). #: #: ★야외 상수(`SEARCH_IMAGE_RESULTS=6`·`MAX_PICK_CANDIDATES=12`)와는 여전히 #: 다르다 — 그건 구조물 판정용 계약이다. def _per_round_cap() -> int: from app.modules.pipeline.era_research import MAX_CANDIDATES return MAX_CANDIDATES PER_ROUND_CAP = _per_round_cap() #: 두 라운드를 합쳐 **장부에** 남기는 상한. 판정 모집단이 아니라 감사 범위다. TOTAL_AUDIT_CAP = PER_ROUND_CAP * 2 #: 최초 1회 + 좁힌 재검색 최대 1회. MAX_ROUNDS = 2 #: 다음에 무엇을 할까. NEXT_SELECT = "select" # eligible 이 있다 — 고르고 끝 NEXT_SELECT_CLOSEST = "select_closest" # ★마지막 라운드까지 없다 — 가장 닮은 것을 **반드시** 고른다 NEXT_NARROW_RETRY = "narrow_retry" # ★이름은 옛것(durable enum 호환) — 뜻은 **뼈대만 남겨 넓혀** 한 번 더 (2026-09-03) NEXT_NO_MATCH = "no_match" # 두 라운드 다 돌았는데 없다 — **terminal** NEXT_RETRYABLE = "retryable" # provider·시간·예산 — 「없다」가 **아니다** #: 후보 하나의 처지. DISP_JUDGED = "judged" DISP_DUPLICATE = "duplicate" # ★버리지 않고 **기록한다** DISP_OVER_CAP = "over_cap" def dedupe_candidates(rows: Sequence[Dict[str, Any]], seen_urls: Sequence[str] = (), seen_hashes: Sequence[str] = (), *, cap: int = PER_ROUND_CAP) -> Dict[str, Any]: """라운드 후보를 **판정할 것과 아닌 것**으로 가른다. ★중복은 **버리지 않는다** — `duplicate` 처지로 장부에 남긴다. 버리면 「몇 장을 받았나」와 「몇 장을 봤나」가 갈라지는데 그 차이가 안 보인다. ★상한을 넘은 것도 `over_cap` 으로 남긴다. 조용히 자르지 않는다. """ url_seen = {str(u) for u in seen_urls if u} hash_seen = {str(h) for h in seen_hashes if h} judged: List[Dict[str, Any]] = [] ledger: List[Dict[str, Any]] = [] for r in rows: url, h = str(r.get("url") or ""), str(r.get("content_hash") or "") if (url and url in url_seen) or (h and h in hash_seen): ledger.append({**r, "disposition": DISP_DUPLICATE}) continue if len(judged) >= cap: ledger.append({**r, "disposition": DISP_OVER_CAP}) continue if url: url_seen.add(url) if h: hash_seen.add(h) judged.append(r) ledger.append({**r, "disposition": DISP_JUDGED}) return {"judged": judged, "ledger": ledger, "judged_count": len(judged), "duplicate_count": sum(1 for x in ledger if x["disposition"] == DISP_DUPLICATE), "over_cap_count": sum(1 for x in ledger if x["disposition"] == DISP_OVER_CAP)} def decide_next_round(round_no: int, combined: Dict[str, Any], *, new_candidate_count: int, limit_kind: str = "", error: str = "", total_candidate_count: Optional[int] = None) -> Dict[str, Any]: """이 라운드 뒤에 무엇을 할까. ★**「없다」와 「못 봤다」를 가른다.** - eligible 이 하나라도 있으면 **즉시 고르고 재검색하지 않는다** - provider 오류·시간/예산 중단은 **`선택 0` 이 아니다** — `retryable` 로 남긴다. 그걸 narrow retry 로 바꾸면 「다 보고 없었다」가 거짓이 된다 - 1라운드에 없으면 **질의를 좁혀 한 번 더** - 2라운드에 **새 고유 후보가 0** 이면 그 자리에서 `no_match` - 2라운드까지 없으면 `no_match` — **terminal** """ total = int(total_candidate_count if total_candidate_count is not None else new_candidate_count) no_judge = int(combined.get("candidate_count") or 0) > 0 and combined.get("judges") == [] if error or limit_kind or no_judge: # ★★Codex BLOCK 3 (2026-09-03): 판정·검색 실패가 「반드시 한 장」을 깨면 안 된다. # 마지막 라운드가 아니면 실패를 **기록하고 넓혀 한 번 더**. 마지막 라운드면 받아 둔 후보가 # 하나라도 있을 때 가장 닮은(없으면 검색 순위 첫) 한 장을 강제로 고른다. 두 라운드 다 # 받은 것이 0 이고 실패였으면 **못 봤다**(retryable) — 「없다」가 아니다. fail = limit_kind or error or "판정한 심판이 없다" if round_no < MAX_ROUNDS: return {"next": NEXT_NARROW_RETRY, "round_no": round_no, "why": f"이 라운드는 못 봤다({fail}) — 실패를 적고 넓혀 한 번 더", "limit_kind": limit_kind, "error": error or ("" if not no_judge else fail)} if total > 0: return {"next": NEXT_SELECT_CLOSEST, "round_no": round_no, "why": (f"마지막 라운드를 못 봤다({fail}) — 받아 둔 {total}장 중 하나를 " "강제로 고른다(판정 없으면 검색 순위 첫 장)"), "forced_reason": "judge_unavailable" if (no_judge or error) else "limit", "limit_kind": limit_kind, "error": error or ("" if not no_judge else fail)} return {"next": NEXT_RETRYABLE, "round_no": round_no, "why": f"다 못 봤다 — {fail} (받은 후보 0)", "limit_kind": limit_kind, "error": error or ("" if not no_judge else fail)} if combined.get("chosen_index"): return {"next": NEXT_SELECT, "round_no": round_no, "why": combined.get("reason") or "종류가 맞는 후보를 골랐다"} # ★더 **구체적인** 사유를 먼저 본다. 둘 다 `no_match` 지만 「좁혀 봤는데 # 새 후보 자체가 없었다」와 「봤는데 종류가 안 맞았다」는 다음에 할 일이 # 다르다 — 앞의 것은 질의를, 뒤의 것은 대상을 다시 봐야 한다. # ★★사용자 5단계 ⑤ (2026-09-03): 마지막 라운드까지 기준에 맞는 것이 없어도 **후보가 있으면** # 가장 닮은 것을 반드시 고른다. `total_candidate_count` 는 모든 라운드에서 받은 후보 수 — # 부르는 쪽이 넘긴다(없으면 이 라운드의 새 후보 수로 본다). if round_no > 1 and new_candidate_count <= 0 and total <= 0: return {"next": NEXT_NO_MATCH, "round_no": round_no, "why": "다시 찾았는데 **새 후보가 없다** — 받은 후보도 없어 고를 것이 없다"} if round_no >= MAX_ROUNDS: if total > 0: return {"next": NEXT_SELECT_CLOSEST, "round_no": round_no, "why": (f"{MAX_ROUNDS}라운드까지 기준에 맞는 후보가 없다 — 받은 {total}장 중 " "가장 닮은 것을 고른다(마지막엔 반드시 한 장)")} return {"next": NEXT_NO_MATCH, "round_no": round_no, "why": f"{MAX_ROUNDS}라운드까지 받은 후보가 없다"} return {"next": NEXT_NARROW_RETRY, "round_no": round_no, "why": "종류가 맞는 후보가 없다 — 질의를 좁혀 한 번 더"} NARROW_STEM = "narrow_retry_hint" def load_narrow_hint(*, db=None, version: Optional[str] = None) -> str: """좁혀 다시 찾을 때 붙이는 문안. ★야외 구조물 말투를 일반화한 새 팩이다. 옛 `search_grounded_ref.load_narrow_retry_hint` 는 「주 **구조물** 하나로」 라 야외 전용이다. 여기 것은 「주 **대상 엔티티** 하나로」이고, **하나**는 물리 개수가 아니라 **하나의 entity subject** 다 — 「시외버스 두 대」를 「한 대」로 바꾸지 않는다. ★시대·지역·제조사·모델 좌표는 **질의에 남긴다.** 좁힌다는 것은 *다른 것들*을 덜어내는 것이지 대상을 못박는 좌표를 버리는 것이 아니다. VLM 이 그 좌표를 **검증하지 않을 뿐**이다. """ from app.modules.prompt_loader import resolve_effective got = resolve_effective(PROMPT_MODULE, NARROW_STEM, kind="prompt", version=version or PROMPT_PACK_VERSION) if not got or not got.get("content"): raise ValueError(f"{PROMPT_MODULE}/{NARROW_STEM} 을 못 읽었다") return got["content"].strip()