"""GROUNDING-V2 §4b — 출처 붙은 claims 를 **실제로 검색해서** 받는다. ★계약 순서: 계약(§75) → 정본 테이블(§77) → **여기**. 유료 호출이 durable 기록보다 먼저 생기면 안 된다. ★`typology_prior` 에서 **저수준 기구만** 가져온다 — 검색 전송·호출 기록· JSON 꺼내기. **프롬프트 의미는 안 섞는다** (Codex). 팩은 새 디렉토리다. ★검색에 나가는 지시문은 **원본어**이고, 언어 잠금이 **맨 앞**에 있다 (2026-08-03 사용자 재지시). 뒤에 적으면 길어질수록 묻힌다. ★`build_web_search_tool(want_images=False)` — **글만** 받는다. 사진은 그 뒤 참조 층이 따로 산다. 여기서 같이 받으면 ①안 쓸 사진 값을 매번 치르고 ②claim ↔ source 결속이 caption 쪽으로 흐려진다. ## batch 는 자르기만 한다 한 호출에 여러 subject 를 넣어도 **subject 마다 판정·claim·출처는 독립**이다. 그것이 지켜지는지가 §4b 실험이 재는 것이고, 그래서 **기본 batch 크기를 여기서 정하지 않는다** — 호출부가 명시한다. 실험이 끝난 뒤 마지막 커밋에서 배선한다. """ from __future__ import annotations import json import logging import time from typing import Any, Dict, List, Optional, Sequence from app.core.research_call_budget import bind_current_research_budget from app.modules.pipeline.grounding_claims import (LIMIT_ADMISSION, TRANSMISSION_NOT_SENT) from app.modules.pipeline.llm_deadline import call_with_deadline from app.modules.pipeline.search_grounded_ref import build_web_search_tool from app.modules.prompt_loader import resolve_effective logger = logging.getLogger(__name__) _MODULE = "grounding_claims_search" #: ★명시 pin. latest 자동 선택에 기대지 않는다 — 같은 module 에 버전을 더하면 #: 옛 호출도 새 prompt 를 읽는 부류를 v2 에서 원천 차단한다. #: ★2 = 대상 결속 계약. 3 = 그 계약을 **자리표시자**로 다시 씀. #: ★★2 는 실패한 판의 명사(장소·탈것·연도)를 그대로 예시에 박았다 — #: 그건 **알려진 실패를 보고 답을 가르친 회귀 튜닝**이지 시나리오 일반 규칙이 #: 아니다 (Codex). 3 은 / 로만 말한다. #: ★작품 고유명사를 프롬프트에 넣지 않는다는 프로젝트 규칙과도 같은 자리다. PROMPT_PACK_VERSION = "3.202608302300" SYSTEM_STEM = "system" SCHEMA_STEM = "claims_schema" STEP_NAME = "grounding_claims_search" #: ★**기본값을 안 둔다.** batch 크기는 §4b 실험이 정한다 — 여기 임시값을 넣으면 #: 그 값이 계약처럼 굳고, 실험 결과와 무관하게 프로덕션이 그것을 쓴다 (Codex). #: 호출부가 명시하지 않으면 선다. BATCH_SIZE_UNSET = None #: ★★§4b 실험의 **계약**. 도구가 아니라 여기 둔다 — 도구에 두면 시험이 #: `tools.prompt_measure` 를 import 해야 하고, 그건 수집 순서에 따라 안 잡힌다 #: (namespace 패키지). 그리고 이 값들은 프로덕션 기본값을 정하는 근거라 #: **모듈의 계약**이 맞다. #: #: 표본 수. ★**9다, 12가 아니다** — A0 승격 뒤 이 에피소드의 원고 기반 subject #: 가 9개이고, 모자란 몫은 facet producer(§2-6.5)를 기다리는 `deferred` 다. #: 표본 수 때문에 그 유료 단계를 앞당기지 않는다 (Codex). EXPERIMENT_SAMPLE_SIZE = 9 #: 잴 batch 크기. ★**12는 없다** — 9개를 한 묶음으로 보낼 뿐이라 12-subject #: 결속·leakage 를 검증하지 못한다. 이 판으로 고를 수 있는 최대 기본값은 **8** #: 이고, 12는 원고 기반 고유 subject 가 12개 생긴 뒤 **별도 재검증**이다. #: 논리 호출 = 9 + 5 + 3 + 2 = **19회**(크기 상한에 안 걸릴 때). EXPERIMENT_BATCH_SIZES = (1, 2, 4, 8) #: ★★**상한이 둘이다** — subject 수와 **직렬화된 prompt 크기** (Codex). #: §4b 실험은 1.3KB 검증 원고로 하므로, 거기서 나온 subject 수를 실제 #: 에피소드에 그대로 쓰면 prompt 가 몇 배가 된다. 원고 길이가 대상마다 다르니 #: **개수만으로는 크기를 못 정한다**. #: ★크기를 넘으면 **원문을 자르지 않는다** — 다음 batch 로 결정적으로 넘긴다. #: 자르는 순간 그 대상은 다른 것을 조사한 것이 된다(절대 규칙). MAX_PROMPT_BYTES = 24_000 #: ★★**soft 와 hard 를 가른다** (Codex). #: soft(`MAX_PROMPT_BYTES`) 를 혼자 넘는 대상은 **singleton 으로 보낸다** — #: 건너뛰면 조용히 조사에서 빠진다. 그 수를 `oversized_subject_count` 로 남긴다. #: hard 를 넘으면 **아예 안 보낸다** — 보내 봐야 provider 가 거절하고, 그 #: 거절을 「조사했는데 못 찾았다」로 읽으면 거짓이 된다. `unresolved` 로 남긴다. HARD_PROMPT_BYTES = 180_000 #: ★★★**요청 계약의 정본은 여기 하나다.** manifest 는 이 값들을 잠그고 #: `load_manifest` 가 지금 값과 하나씩 대조한다. 요청문에 값을 직접 적어 넣고 #: manifest 에 또 적으면 **한쪽만 고쳐진다** — 실제로 그렇게 됐었다(Codex): #: 코드는 `action.sources`·`store=False`·HTTP 확인인데 잠근 것은 #: `results`(beta)·`store=null`·snippet 요구였고, dry-run 은 초록이었다. RESPONSE_INCLUDE = ("web_search_call.action.sources",) #: ★provider **보존 안 함**. `store=True` 는 원고 인용이 외부 보존 대상이 되는 #: 데이터 정책 변경이다 (Codex 거부). 응답은 **로컬에** 통째로 남긴다. PROVIDER_STORE = False #: 어디서 출처를 줍나 — 공식 stable 경로만. `results`(beta)는 안 쓴다. SOURCE_CAPTURE = "action.sources(stable) + url_citation — snippet 은 안 온다" #: ★글만 받는다 — 사진은 참조 층이 **따로** 산다. TEXT_ONLY_SEARCH = True #: ★★★**켠다.** 전에는 「검색 도구와 같이 켜면 provider 가 거절한다」고 적어 #: 뒀는데 그건 **확인 안 된 짐작**이었고, 그 상태로 산 주행에서 claim 16개가 #: **전부** `required` 칸을 빠뜨려 계약에서 거부됐다(2026-08-30 실측). #: ★확인된 것은 「strict=false 에서 16/16 이 그 칸을 안 냈다」뿐이다 — #: 「`required` 가 JSON Schema 예약어라 모델이 헷갈렸다」는 **확정 사실이 #: 아니다**(Codex). 팩은 그 칸을 이미 명시하고 있고, 이 schema 는 모든 #: object 에 `additionalProperties:false` · 모든 칸이 required 라 그대로 #: strict 로 쓸 수 있다. #: ★거절당하면 **false 로 되돌리지 않는다** — 기록하고 선다. SCHEMA_STRICT = True #: ★SDK 재시도 0 — 켜면 물리 전송이 논리 호출 수와 어긋나 「몇 회 샀나」가 틀린다. SDK_RETRIES = 0 #: ★★한 호출의 **벽시계**. `litellm` 의 `timeout` 은 per-read 라 slow-stream #: 을 못 잡는다 — 그 모듈에 「`timeout=180` 이 21분 매달렸다」고 적혀 있다. #: ★검색 호출은 원래 느리다(실측 40~100초). 넉넉히 두되 **무한은 아니다**. PER_CALL_DEADLINE_SECONDS = 300.0 #: 인용이 실제로 그 페이지에 있나를 무엇으로 보나. SUPPORT_POLICY = ("공식 출처 URL 을 우리가 직접 열어(HTTP) 대조한다 — " "provider snippet 에 안 기댄다") #: 더 사기 전 probe 가 요구하는 것. ★둘 다여야 한다. PROBE_REQUIREMENT = "공식 URL 존재 AND 정상 response_dump" def acquisition_contract() -> Dict[str, Any]: """★★★**무엇을 사는지**를 가르는 좌표만. 이게 다르면 **다른 요청**이다. ★그래서 `payload_hash` 에 **접는다**. 접기 전에는 프롬프트·subject 만 같으면 `include`·`store`·글만 여부·`strict` 가 바뀌어도 **옛 응답을 그대로 재사용**했다 — manifest 를 새로 굳혀도 journal 재사용은 manifest 를 안 본다 (Codex). ★모델은 여기 안 넣는다 — 지금처럼 **따로** 대조한다. ★`SUPPORT_POLICY` 처럼 **무료로 다시 채점할 수 있는 후처리 정책**도 안 넣는다. 그건 산 것을 바꾸지 않는다. """ return { "include": list(RESPONSE_INCLUDE), "store": PROVIDER_STORE, "text_only_search": TEXT_ONLY_SEARCH, "schema_strict": SCHEMA_STRICT, } def request_contract() -> Dict[str, Any]: """지금 코드가 실제로 보내는 **요청의 뜻**. manifest 대조용 한 벌.""" return { **acquisition_contract(), # ★아래 셋은 산 것 자체를 안 바꾸거나(정책) 물리 전송만 바꾼다 — # 그래도 **잠그고 대조**한다. 바뀌면 다른 실험이다. "sdk_retries": SDK_RETRIES, "source_capture": SOURCE_CAPTURE, "support_policy": SUPPORT_POLICY, "probe_requirement": PROBE_REQUIREMENT, } def _sha16(text: str) -> str: import hashlib return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16] def load_pack(*, db=None, version: Optional[str] = None) -> Dict[str, Any]: """팩을 통째로 읽고 **completeness 를 확인**한다. stem 하나라도 빠지면 선다 — 반쪽 팩으로 조용히 돌지 않는다. """ 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"{stem}:{r['source']}:{r['version']}:{r['raw_content_hash']}" for stem, r in sorted(resolved.items()))), } def build_user_prompt(subjects: Sequence[Dict[str, Any]], *, era: str, region: str) -> str: """조사할 대상 목록 → user prompt. ★원문 근거를 **자르지 않는다**. ★`research_subject_id` 를 **그대로 적어 준다.** 모델이 그것을 돌려주므로, 돌아온 줄이 어느 대상의 것인지 이름으로 짐작할 필요가 없다 — 이름이 비슷한 두 대상(같은 에피소드의 1호차/2호차 같은)에서 짐작은 반드시 틀린다. """ if not str(era or "").strip() or not str(region or "").strip(): raise ValueError( f"시대/지역이 없다 (era={era!r}, region={region!r}) — " "맥락 없이 조사하면 무엇과 비교할지가 없다") lines: List[str] = [ "## 목표 맥락", f"- 시대: {era}", f"- 지역: {region}", "", "## 조사할 대상", ] for s in subjects: sid = str((s or {}).get("research_subject_id") or "").strip() if not sid: # ★빈 id 를 넣으면 돌아온 줄을 어느 대상에도 못 붙인다. raise ValueError("research_subject_id 가 없는 대상이 있다") lines.append("") lines.append(f"### {sid}") lines.append(f"- 표기: {(s or {}).get('surface_form') or ''}") lines.append(f"- 종류: {(s or {}).get('owner_type') or ''}") lines.append(f"- 원문: {(s or {}).get('source_quote') or ''}") return "\n".join(lines) def _extract_json_object(text: str) -> Optional[Dict[str, Any]]: """응답에서 JSON 객체 하나를 꺼낸다(코드펜스 허용). ★`typology_prior` 의 같은 기구를 **가져다 쓰지 않고 여기 둔다** — 그쪽은 프롬프트 의미까지 얽혀 있어 import 하면 그 의미가 따라온다. """ import re s = (text or "").strip() if not s: return None fence = re.search(r"```(?:json)?\s*(.+?)\s*```", s, re.S) if fence: s = fence.group(1).strip() start, depth = s.find("{"), 0 if start < 0: return None for i in range(start, len(s)): if s[i] == "{": depth += 1 elif s[i] == "}": depth -= 1 if depth == 0: try: obj = json.loads(s[start:i + 1]) except json.JSONDecodeError: return None return obj if isinstance(obj, dict) else None return None def _get(obj: Any, name: str) -> Any: """객체든 dict 든 **같은 방식으로** 읽는다. ★SDK BaseModel 이 `extra="allow"` 라, **선언 안 된 필드는 파싱을 안 거치고 dict 로 그대로** 들어온다. `getattr` 로만 읽으면 조용히 빈 값이 되고, 「포착 0건」이 「API 가 안 준다」로 오독된다 — 유료 1회로 확인했다. """ if isinstance(obj, dict): return obj.get(name) return getattr(obj, name, None) def _collect_sources(resp: Any) -> List[Dict[str, Any]]: """검색이 **실제로 본 것**을 모은다. ★두 모양을 **둘 다** 읽는다. 설치된 stable SDK(`openai 1.109`)의 `ResponseFunctionWebSearch` 는 필드가 `id·action·status·type` 뿐이고 **`results` 가 없다**. `action.sources` 는 있는데 그 원소가 `{type, url}` — **snippet 이 없다** (Codex 실측 확인). ★그런데 SDK BaseModel 이 `extra="allow"` 라 API 가 `results` 를 얹어 주면 그대로 실린다. 그래서 **둘 다 읽고, 무엇으로 읽었는지 남긴다.** ★★**snippet 이 없으면 `evidence_span` 을 확인할 수 없다.** 그건 「문제없음」이 아니라 이 실험이 성립하지 않는다는 뜻이다 — 호출부가 그것을 보고 **더 사기 전에 서야** 한다. 그래서 `has_snippet` 를 같이 낸다. """ out: List[Dict[str, Any]] = [] for o in (_get(resp, "output") or []): if _get(o, "type") != "web_search_call": continue act = _get(o, "action") q = _get(act, "query") or _get(act, "queries") or "" qs = q if isinstance(q, str) else json.dumps(q, ensure_ascii=False) # ① beta/extra 모양. ★★**dict 로 올 수 있다** — SDK BaseModel 이 # `extra="allow"` 라 **선언 안 된 필드는 파싱을 안 거치고 그대로** # 들어온다. `getattr` 로만 읽으면 전부 빈 값이 된다(유료 1회로 확인). for r in (_get(o, "results") or []): out.append({ "shape": "results", "query": qs, "url": str(_get(r, "url") or ""), "title": str(_get(r, "title") or ""), # ★★키 이름을 **추측하지 않는다.** 실제 이름을 모르는 채 # `snippet`/`text`/`content` 를 순서대로 보는 것은 추측이고, # 그걸 넣고 사면 「안 왔다」와 「이름이 달랐다」가 안 갈린다 # (Codex 금지). 모양이 확정될 때까지 `snippet` 하나만 본다 — # 못 읽으면 `capture_shape` 가 `ok=False` 로 세운다. "snippet": str(_get(r, "snippet") or ""), }) # ② stable 모양 — **URL 뿐**이다 for src in (_get(act, "sources") or []): u = str(_get(src, "url") or "") if u: out.append({"shape": "action.sources", "query": qs, "url": u, "title": "", "snippet": ""}) # ★같은 주소가 두 모양으로 오면 **snippet 있는 쪽을 남긴다.** 순서에 따라 # 고르면 같은 응답을 두 번 읽었을 때 결과가 달라진다. best: Dict[str, Dict[str, Any]] = {} for row in out: u = row["url"] cur = best.get(u) if cur is None or (not cur["snippet"] and row["snippet"]): best[u] = row return [r for r in out if best.get(r["url"]) is r] def _url_citations(resp: Any) -> List[str]: """메시지 annotation 의 `url_citation` 주소. ★**소유권 감사에만** 쓴다. 공식 경로로 「모델이 무엇을 인용했다고 표시했나」를 준다. 본문(snippet)은 안 주므로 `evidence_span` 확인에는 못 쓴다 — 그건 별도 축이다 (Codex). """ out: List[str] = [] for o in (_get(resp, "output") or []): for c in (_get(o, "content") or []): for a in (_get(c, "annotations") or []): if _get(a, "type") == "url_citation": u = str(_get(a, "url") or "") if u and u not in out: out.append(u) return out def raw_source_shape(resp: Any) -> List[Dict[str, Any]]: """포착 원본의 **키 이름**을 그대로 남긴다. ★한 번 사면 다 알게. 유료 1회에서 `results` 는 왔는데 `url`·`snippet` 이 전부 비었다. 원인은 `getattr` 이 dict 를 못 읽은 것이었지만, **실제 키 이름이 무엇인지는 그 호출로도 알 수 없었다** — 안 남겼기 때문이다. ★그래서 다음 probe 는 **키와 짧은 값**을 그대로 남긴다. 모양을 모르는 자리에서는 「무엇이 왔는지」를 통째로 남기는 것이 가장 싸다. """ out: List[Dict[str, Any]] = [] for o in (_get(resp, "output") or []): if _get(o, "type") != "web_search_call": continue act = _get(o, "action") for name, rows in (("results", _get(o, "results")), ("action.sources", _get(act, "sources"))): for r in (rows or []): if isinstance(r, dict): keys, sample = sorted(r), r else: keys = sorted(k for k in dir(r) if not k.startswith("_")) sample = {k: getattr(r, k, None) for k in keys[:12]} out.append({ "where": name, "keys": keys[:20], "sample": {k: (str(v)[:120] if not isinstance(v, (dict, list)) else str(v)[:120]) for k, v in list(dict(sample).items())[:12] if not callable(v)}, }) return out def capture_shape(batch: Dict[str, Any]) -> Dict[str, Any]: """이 호출이 **무엇을 포착했나**. ★더 사기 전에 보는 것. ``ok`` 가 거짓이면 `evidence_span` 을 확인할 수 없고, 그러면 이 실험은 성립하지 않는다 — **나머지를 사지 말아야 한다** (Codex). """ src = batch.get("sources") or [] urls = [s for s in src if str(s.get("url") or "").strip()] with_snip = [s for s in src if str(s.get("snippet") or "").strip()] cited = batch.get("citations") or [] shapes = sorted({str(s.get("shape") or "?") for s in src}) # ★★**주소가 오면 된다.** 전에는 snippet 을 요구했는데, 공식 stable API 는 # `action.sources` 로 **URL 만** 준다 — 그러면 probe 가 첫 호출에서 늘 서서 # HTTP 인용 확인까지 **갈 수가 없다** (Codex). # ★인용이 그 페이지에 있는지는 이제 `grounding_claims_support` 가 # **그 URL 을 직접 열어** 본다. provider snippet 에 안 기댄다. have = bool(urls) or bool(cited) # ★★**산 것의 원형이 로컬에 남았나**도 같이 본다 (Codex). 주소만 보고 # 통과시키면, dump 가 비었거나 실패한 채로 진단이 계속 굴러가 **유료 # 산출의 원형을 또 잃는다**. dump 실패는 그 호출을 기록한 채 **선다** — # 다시 사지 않는다. 정상 경로는 늘 dump 를 넣으므로 부담이 없다. dump = batch.get("response_dump") dump_ok = bool(isinstance(dump, dict) and dump and not dump.get("_dump_failed")) ok = have and dump_ok why = "" if not have: why = ("포착된 **주소**가 하나도 없다 — 어느 페이지를 열어 확인할지 " "모른다. 이 실험은 성립하지 않는다") elif not dump_ok: why = ("응답 원형(`response_dump`)이 없거나 실패했다 — 산 것의 원형을 " "잃은 채로는 진단을 이어가지 않는다") return { "sources": len(src), "urls": len(urls), "citations": len(cited), "with_snippet": len(with_snip), "shapes": shapes, "response_dump_ok": dump_ok, "ok": ok, "why": why, # ★snippet 은 이제 **정보**다. 있으면 좋지만 없다고 서지 않는다. "snippet_note": ("공식 stable API 는 본문을 안 준다 — 인용 확인은 " "그 URL 을 직접 열어서 한다"), } def _limit_kind_of(exc: BaseException) -> str: """이 예외가 **이 주행의 제한** 때문인가. 맞으면 그 갈래를 낸다. ★★타입으로 가른다 — 글자로 가르면 provider 오류 문구가 바뀔 때마다 조용히 틀린다. `GAP_REASONS` 는 안 늘리고 `limit_kind` 로만 나눈다 (Codex). """ from app.core.image_call_budget import ImageCallBudgetExceeded from app.core.research_call_budget import ResearchCallBudgetExceeded from app.modules.pipeline.grounding_claims import ( LIMIT_PER_CALL_DEADLINE, LIMIT_RUN_DEADLINE, LIMIT_TRANSMISSION_BUDGET) from app.modules.pipeline.llm_deadline import HardDeadlineExceeded if isinstance(exc, HardDeadlineExceeded): return LIMIT_PER_CALL_DEADLINE if isinstance(exc, (ResearchCallBudgetExceeded, ImageCallBudgetExceeded)): return LIMIT_TRANSMISSION_BUDGET # ★주행 마감은 `install_stop_check` 이 올리는 예외다 — 그 타입은 호출부가 # 정하므로(보통 `CancellationToken`), **표식으로** 가른다. if getattr(exc, "is_run_deadline", False): return LIMIT_RUN_DEADLINE return "" def _is_cancellation(exc: BaseException) -> bool: """사람이 멈춘 것인가. ★우리 상한과 **다르다** — 접지 않고 그대로 올린다. ★코드로 가른다(`step.cancelled`) — 글자로 가르면 문구가 바뀔 때 조용히 틀린다. 타입만 보면 `AppError` 는 다른 것도 많이 쓴다. """ return str(getattr(exc, "code", "")) == "step.cancelled" def _armed_create(client: Any, kwargs: Dict[str, Any]) -> Any: """provider 호출 하나. ★**여기 안의 전송만** 조사 호출로 센다. ★`call_with_deadline` 은 이것을 딴 스레드에서 돌린다. 그래서 예산·정지 표를 **실어 보내야** 한다 — 스레드 지역이라 그냥은 안 보인다. """ from app.core.research_call_budget import research_calls_armed with research_calls_armed(): return client.responses.create(**kwargs) def payload_identity(system: str, user: str, schema: Any) -> str: """이 호출의 신원. ★**한 곳에서만** 만든다 — 두 곳에 적으면 한쪽만 고쳐진다. 프롬프트·subject 에 더해 **획득 좌표**(`acquisition_contract`)를 접는다. 안 접으면 `include` 하나만 바꿔도 옛 응답이 그대로 재사용된다. """ acq = json.dumps(acquisition_contract(), ensure_ascii=False, sort_keys=True) return _sha16("\x1e".join((system, user, str(schema), acq))) def _chunks(rows: Sequence[Dict[str, Any]], batch_size: int, max_bytes: int, *, era: str, region: str ) -> List[List[Dict[str, Any]]]: """개수와 **크기** 두 상한으로 자른다. ★원문은 **안 자른다**. 크기를 넘으면 그 대상을 **다음 batch 로** 넘긴다 — 자르는 순간 그 대상은 다른 것을 조사한 것이 되고, 그 결과는 `subject_payload_hash` 와도 안 맞는다. ★대상 하나가 혼자서 상한을 넘으면 **그 하나로 batch 를 만든다.** 건너뛰면 조용히 조사에서 빠지고, 「다 샀다」가 거짓이 된다. """ out: List[List[Dict[str, Any]]] = [] cur: List[Dict[str, Any]] = [] for r in rows: cand = cur + [r] size = len(build_user_prompt(cand, era=era, region=region).encode()) if cur and (len(cand) > batch_size or size > max_bytes): out.append(cur) cur = [r] else: cur = cand if cur: out.append(cur) return out def plan_calls(subjects: Sequence[Dict[str, Any]], batch_size: int, *, era: str, region: str, max_prompt_bytes: int = MAX_PROMPT_BYTES) -> int: """**보내기 전에** 실제 호출 수를 센다. ★무료 — 바깥 호출 없음. ★`ceil(len/batch)` 로는 모자란다 — 크기 상한에 걸리면 batch 가 더 쪼개져 **계획보다 많이 나간다**. 비용 가드가 계획된 수만 보면 그 초과분을 못 막고, 이미 쓴 뒤에 다음 회차에서야 걸린다. """ rows = [s for s in subjects if s] if not rows: return 0 return len(_chunks(rows, batch_size, max_prompt_bytes, era=era, region=region)) def planned_payloads(subjects: Sequence[Dict[str, Any]], batch_size: int, *, era: str, region: str, db=None, pack_version: Optional[str] = None, max_prompt_bytes: int = MAX_PROMPT_BYTES) -> List[str]: """**보내기 전에** 이 주행이 낼 호출들의 **신원**을 낸다. ★무료. ★`search_claims` 와 **같은 함수들**로 만든다 — `_chunks` · `build_user_prompt` · `payload_identity`. 손으로 다시 만들면 두 벌이 되고, 그러면 preflight 가 세는 것과 실제로 나가는 것이 갈린다. """ rows = [s for s in subjects if s] if not rows: return [] # ★`search_claims` 와 **같은 자리에서** 읽는다 — 칸 이름이 다르면 신원이 # 달라져 preflight 가 「전부 새것」이라고 말한다. pack = load_pack(db=db, version=pack_version) system = pack["stems"][SYSTEM_STEM]["content"] schema = pack["stems"][SCHEMA_STEM] out = [] for chunk in _chunks(rows, batch_size, max_prompt_bytes, era=era, region=region): user = build_user_prompt(chunk, era=era, region=region) out.append(payload_identity(system, user, schema)) return out def search_claims( client: Any, subjects: Sequence[Dict[str, Any]], *, model: str, era: str, region: str, batch_size: int, max_prompt_bytes: int = MAX_PROMPT_BYTES, hard_prompt_bytes: int = HARD_PROMPT_BYTES, db=None, pack_version: Optional[str] = None, opik_metadata: Optional[Dict[str, Any]] = None, on_batch: Optional[Any] = None, already: Optional[Sequence[Dict[str, Any]]] = None, ) -> Dict[str, Any]: """대상들을 **batch 로 잘라** 검색한다. 자르기만 한다. ``batch_size`` 는 **필수**다 — 기본값을 두면 실험이 정하기 전에 그 값이 계약처럼 굳는다 (Codex). Returns: ``batches`` — 호출 하나당 한 줄. 각 줄에 ``requested`` · ``raw`` · ``parsed`` · ``sources`` · ``provenance`` · ``error``. ★**판정을 여기서 하지 않는다.** 이 함수는 받은 것을 그대로 남기고, claim 검증과 결속 검사는 `grounding_claims` 와 acceptance 가 한다 — 조사와 채점이 같은 함수에 있으면 채점이 조사를 봐준다. """ # ★`isinstance(True, int)` 는 참이다 — 그냥 두면 `batch_size=True` 가 # batch 1 로 조용히 돈다. 이 판에서 `bool` 로 **다섯 번째** 데이는 부류다 # (`required`·`gradable`·`required_discriminator`·`time_capped` 에 이어). if isinstance(batch_size, bool) or not isinstance(batch_size, int) \ or batch_size < 1: raise ValueError( f"batch_size 는 1 이상 정수여야 한다: {batch_size!r} — " "§4b 실험이 정하기 전까지 기본값은 없다") for _n, _v in (("max_prompt_bytes", max_prompt_bytes), ("hard_prompt_bytes", hard_prompt_bytes)): if not isinstance(_v, int) or isinstance(_v, bool) or _v < 1: raise ValueError(f"{_n} 는 1 이상 정수여야 한다: {_v!r}") if hard_prompt_bytes < max_prompt_bytes: # ★hard 가 soft 보다 작으면 singleton 갈래가 통째로 죽는다 — # 모든 대상이 hard 를 넘어 아무것도 안 나간다. raise ValueError( f"hard({hard_prompt_bytes}) 가 soft({max_prompt_bytes}) 보다 작다") rows = [s for s in subjects if s] if not rows: return {"module": _MODULE, "batches": [], "batch_size": batch_size} from app.modules.llm.image_tracer import ( ambient_call_meta, record_provider_call, resolve_step_name) pack = load_pack(db=db, version=pack_version) system = pack["stems"][SYSTEM_STEM]["content"] schema = pack["stems"][SCHEMA_STEM] schema_body = schema["content"] if not isinstance(schema_body, dict) or "properties" not in schema_body: # ★모양이 아니면 **선다.** 산문만 보내고 「schema 를 썼다」로 쓰면 # 그 말이 거짓이 된다. raise ValueError(f"schema 가 JSON schema 가 아니다: {type(schema_body)}") meta = ambient_call_meta() if opik_metadata: meta = {**(meta or {}), **opik_metadata} step = resolve_step_name(STEP_NAME, meta) # ★★**보존과 재개는 다르다** (Codex). 앞 주행이 끊겨 줄이 남아 있어도, # 다시 돌릴 때 그걸 안 보면 **처음부터 다시 산다**. 이미 산 호출은 # 건너뛴다 — 키는 그 호출에 **넣은 subject 목록**이다. # ★깨진 줄은 재사용 안 한다 — 그건 안 산 것과 같다. # ★★키는 **subject 목록이 아니라 그 호출의 payload_hash 다.** id 만 # 보면 model·era·region·팩이 바뀌어도 옛 답을 그대로 쓴다 — 실측으로 # `model-A/1983/KR` 뒤 `model-B/2026/US` 가 재사용됐다 (Codex). # ★★**이번에 보낼 것부터 정한다.** 그래야 「앞 주행의 어느 줄이 *우리* # 호출이었나」를 payload 로 가릴 수 있다 — 모델만 같으면 같은 판으로 보면 # 옛 팩의 실패가 새 판을 막는다 (Codex). planned = [] for chunk in _chunks(rows, batch_size, max_prompt_bytes, era=era, region=region): user = build_user_prompt(chunk, era=era, region=region) planned.append((chunk, user, payload_identity(system, user, schema))) mine = {ph for _c, _u, ph in planned} done: Dict[str, Dict[str, Any]] = {} for r in (already or ()): if not isinstance(r, dict) or r.get("error") or r.get("not_sent"): continue prov = r.get("provenance") or {} ph = str(prov.get("payload_hash") or "") # ★신원이 안 맞으면 **남의 일**이다. 기록은 보존하되 이 판을 안 막는다. if ph not in mine or str(prov.get("model") or "") != str(model): continue # ★★★**같은 호출**의 포착 실패만 막는다. 이미 확정된 호환 불가를 # 재실행마다 다시 사지 않으면서, 새 팩은 막지 않는다. cap = capture_shape(r) if not cap["ok"]: raise ValueError( f"같은 호출(payload {ph})의 앞 주행이 포착에 실패했다: " f"{cap['why']} (모양={cap['shapes']}) — **다시 사지 않는다**. " "이 조합으로는 실험이 성립하지 않는다") done.setdefault(ph, r) batches: List[Dict[str, Any]] = [] reused = 0 for chunk, user, payload_hash in planned: if payload_hash in done: reused += 1 row = dict(done[payload_hash]) row["reused_from_earlier_run"] = True batches.append(row) # ★★재사용 행은 `on_batch` 에 **안 보낸다.** 보내면 매 재시작마다 # 같은 줄이 journal 에 또 쌓인다 (Codex). continue nbytes = len(user.encode()) if nbytes > hard_prompt_bytes: # ★**안 보낸다.** 보내 봐야 provider 가 거절하고, 그 거절을 # 「조사했는데 못 찾았다」로 읽으면 거짓이 된다. batches.append({ "requested": [str(s.get("research_subject_id")) for s in chunk], "prompt_bytes": nbytes, "subject_count": len(chunk), "raw": "", "parsed": None, "sources": [], "error": f"prompt 가 hard 상한을 넘어 보내지 않았다 " f"({nbytes} > {hard_prompt_bytes})", "not_sent": True, # ★우리 상한에 걸린 것이다 — `time_capped`(admission_limit)로 # 접혀 **retryable** 로 남는다. "limit_kind": LIMIT_ADMISSION, # ★★★전송은 없었지만 **무엇을 보내려 했는지는 안다** — 팩 # 좌표와 그 호출의 신원(payload identity)은 실재한다. # ★전에는 `None` 이었다. 그러면 저장 계약이 터지거나(스텝 # 예외) 걸러 버려 **조용히 사라진다** — 둘 다 틀렸다 (Codex). # 전송 0인 **bounded-run 미결**로 남아 재개할 수 있어야 한다. # ★`transmission_status="failed"` 이므로 `provider_request_id` # 는 비어도 된다 — 응답이 없었다는 사실 자체가 맞다. "provenance": { "provider": "openai", "model": model, "prompt_pack_version": pack["version"], "prompt_locator": pack["stems"][SYSTEM_STEM].get( "locator", ""), "schema_locator": pack["stems"][SCHEMA_STEM].get( "locator", ""), "prompt_raw_hash": pack["stems"][SYSTEM_STEM][ "raw_content_hash"], "schema_raw_hash": pack["stems"][SCHEMA_STEM][ "raw_content_hash"], "payload_hash": payload_hash, "local_trace_id": record_provider_call( step=step, model=model, prompt=f"{system}\n---\n{user}", status="error", duration_ms=0, meta=meta, operation="web_search", provider="openai", error=f"[not_sent] hard 상한 초과 {nbytes}") or "", "provider_request_id": "", # ★★★**provider 실패가 아니다.** 물리 전송이 **0** 인 # 로컬 상한이다 — `failed` 로 적으면 「provider 장애」로 # 오독되고 고칠 곳이 달라진다 (Codex). "transmission_status": TRANSMISSION_NOT_SENT, "transmission_error": (f"prompt 가 hard 상한을 넘어 보내지 " f"않았다 ({nbytes} > " f"{hard_prompt_bytes})"), }, }) # ★★**여기서도 바로 넘긴다.** 안 넘기면 「호출마다 즉시 저장」이 # 이 행에는 안 닿아, 안 보낸 사실이 주행이 끝나야 남는다 — # 중간에 끊기면 통째로 사라진다 (Codex). if on_batch: on_batch(batches[-1]) continue t0 = time.monotonic() row: Dict[str, Any] = { "requested": [str(s.get("research_subject_id")) for s in chunk], "not_sent": False, # ★**실제 크기를 남긴다** — 이게 없으면 검증 원고에서 정한 수를 # 실제 에피소드에 그대로 쓸 수 있는지 판단할 근거가 없다 (Codex). "prompt_bytes": len(user.encode()), "subject_count": len(chunk), "raw": "", "parsed": None, "sources": [], "error": "", } try: # ★★★**세 겹 중 ①·②·③ 이 여기서 만난다** (GROUNDING-V2 §8.5). # ①`call_with_deadline` — 한 호출의 벽시계. litellm 의 `timeout` # 은 per-read 라 slow-stream 을 못 잡는다. # ②·③ 전송 예산과 주행 마감은 `research_calls_armed()` 안에서 # **물리 전송 직전**에 걸린다 — 팔을 여기서 든다. # ★팔을 이 블록에만 드는 이유: 같은 스레드의 **다른** OpenAI # 호출이 조사 예산을 먹으면 안 된다 (Codex ②). # ★★`call_with_deadline` 은 **딴 스레드**에서 돌린다. 예산과 # ★★`strict` 를 **켠다**(`SCHEMA_STRICT=True`). 안 켠 판에서 # claim 16개가 **전부** `required` 칸을 빠뜨려 계약에서 # 거부됐다(2026-08-30 실측). 검색 도구와 같이 켜도 provider 는 # **거절하지 않는다**. # ★그래도 **우리 검증은 그대로 둔다** — provider 계약 하나에만 # 기대지 않는 독립 fail-closed 방어다. # 정지 표는 스레드 지역이라 그냥은 **안 보인다** — 안 실어 보내면 # 상한이 그 자리에서 통째로 no-op 이 된다. resp = call_with_deadline( bind_current_research_budget(_armed_create), client, dict( model=model, instructions=system, input=[ {"role": "user", "content": [{"type": "input_text", "text": user}]}], tools=[build_web_search_tool( want_images=not TEXT_ONLY_SEARCH)], include=list(RESPONSE_INCLUDE), store=PROVIDER_STORE, text={"format": {"type": "json_schema", "name": "grounding_claims", "strict": SCHEMA_STRICT, "schema": schema_body}}), deadline_seconds=PER_CALL_DEADLINE_SECONDS) except Exception as exc: # noqa: BLE001 — 전송 실패도 **기록한다** # ★★★**사용자 취소는 그대로 올린다** (Codex 실측). 여기서 잡아 # 행으로 바꾸면 ①취소가 「provider 장애」로 잘못 기록되고 # ②호출자에게 안 전해져 **다음 batch 가 계속 돈다**. # ★우리 상한(호출 마감·전송 예산·주행 마감)만 접는다 — 그건 # 계약상 `unresolved`·retryable 이고, 취소는 그게 아니다. if _is_cancellation(exc): raise ms = int((time.monotonic() - t0) * 1000) # ★★**「이 주행의 제한에 걸린 것」과 「전송이 깨진 것」을 가른다.** # 둘 다 결론을 못 냈지만 **고칠 곳이 다르다** — 앞의 것은 상한· # 마감을 손보는 자리이고 뒤의 것은 provider·네트워크다. # ★그래도 판정은 같다: `unresolved` · retryable (계약 §3). limit_kind = _limit_kind_of(exc) row["limit_kind"] = limit_kind # ★비용·trace 는 **걸린 경우에도 남긴다** (Codex ④). 늦게 온 # worker 가 판정·CP·DB 를 못 바꾸는 것과, 그 시도가 기록에서 # 사라지는 것은 다른 얘기다. call_id = record_provider_call( step=step, model=model, prompt=f"{system}\n---\n{user}", status="error", duration_ms=ms, meta=meta, operation="web_search", provider="openai", error=f"[{limit_kind}] {exc}"[:500] if limit_kind else str(exc)[:500]) # ★전송 실패는 **행으로 남는다** — `build_revision_row` 가 # `transmission_status="failed"` 로 받아 retryable 로 저장한다. row["error"] = str(exc)[:500] row["provenance"] = { "provider": "openai", "model": model, "prompt_pack_version": pack["version"], "prompt_locator": pack["stems"][SYSTEM_STEM].get("locator", ""), "schema_locator": pack["stems"][SCHEMA_STEM].get("locator", ""), "prompt_raw_hash": pack["stems"][SYSTEM_STEM][ "raw_content_hash"], "schema_raw_hash": pack["stems"][SCHEMA_STEM][ "raw_content_hash"], "payload_hash": payload_hash, "local_trace_id": str(call_id or ""), "transmission_status": "failed", "transmission_error": str(exc)[:500], } batches.append(row) if on_batch: on_batch(row) continue ms = int((time.monotonic() - t0) * 1000) call_id = record_provider_call( step=step, model=model, prompt=f"{system}\n---\n{user}", status="success", duration_ms=ms, meta=meta, operation="web_search", provider="openai", output_text="[claims search completed]") said = "" for o in (getattr(resp, "output", None) or []): if getattr(o, "type", "") == "message": for c in (getattr(o, "content", None) or []): said += getattr(c, "text", "") or "" row["raw"] = said.strip() # ★공식 인용 경로 — 메시지 annotation 의 `url_citation`. `action.sources` # 와 함께 **소유권 감사에만** 쓴다(본문은 안 준다). row["citations"] = _url_citations(resp) row["parsed"] = _extract_json_object(said) row["sources"] = _collect_sources(resp) # ★★모양을 모르는 자리에서는 **무엇이 왔는지를 통째로** 남긴다. # 안 남겨서 유료 1회를 쓰고도 키 이름을 몰랐다. row["source_shape_probe"] = raw_source_shape(resp) # ★★**받자마자 통째로 남긴다.** provider 보존(`store`)에 기대지 않고 # 되짚을 수 있게 — 유료로 산 것의 원형을 잃은 적이 있다. try: row["response_dump"] = resp.model_dump(mode="json") except Exception: # noqa: BLE001 — 덤프 실패가 조사를 죽이면 안 된다 row["response_dump"] = {"_dump_failed": True} row["provenance"] = { "provider": "openai", "model": model, "prompt_pack_version": pack["version"], "prompt_locator": pack["stems"][SYSTEM_STEM].get("locator", ""), "schema_locator": pack["stems"][SCHEMA_STEM].get("locator", ""), "prompt_raw_hash": pack["stems"][SYSTEM_STEM]["raw_content_hash"], "schema_raw_hash": pack["stems"][SCHEMA_STEM]["raw_content_hash"], "payload_hash": payload_hash, "local_trace_id": str(call_id or ""), "provider_request_id": str(getattr(resp, "id", "") or ""), "transmission_status": "ok", } batches.append(row) # ★★**한 호출이 끝나면 바로 넘긴다.** 전에는 그 batch 크기의 호출이 # 전부 끝난 뒤에야 호출부가 결과를 받았다 — 9회 중 7회째에 끊기면 # 앞의 7회를 **통째로 잃고 다시 사야** 했다 (Codex). if on_batch: on_batch(row) logger.info("%s: %d 대상 / batch %d → 호출 %d회 (실패 %d)", _MODULE, len(rows), batch_size, len(batches), sum(1 for b in batches if b["error"])) return { "module": _MODULE, "pack_version": pack["version"], "pack_manifest_hash": pack["pack_manifest_hash"], "batch_size": batch_size, "max_prompt_bytes": max_prompt_bytes, "batches": batches, "hard_prompt_bytes": hard_prompt_bytes, # ★실제로 나간 크기의 최댓값 — 다음 판의 상한을 이걸로 정한다. "max_prompt_bytes_seen": max((b["prompt_bytes"] for b in batches), default=0), # ★soft 를 혼자 넘어 singleton 으로 간 수. 이게 늘면 soft 상한이 # 현실과 안 맞는다는 뜻이다 — batch 크기를 정할 때 같이 본다. "oversized_subject_count": sum( 1 for b in batches if b["subject_count"] == 1 and b["prompt_bytes"] > max_prompt_bytes), # ★hard 를 넘어 **아예 안 나간** 것. 「조사했다」에 안 든다. "not_sent_count": sum(1 for b in batches if b.get("not_sent")), # ★논리 호출과 물리 전송을 **갈라 적는다** — 이 판에서 세 번 고친 혼동이다. "logical_calls": len(batches), # ★★**산 것과 재사용한 것을 갈라 적는다.** 합쳐 두면 「19회 샀다」가 # 거짓이 된다 — 재개하면 실제로 산 것은 그보다 적다. "reused_calls": reused, # ★★★**안 보낸 행은 산 것이 아니다** (Codex). `len(batches)` 로 세면 # 물리 전송 0인 로컬 상한 행이 「유료 논리 호출 1」로 오독된다. # ★「샀다」는 **provider 를 부르려고 시도한 행**만이다. "bought_calls": sum(1 for b in batches if not b.get("reused_from_earlier_run") and not b.get("not_sent")), "not_sent_calls": sum(1 for b in batches if b.get("not_sent")), }