"""두 VLM 을 각각 한 번씩 부르고 **합의는 안 하는** 실행부 (2026-08-27, #92). 사용자 지시: gpt sol vlm 은 성능이 안좋아 … **무조건 gemini 3.1 pro 와 grok 최신 모델 둘을 사용해야해** / 아홉 전부 바꿔 ## 이 모듈이 하는 것과 안 하는 것 한다 두 alias 를 각각 **한 번씩** 부른다 fallback 을 봉인한다 모델별 raw·물리 모델·오류·토큰·비용을 **같은 모양으로** 남긴다 안 한다 합의 · 기각 · 승자 선정 · 재시도 한 모델 실패를 다른 모델 두 번째 호출로 메우기 ★**합의를 여기 넣으면 안 된다** (2026-08-27 Codex 판정). 업무마다 실패의 뜻이 다르다: binary defect gate 둘 다 broken 일 때만 기각, 불일치는 split ranking·selection 모델별 점수를 보존해 호출자 규칙으로 합산 관찰·critique 모델별 관찰을 기록하고 기존 issue 계약으로 결합 geometry·추출 구조화 ID 합의 — **좌표 평균 금지** 예가 코드에 있다. `ref_image_pipeline` 의 두 호출자는 같은 운반층을 쓰는데 하나(`validate_reference_image`)는 severe 면 **돈 내고 재생성**하고 다른 하나(`compare_two_images`)는 이미 산 두 장 중 승자를 고른다. 같은 합의로 묶으면 실패의 뜻이 섞인다. ## fallback 봉인이 왜 계약인가 `call_structured` 는 기본이 `enable_fallback=True` 라 Gemini/Grok 이 실패하면 **GPT Sol 이 다시 들어온다.** 그러면 사용자 결정을 조용히 어기고 기록에도 Sol 이 남지 않는다. 여기서 못박는다. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Sequence # 사용자 지시가 정한 짝. 순서가 곧 기록 순서다. DUAL_VLM_ALIASES: tuple = ("gemini-pro", "grok") @dataclass class OneCall: """한 모델의 한 번 — 성공이든 실패든 **같은 모양**으로 남는다.""" alias: str ok: bool payload: Optional[Dict[str, Any]] = None error: Optional[str] = None usage: Dict[str, Any] = field(default_factory=dict) @property def physical_model(self) -> str: """응답이 말한 물리 모델 — 설정이 아니라. 못 읽었으면 빈 문자열이다. **alias 로 대신 채우지 않는다** — 그러면 「무엇이 답했는지」와 「무엇을 부르려 했는지」가 섞인다. """ return str(self.usage.get("physical_model") or "") @dataclass class DualResult: """두 호출의 결과. **판단은 안 들어 있다** — 호출자가 한다.""" calls: List[OneCall] @property def complete(self) -> bool: """둘 다 답했나. ★거짓이면 **호출자는 기각 권한이 없다.** 한쪽만 보고 돈을 더 쓰거나 산출을 버리면, 실패한 모델이 무슨 말을 했을지 모르는 채 결정하는 것이다. 이 값이 그 계약을 자료로 만든다 — 판정은 호출자 몫이다. """ return bool(self.calls) and all(c.ok for c in self.calls) @property def payloads(self) -> List[Dict[str, Any]]: """성공한 것만, **요청한 순서 그대로**.""" return [c.payload for c in self.calls if c.ok and c.payload is not None] def by_alias(self, alias: str) -> Optional[OneCall]: for c in self.calls: if c.alias == alias: return c return None def provenance(self) -> Dict[str, Any]: """기록에 그대로 넣을 모양 — 모델별 raw 와 비용을 다 담는다.""" return { "contract": "dual_vlm_v1", "aliases": [c.alias for c in self.calls], "complete": self.complete, "calls": [ { "alias": c.alias, "physical_model": c.physical_model, "ok": c.ok, "error": c.error, # ★미보고 키는 **없는 채로** 둔다 — 0 으로 채우면 # 「안 썼다」로 읽힌다. "usage": dict(c.usage), "raw": c.payload, } for c in self.calls ], } def ask_both( step: str, system_prompt: str, user_prompt: "str | list", response_schema: Dict[str, Any], *, aliases: Sequence[str] = DUAL_VLM_ALIASES, schema_name: str = "response", opik_metadata: Optional[Dict] = None, temperature: float = 0.2, max_tokens: Optional[int] = None, parallel: bool = False, ) -> DualResult: """두 alias 를 **각각 한 번씩** 부른다. 한 쪽이 죽어도 다른 쪽은 부른다 — 둘의 말을 다 듣는 것이 이 판의 목적이고, 한쪽 실패를 이유로 나머지를 안 물으면 `complete=False` 인지 「둘 다 나쁘다」인지 구분이 안 된다. ★**재시도하지 않는다.** 실패한 모델을 다시 부르면 그 모델이 두 번 들어와 「독립 두 의견」이라는 전제가 깨지고 돈도 두 번 나간다. Args: aliases: 부를 순서. 기본은 사용자 지시가 정한 짝. step: 각 호출의 스텝 태그 — alias 별로 접미사를 붙여 기록에서 갈린다(`__gemini-pro`). parallel: 두 alias 를 **동시에** 부른다 (2026-08-29 사용자 지시). ★기본은 False 다 — **호출부가 켠다**(Codex 설계 리뷰). 이 함수는 cine 온전성 검증 말고도 참조 검증·비교가 쓰므로 전역 기본을 바꾸면 손대지 않은 자리의 동시성까지 달라진다. ★결과는 **완료 순서가 아니라 `aliases` 순서**로 담는다. 실패도 같다 — 기록이 순서에 흔들리면 「어느 모델이 무엇을 말했나」가 호출마다 달라진다. ★worker 에 ContextVar 셋을 **명시 전파**한다(예산·capture·trace). 롤 생성 병렬(`multiroll_select.py:1491-1502`)과 같은 처리이고, 그 주석이 「하나라도 빠뜨리면 그 축만 조용히 무너진다」고 적고 있다. ★이것은 **산출 계약이 아니다** — 입력·모델·결과 순서가 같으므로 지문을 움직이지 않는다. Returns: [[DualResult]] — 합의 없음. `complete` 와 모델별 raw 만 있다. """ from app.modules.llm.llm_client import call_structured def _one(alias: str) -> OneCall: tag = f"{step}__{alias}" sink: Dict[str, Any] = {} try: payload = call_structured( tag, system_prompt, user_prompt, response_schema, project_config={tag: {"model": alias}}, schema_name=schema_name, opik_metadata=opik_metadata, temperature=temperature, max_tokens=max_tokens, # ★계약 — 여기가 열리면 Gemini/Grok 실패 뒤 Sol 이 조용히 # 돌아온다. 시험이 이 인자를 AST 로 잠근다. enable_fallback=False, # ★★**그것만으로는 「각 한 번」이 안 된다** (2026-08-27 # Codex BLOCK). `enable_fallback` 은 `call_structured` 의 # Tier 2/3 만 닫고, Router 는 `num_retries=3` 으로 지어져 # **한 alias 가 최대 4번 전송**될 수 있다. 재시도는 다 # 과금되는데 `usage_sink` 는 마지막 응답만 보므로 앞선 # 시도가 기록에서 사라진다. # ★막으면 일시 오류에 그 심판을 잃는다 — 그것이 맞다. # `complete=False` 가 기각 권한을 거두므로 조용한 격하가 # 아니고, **모르는 사이에 네 번 사는 것**보다 정직하다. num_retries=0, usage_sink=sink, ) return OneCall(alias=alias, ok=True, payload=payload, usage=sink) except Exception as exc: # noqa: BLE001 — 한쪽 실패가 나머지를 안 막는다 return OneCall( alias=alias, ok=False, error=f"{type(exc).__name__}: {exc}"[:400], usage=sink) if not parallel or len(aliases) < 2: return DualResult(calls=[_one(a) for a in aliases]) from concurrent.futures import ThreadPoolExecutor from app.core.image_call_budget import bind_current_budget from app.core.send_ledger import bind_current_ledger from app.modules.llm.opik_trace import bind_current_trace from app.services.image_capture.context import ( bind_current_generation_context, ) # ★셋 다 ContextVar 다 — 하나라도 빠뜨리면 그 축만 조용히 무너진다 # (롤 생성 병렬의 주석 그대로). # ★**발송 장부도 나른다** (2026-09-20 Codex BLOCK). 안 실으면 이 # pool 안의 판정 호출이 통째로 안 잡혀 **「관측 0」**이 된다 — # 롤 팬아웃만 실어서는 **판정 pool 에 전파되지 않는다**. bound = bind_current_budget( bind_current_ledger( bind_current_generation_context(bind_current_trace(_one)))) with ThreadPoolExecutor(max_workers=len(aliases)) as pool: futures = {a: pool.submit(bound, a) for a in aliases} # ★완료 순서가 아니라 **alias 순서**로 모은다. return DualResult(calls=[futures[a].result() for a in aliases])