"""LiteLLM + Opik 통합 LLM 클라이언트. 모든 텍스트 LLM 호출을 LiteLLM Router를 통해 수행하고, Opik callback으로 자동 추적한다. 기존 GeminiTextClient / OpenAIClient / gemini_key_pool / llm_logger를 대체. 이미지 생성(gemini_image_client)은 별도 유지. """ import json import logging import os import threading from pathlib import Path from typing import Any, Callable, Dict, List, NamedTuple, Optional import jsonschema import litellm from litellm import Router logger = logging.getLogger(__name__) class EmptyLLMResponse(RuntimeError): """모델이 200 으로 답했는데 내용이 비었다. 원인은 이것만으로 모른다 — 검열일 수도, 토큰 소진일 수도, 제공자 사정일 수도 있다. 그래서 메시지에 `finish_reason` 을 같이 싣는다. ``RuntimeError`` 하위라 기존에 이것을 잡던 자리는 그대로 잡는다. """ def _is_local_schema_validate_enabled() -> bool: """local jsonschema 검증 toggle 평가 (problems.md #13, review I1). 우선순위: ``LLM_LOCAL_SCHEMA_VALIDATE`` ENV → ``settings.llm_local_schema _validate`` (Pydantic). 둘 다 미설정 시 default True. provider strict mode (LiteLLM ``response_format=json_schema strict=True``) 가 일부 provider 에서 일부 필드만 enforce 할 수 있으므로 JSON 파싱 후 ``jsonschema.validate`` 로 한 번 더 shape 검증한다. 운영 중 schema 배포 시 즉시 disable 가능. """ raw = os.environ.get("LLM_LOCAL_SCHEMA_VALIDATE") if raw is not None: return raw.strip().lower() in ("1", "true", "yes", "on") try: from app.core.config import settings return bool(getattr(settings, "llm_local_schema_validate", True)) except Exception: return True # structured output 이 tool-call 래퍼를 흘릴 때 최상위에 붙는 키. # 2026-08-06 실측: Anthropic 경로가 간헐적으로 `{"parameters": {…}}` 로 한 겹 # 감싸 돌려준다(같은 프롬프트 4회 중 1회). 안쪽은 스키마와 정확히 일치한다. # 벗기지 않으면 `_validate_local_schema` 가 root 에서 실패하고 call_structured # 가 Tier 2 로 내려가는데, Tier 2 는 **입력을 sanitize 한 다른 프롬프트**로 # 다시 묻는다 — 판정에 조용한 계약 변경이 들어간다(223쌍 재판정에서 92건). _TOOL_ENVELOPE_KEYS = frozenset({"parameters", "arguments", "input"}) def _unwrap_tool_envelope(payload: Any, schema: Dict[str, Any]) -> Any: """tool-call 래퍼 한 겹을 벗긴다 — 스키마가 그 키를 원하지 않을 때만. 보수적으로만 벗긴다: ①최상위가 dict 이고 키가 **정확히 하나** ②그 키가 알려진 래퍼 이름 ③스키마의 properties 에 그 키가 **없다** ④안쪽이 dict. 넷을 다 만족하면 스키마가 그 키를 담을 수 없으므로 래퍼가 확실하다. 하나라도 어긋나면 원본을 그대로 돌려준다. """ if not isinstance(payload, dict) or len(payload) != 1: return payload (key,), (inner,) = payload.keys(), payload.values() if key not in _TOOL_ENVELOPE_KEYS or not isinstance(inner, dict): return payload props = schema.get("properties") if isinstance(schema, dict) else None if isinstance(props, dict) and key in props: return payload logger.info("structured output: '%s' 래퍼 한 겹 벗김", key) return inner def _validate_local_schema( payload: Dict[str, Any], response_schema: Dict[str, Any], *, step: str, schema_name: str, ) -> None: """``jsonschema.validate`` wrapper — 실패 시 ``SchemaValidationError`` raise. ENV ``LLM_LOCAL_SCHEMA_VALIDATE=false`` 면 no-op. ``response_schema`` 가 빈 dict 이거나 비-dict 면 검증 skip (legacy/free-form caller 보호). ``ValidationError`` 의 path/message 를 보존하여 Tier 2/3 로직과 logger 가 root cause 진단 가능. """ from app.modules.llm.safety import InvalidSchemaError, SchemaValidationError if not _is_local_schema_validate_enabled(): return if not isinstance(response_schema, dict) or not response_schema: return try: jsonschema.validate(instance=payload, schema=response_schema) except jsonschema.ValidationError as exc: path = "/".join(str(p) for p in exc.absolute_path) or "" raise SchemaValidationError( f"Local schema validation failed for step={step} schema={schema_name} " f"at path={path}: {exc.message}" ) from exc except jsonschema.SchemaError as exc: # caller 가 invalid jsonschema 전달 — fallback 진행해도 동일 결과이므로 # InvalidSchemaError(ValueError) 로 래핑하여 _NON_SAFETY 분류 + ValueError # 호환성 양쪽 보존 (problems.md #13 B2). raise InvalidSchemaError( f"Invalid jsonschema spec for step={step} (schema_id={schema_name}): " f"{exc.message}" ) from exc # ── Thread-local Opik context (StepRunner에서 설정, call_* 에서 자동 병합) ── _thread_local = threading.local() def set_opik_context(meta: Optional[Dict] = None) -> None: """현재 스레드에 Opik metadata context 설정. call_* 호출 시 자동 병합.""" _thread_local.opik_meta = meta def _get_thread_opik_meta() -> Optional[Dict]: return getattr(_thread_local, "opik_meta", None) def _build_opik_metadata(step: str, opik_metadata: Optional[Dict] = None) -> Dict: """Opik metadata 구성 — thread-local context 자동 병합. 우선순위: 명시적 opik_metadata > thread-local context > step tag만 """ metadata = {"opik": {"tags": [step]}} # 1) thread-local context (StepRunner에서 설정) thread_meta = _get_thread_opik_meta() if thread_meta: extra_tags = thread_meta.get("tags", []) merged = {k: v for k, v in thread_meta.items() if k != "tags"} metadata["opik"].update(merged) if extra_tags: metadata["opik"]["tags"].extend(extra_tags) # 2) 명시적 파라미터 (덮어쓰기) — 원본 dict 변경 안 함 if opik_metadata: extra_tags = opik_metadata.get("tags", []) merged = {k: v for k, v in opik_metadata.items() if k != "tags"} metadata["opik"].update(merged) if extra_tags: metadata["opik"]["tags"].extend(extra_tags) # ── v2 (2026-08-23): litellm 이 실제로 읽는 키만 채운다 ────────────── # litellm 이 metadata["opik"] 에서 읽는 것은 넷뿐이다: # project_name · current_span_data · tags · thread_id # 우리가 싣던 `trace_name` 은 litellm 소스에 0회 — 죽은 키였다. # `session_id` 도 litellm 은 안 본다(thread_id 를 본다) — 그래서 텍스트 # 호출은 thread 가 통째로 없었다(실측: 76%가 None). try: from app.core.config import settings if getattr(settings, "opik_trace_v2_enabled", False): from app.modules.llm.opik_trace import ( build_axis_tags, current_trace, is_axis_tag) opik_block = metadata["opik"] # 부모 trace 가 열려 있으면 litellm 은 trace 를 만들지 않고 # span 만 붙인다 → `chat.completion` 이 사라진다. parent = current_trace() if parent is not None: opik_block["current_span_data"] = {"trace_id": parent.uid} # 죽은 키는 내보내지 않는다 — trace metadata 만 더럽힌다. opik_block.pop("trace_name", None) opik_block.pop("session_id", None) # ── 태그를 축으로 (2026-08-24 Codex BLOCK 2) ────────────── # litellm 이 만드는 span 의 **이름은 우리가 못 정한다** # (`{model}_{obj_type}_{created}`, 설계 ⑤). 그래서 그 span 이 # 무엇인지는 태그가 말해야 하는데, 여기서 태그가 맨 이름으로 # 시작하고(위 `{"tags": [step]}`) 호출자가 실은 맨 이름까지 # 그대로 붙어 나갔다 — `entity_extractor_v3.py:446` 은 엔티티 # 이름(한글 고유명사)을 태그로 싣는다. 축도 안 갈리고 # 카디널리티도 터진다. # 호출 단계는 `op:` 축이다(스텝 묶음은 StepRunner 가 `step:` 으로 # 이미 실어 보낸다). 맨 이름은 버린다 — 설계 ⑥ 「맨 이름: # 접두사를 붙여 축을 밝힌다」. # ★축 판별은 허용 접두사 whitelist 다 — `":" in tag` 로 보면 # URL·동적 이름이 축 태그로 둔갑한다(2026-08-24 Codex 재리뷰). axis_tags = build_axis_tags(op=step) for tag in (opik_block.get("tags") or []): if is_axis_tag(tag) and tag not in axis_tags: axis_tags.append(tag) opik_block["tags"] = axis_tags except Exception as exc: # noqa: BLE001 — 기록은 본 작업을 안 막는다 logger.debug("_build_opik_metadata v2 배선 실패 (non-fatal): %s", exc) return metadata def build_call_metadata(op: str, tags: Optional[List[str]] = None) -> Dict: """`router_completion` 을 **직접** 부르는 자리가 쓰는 표준 metadata. `call_structured` 계열은 내부에서 `_build_opik_metadata` 를 타지만, router 를 직접 부르는 자리는 그것을 건너뛴다. 그래서 나가는 태그가 맨 이름 하나뿐이었고(`["ref_validation"]`), 축도 안 갈리고 부모 trace 에도 안 붙어 `chat.completion` 으로 홀로 남았다. 2026-08-25 실측(최소 검증판 주행): span 26/252 가 스텝 축을 잃었고 그중 14 건이 참조 검증이었다. 이 함수를 쓰면 축 태그·thread_id·부모 trace 가 다른 호출과 **같은 규칙**으로 실린다. """ return _build_opik_metadata(op, {"tags": list(tags)} if tags else None) # ── Opik 콜백 설정 ── _opik_initialized = False # 전역 lazy init 보호 (멀티 worker에서 _init_opik/_get_router 동시 호출 방지). # RLock — _get_router 내부에서 _init_opik 호출 시 재진입 가능해야 함. _init_lock = threading.RLock() # 감싼 함수임을 표시하는 이름 — 전역 변수 대신 함수 속성에 남긴다. 모듈이 # 다시 읽히면 전역은 초기화되지만 설치본 모듈에 꽂아 둔 함수는 그대로라, # 속성으로 봐야 이중 래핑을 실제로 막는다. _OPIK_CACHE_WRAP_MARK = "_theroad_opik_cache_fields_wrapper" # 캐시 항목 배선이 실제로 걸렸는가. litellm 구조가 바뀌어 배선이 무력화되면 # 토큰 기록은 계속 남고 캐시 칸만 비어, 겉으로는 "캐시가 안 걸린 주행"과 # 구분이 안 된다. 그래서 상태를 남기고 콜백 켬 로그에 함께 적는다. _opik_cache_wiring_ok = False def _cache_fields_of(usage) -> Dict[str, int]: """usage 에서 캐시로 재사용된 입력 토큰을 꺼낸다 (없으면 빈 dict). 두 자리를 본다 — `prompt_tokens_details.cached_tokens` (OpenAI 형식, litellm 이 Gemini 의 cachedContentTokenCount 도 여기 채운다) 와 `cache_read_input_tokens` (Anthropic 형식). 값이 없으면 키를 만들지 않는다: 없는 것을 0 으로 적으면 적중률 0% 라는 거짓 기록이 남는다. """ def _int_or_none(value): if isinstance(value, bool) or not isinstance(value, int): return None return value fields: Dict[str, int] = {} details = getattr(usage, "prompt_tokens_details", None) cached = None if isinstance(details, dict): cached = details.get("cached_tokens") elif details is not None: cached = getattr(details, "cached_tokens", None) cached = _int_or_none(cached) if cached is not None: fields["cached_tokens"] = cached read = _int_or_none(getattr(usage, "cache_read_input_tokens", None)) if read is not None: fields["cache_read_input_tokens"] = read return fields def _wrap_opik_usage_object() -> None: """litellm 의 Opik 콜백이 버리는 캐시 항목을 usage 에 되살린다. `litellm/integrations/opik/utils.py` 의 `create_usage_object` 는 세 값 (completion/prompt/total)만 담는다. 캐시로 재사용된 입력 토큰은 거기서 사라져 Opik span 에 남지 않는다 — 값은 제공사가 주는데 기록만 없다. 그 상태로는 프롬프트 조립 순서를 바꿔도 효과를 잴 수 없다. 설치본 파일은 고치지 않는다(재설치 때 날아간다). 호출 지점이 `utils.create_usage_object(...)` 로 모듈 속성을 매번 찾으므로 속성만 바꿔 두면 걸린다. 원 함수는 `__wrapped__` 로 들고 있고, 표식을 보고 두 번 감싸지 않는다. 배선이 안 걸리면 **ERROR 로 남기고** `_opik_cache_wiring_ok` 를 False 로 둔다. 이 실패는 조용하면 안 된다 — 토큰 기록은 그대로 남고 캐시 칸만 비어서, 나중에 보면 "캐시가 안 걸린 주행"과 구분이 안 된다. 다만 호출 자체는 죽이지 않는다(기록 보강이 본 작업을 막지 않는다는 기존 관례). """ global _opik_cache_wiring_ok _opik_cache_wiring_ok = False try: from litellm.integrations.opik import utils as opik_utils except Exception as exc: # litellm 버전이 바뀌어 경로가 없어진 경우 logger.error( "Opik usage 캐시 항목 배선 실패 — litellm opik utils 없음: %s. " "캐시 적중은 기록되지 않는다(토큰 세 값은 남는다).", exc) return original = getattr(opik_utils, "create_usage_object", None) if original is None: logger.error( "Opik usage 캐시 항목 배선 실패 — create_usage_object 없음 " "(litellm 구조 변경). 캐시 적중은 기록되지 않는다" "(토큰 세 값은 남는다).") return if getattr(original, _OPIK_CACHE_WRAP_MARK, False): _opik_cache_wiring_ok = True return # 이미 감쌌다 def create_usage_object(usage): usage_dict = original(usage) try: usage_dict.update(_cache_fields_of(usage)) except Exception as exc: # 기록 보강이 호출을 죽이면 안 된다 — 세 값은 그대로 남긴다. logger.warning("Opik usage 캐시 항목 추출 실패: %s", exc) return usage_dict setattr(create_usage_object, _OPIK_CACHE_WRAP_MARK, True) create_usage_object.__wrapped__ = original try: opik_utils.create_usage_object = create_usage_object except Exception as exc: logger.error( "Opik usage 캐시 항목 배선 실패 — 함수 교체 불가: %s. " "캐시 적중은 기록되지 않는다(토큰 세 값은 남는다).", exc) return _opik_cache_wiring_ok = True logger.info("Opik usage 에 캐시 항목 기록 켬 (create_usage_object 래핑)") def _init_opik(): global _opik_initialized if _opik_initialized: return with _init_lock: if _opik_initialized: # double-checked locking return from app.core.config import settings # 셀프 호스팅(2026-08-14): 키 없이 주소만으로도 켠다 — SDK 는 # 환경 변수만 읽으므로 .env 값을 여기서 올린다. if settings.opik_api_key or settings.opik_url_override: if settings.opik_api_key: os.environ["OPIK_API_KEY"] = settings.opik_api_key if settings.opik_url_override: os.environ["OPIK_URL_OVERRIDE"] = settings.opik_url_override os.environ["OPIK_WORKSPACE"] = settings.opik_workspace os.environ["OPIK_PROJECT_NAME"] = settings.opik_project_name # 콜백을 켜기 전에 감싼다 — 켠 뒤에 감싸면 그 사이 호출이 # 캐시 항목 없이 기록된다. _wrap_opik_usage_object() litellm.callbacks = ["opik"] logger.info( "Opik callback enabled (project: %s, url: %s, cache_wiring=%s)", settings.opik_project_name, settings.opik_url_override or "cloud", "ok" if _opik_cache_wiring_ok else "failed") _opik_initialized = True # ── LiteLLM Router 초기화 ── _router: Optional[Router] = None class _RouterBinding(NamedTuple): """Router 와 **그것을 지을 때 쓴 슬롯**을 하나로 묶는다. [2026-08-01 A5 후속, Codex 재확인 BLOCKING] Router 는 어느 슬롯 키로 지어졌는지 기록하지 않았고, ``_completion`` 은 호출 직전 전역 활성 슬롯을 다시 읽어 그것을 자기 슬롯으로 삼았다. 그래서 이런 창이 열린다: ① 한 요청이 primary 키로 지은 Router 를 이미 들고 있다 ② 다른 요청이 전역을 secondary 로 옮긴다 ③ ①이 그 stale Router 로 호출한다 → 실제로는 primary 로 나간다 ④ primary billing 실패가 "secondary 실패"로 보고된다 ⑤ 마지막 슬롯 소진 판정 — **보조 키를 한 번도 안 써 보고 죽는다** 슬롯과 Router 를 짝으로 다루면 ③에서 실패한 슬롯을 정확히 지목한다. """ slot: Optional[str] router: Router _binding: Optional[_RouterBinding] = None def _load_gemini_keys() -> List[str]: """backend/.env 에서 GEMINI_API_KEY* 수집.""" env_file = Path(__file__).resolve().parent.parent.parent.parent / ".env" keys = [] if not env_file.exists(): return keys for line in env_file.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, _, v = line.partition("=") k = k.strip() v = v.strip().strip('"').strip("'") if k.startswith("GEMINI_API_KEY") and v: if v not in keys: keys.append(v) return keys def _completion(binding: "_RouterBinding", model: str, kwargs: Dict[str, Any]): """Router 호출 — OpenAI **키 수준** 실패면 다음 키 슬롯으로 재시도. 전환 대상은 billing hard limit / quota 소진 / 401·403 뿐이다. 그 외 실패(타임아웃·5xx·단순 429·스키마 위반)는 그대로 올려 기존 재시도· Tier fallback 경로가 처리하게 둔다. 전환이 일어나면 Router 가 무효화 되므로 `_get_router_binding()` 로 새 키가 박힌 짝을 다시 받는다. ★``binding`` 을 받는다 (2026-08-01 A5 후속) — 전역 활성 슬롯을 호출 직전에 다시 읽으면, stale Router 를 든 요청이 **다른 요청이 옮겨 놓은 슬롯**을 자기 것으로 착각한다. 그러면 primary 실패가 secondary 실패로 보고되어 보조 키를 한 번도 안 써 보고 죽는다. 실패한 슬롯은 그 Router 를 지은 슬롯이지, 지금 전역이 가리키는 슬롯이 아니다. ★**정지 관문이 여기 있다** (2026-08-26). 이미지 호출은 `reserve_current_call` 이라는 공통 길목이 있는데 글 호출에는 없어서, `scene_detail` 같은 글 팬아웃은 정지를 전혀 못 들었다. ★처음엔 `router_completion` 에 걸었는데 **헛자리였다** — 그 함수의 docstring 이 「Router 를 거치는 유일한 호출 경로」라고 적혀 있어 믿었지만, `call_structured`·`call_text`·`call_multiturn` 은 전부 이 함수를 **직접** 부른다. 문서를 믿지 말고 호출부를 세야 했다. 진짜 공통 길목은 여기다. ★키 전환 루프 **앞**에 둔다 — 논리 호출당 한 번만 보면 된다. 루프 안에 두면 슬롯 수만큼 조회가 늘어난다. """ from app.core import openai_keys from app.core.research_call_budget import reserve_current_research_call from app.core.send_ledger import GRAIN_SLOT as _GRAIN_SLOT from app.core.send_ledger import record_send as _record_send # ★alias 별 정리를 **여기서도** 건다. `call_structured` 계열은 위에서 # 부르지만 `router_completion` 을 직접 부르는 자리 넷은 안 지난다 # (ref_image_pipeline · text_cleaner · fal_angle_helpers×2). # 규칙은 한 곳에 적고 **모든 길이 그 곳을 지나게** 한다 — 두 곳에 적으면 # 한쪽만 고쳐진다. 두 번 걸려도 무해하다(값 치환·min 뿐). _sanitize_kwargs_for_model(model, kwargs) stop_check = _current_stop_check() if stop_check is not None: stop_check() attempts = max(1, openai_keys.slot_count()) last: Optional[BaseException] = None for _ in range(attempts): try: # ★★**물리 전송 자리**다 (Codex 2026-08-31). 키 슬롯 loop **안**, # `router.completion` **직전**이라 primary→secondary 두 번 나가면 # 예약도 두 번이다. 논리 호출당 한 번 세면 그것을 못 본다. # # ★`research_calls_armed()` **밖에서는 아무 일도 안 한다** — # 팔을 안 든 호출부(scene_detail 등)는 한 글자도 안 바뀐다. reserve_current_research_call( source=f"llm_client._completion[{model}]") # ★**관측 단위를 스스로 밝힌다** (2026-09-20). 여기는 키 슬롯 # 진입이라 primary→secondary 두 번 나가면 두 번 찍힌다. 그런데 # Router 자체에 `num_retries` 가 있어 **그 아래는 안 보인다** — # 이것만 세고 「물리 전송 총수」라고 부르면 안 된다. 그래서 # `slot_entry` 로 적고 집계도 단위별로 갈라 돌려준다. _vlm_send = _record_send( # ★`llm` 이다 — 이 경로는 **텍스트 호출도 지난다**. # `vlm` 이라 부르면 「이미지 관찰 횟수」로 오독된다. kind="llm", granularity=_GRAIN_SLOT, source="llm_client._completion", model=str(model or ""), ) try: _res = binding.router.completion(**kwargs) except Exception: if _vlm_send is not None: # ★「청구 안 됨」이 아니라 **그 진입이 예외로 끝났다**. _vlm_send.failed("router.completion raised") raise if _vlm_send is not None: _vlm_send.ok() return _res except Exception as exc: # noqa: BLE001 last = exc if not openai_keys.failover_on( exc, where=f"router.completion[{model}]", attempted_slot=binding.slot, ): raise binding = _get_router_binding() assert last is not None raise last def _invalidate_router() -> None: """다음 호출에서 Router 를 다시 짓게 한다. OpenAI 키 슬롯이 바뀌면 deployment 에 박힌 api_key 가 옛 키라 그대로 두면 전환이 무의미하다. 대입은 GIL 하에서 원자적이고, 실제 재빌드는 `_get_router_binding` 의 double-checked locking 이 처리한다 — 여기서 `_init_lock` 을 잡으면 호출 중 전환과 엮여 교착이 될 수 있다. """ global _router, _binding _router = None _binding = None logger.info("LiteLLM Router 무효화 — OpenAI 키 슬롯 전환 반영") def _get_router_binding() -> "_RouterBinding": """Router 와 그 생성 슬롯을 **짝으로** 받는다. ★캐시 판정을 "객체가 있는가"가 아니라 **"그 Router 를 지은 슬롯이 아직 현재 슬롯인가"** 로 한다. 전환 훅(`_invalidate_router`)은 `_active_index` 가 바뀐 **뒤 락 밖에서** 돌기 때문에, 그 사이에 다른 스레드가 옛 Router 를 그대로 받아 갈 수 있다. 슬롯을 대조하면 그 창이 닫힌다. """ from app.core import openai_keys b = _binding if b is not None and b.slot == openai_keys.active_slot(): return b with _init_lock: b = _binding if b is not None and b.slot == openai_keys.active_slot(): return b # double-checked locking _init_opik() _build_router() built = _binding if built is None: # 계약 위반이다 — 조용히 넘기면 슬롯 없는 Router 가 돌아다닌다. raise RuntimeError("LiteLLM Router binding 이 생성되지 않았다") return built def router_completion(*, model: str, **kwargs: Any): """Router 를 거치는 **유일한** 호출 경로 — 키 슬롯 전환이 항상 걸린다. [2026-08-01 A5 마무리] 이전에는 `_get_router()` 로 원시 Router 를 꺼내 `.completion(...)` 을 직접 부르는 소비자가 넷 있었고, 그 경로에서는 1차 키가 billing 으로 죽어도 **보조 키를 한 번도 안 써 보고** 예외가 그대로 올라갔다 (직접 재현: active_slot 이 primary 그대로, Router 빌드 1회). "원시 Router 를 꺼내지 마라"는 규칙으로 두면 또 생긴다 — 실제로 네 번 생겼다. 그래서 `_get_router()` 자체를 없애고 이 함수만 남긴다. ★위 「유일한 경로」는 **원시 Router 를 꺼내지 않는다**는 뜻이지 모든 호출이 이 함수를 지난다는 뜻이 아니다. `call_structured` 등은 아래 `_completion` 을 직접 부른다 — 정지 관문이 거기 있는 이유다. """ return _completion(_get_router_binding(), model, {"model": model, **kwargs}) def _current_stop_check(): """이 스레드에 걸린 정지 확인. 순환 import 를 피해 호출 시점에 읽는다.""" try: from app.core.image_call_budget import get_current_stop_check return get_current_stop_check() except Exception: # 표를 못 읽는 것이 호출을 죽이면 안 된다 return None def _build_router() -> Router: """_get_router의 lock 내부 빌드 로직 — _init_lock를 이미 잡고 있어야 한다.""" global _router, _binding from app.core.config import settings model_list = [] # Gemini 키 풀 — 모든 Gemini 텍스트 모델에 동일 키 풀 적용 gemini_keys = _load_gemini_keys() if not gemini_keys and settings.gemini_api_key: gemini_keys = [settings.gemini_api_key] gemini_models = [ ("gemini-pro", settings.gemini_text_model), ("gemini-flash", settings.gemini_flash_model), ("gemini-lite", settings.gemini_lite_model), # gpt-mini: 2026-07-11 Gemini 원복(사용자 goal — 시각 저작 열화 실측). ("gpt-mini", settings.gemini_flash_model), ] for alias, model_id in gemini_models: for i, key in enumerate(gemini_keys): model_list.append({ "model_name": alias, "litellm_params": { "model": f"gemini/{model_id}", "api_key": key, }, "model_info": {"id": f"{alias}-key{i}"}, }) # OpenAI — GPT 5.5 / 5.5 mini / 5.5 nano + 레거시 4.1 # 주의: gpt-5.5* 는 litellm 내장 model registry에 없을 수 있어 # provider를 자동 추론하지 못한다. `openai/` 접두사를 명시해 라우팅 강제. # OpenAI 키는 슬롯 브로커가 정한다 (2026-07-30) — 1차 키가 키 수준 # 실패를 내면 보조 슬롯으로 전환되고, 그때 Router 를 무효화해 새 키로 # 다시 짓는다. 슬롯이 하나뿐이면 기존과 동일하다. from app.core import openai_keys openai_keys.register_switch_hook(_invalidate_router) # ★슬롯과 키를 한 번에 읽는다 (A5) — 따로 읽으면 그사이 전환이 둘을 # 어긋나게 해 "primary 라고 적힌 secondary 키 Router" 가 생긴다. openai_slot, openai_key = openai_keys.active_slot_and_key() if openai_key: # gpt-5* family는 temperature=1만 허용 — drop_params=True를 deployment에 직접 # 박아 호출 측 temperature를 자동 drop. 글로벌 litellm.drop_params만으로는 # router 호출 경로에서 일관되게 적용되지 않는 버전이 있어 deployment level에서 강제. _OPENAI_DROP = {"drop_params": True} # ★[2026-09-08] 추론 강도를 **deployment 에 박는다.** 호출부마다 넘기면 # 109개 스텝 중 어디 하나가 빠져도 아무도 모른다 — 기본은 여기 한 곳이 # 정하고, 필요한 스텝만 override 로 덮는다. # 기본 alias `gpt` = medium · 보조 alias = low (사용자 지시). _EFF = {"reasoning_effort": str( getattr(settings, "openai_reasoning_effort", "medium") or "medium")} _EFF_AUX = {"reasoning_effort": str( getattr(settings, "openai_reasoning_effort_aux", "low") or "low")} # ★VLM 판정 전용 슬롯 — **같은 물리 모델, 강도만 high**. # 판정 자리를 `gpt` 로 같이 쓰면 강도를 못 가른다. _EFF_JUDGE = {"reasoning_effort": str( getattr(settings, "openai_reasoning_effort_judge", "high") or "high")} model_list.append({ "model_name": "gpt", "litellm_params": { "model": f"openai/{settings.openai_model}", "api_key": openai_key, **_OPENAI_DROP, **_EFF, }, }) # 2026-07-11 Gemini 원복: gpt-mini 는 위 gemini_models(flash 매핑)로 # 복귀. gpt-terra/gpt-luna alias 는 override 안전망으로 유지. # ★[2026-09-08 사용자 지시] GPT **텍스트는 전부** gpt-6-astra 로 간다. # 보조 alias 는 판을 가르는 것이 아니라 **추론 강도를 낮게 쓰는 슬롯**이 # 됐다 — 물리 모델은 같고 effort 만 low 다. # ★레거시 `gpt-4.1*` 은 **이름 그대로 둔다.** 그 alias 를 명시로 고른 # 자리는 「옛 판으로 재현한다」는 뜻이라 조용히 최신으로 올리면 안 된다. for alt_model, alt_alias, _eff in [ (settings.openai_model, "gpt-terra", _EFF_AUX), (settings.openai_model, "gpt-luna", _EFF_AUX), (settings.openai_model, "gpt-nano", _EFF_AUX), ("gpt-4.1", "gpt-4.1", {}), ("gpt-4.1-mini", "gpt-4.1-mini", {}), # VLM 판정 alias — 고르는 자리와 결함 관찰이 둘 다 이것을 쓴다. (settings.openai_model, "gpt-high", _EFF_JUDGE), ]: model_list.append({ "model_name": alt_alias, "litellm_params": { "model": f"openai/{alt_model}", "api_key": openai_key, **_OPENAI_DROP, **_eff, }, }) # xAI(OpenRouter 경유) — VLM 판정 alias (2026-08-27 사용자 지시). # # 사용자: 「gpt sol vlm 은 성능이 안좋아 … 무조건 gemini 3.1 pro 와 # grok 최신 모델 둘을 사용해야해」 + 「grok alias 도 만들어」. # # ★**왜 alias 가 필요한가.** `multiroll_gemini` 에는 이미 grok 판정 # 경로가 있는데(`openrouter:` prefix → `ask_openrouter_structured`), # 그 prefix 를 푸는 코드가 **그 모듈 안에만** 있다. 이미지를 보내는 # 다른 여덟 자리는 전부 `call_structured` 를 쓰므로 그 경로를 못 # 쓴다. router 에 alias 로 올려야 `project_config` 로 어디서나 # 고를 수 있다. # # ★키가 없으면 등록하지 않는다 — `claude-opus` 와 같은 형태다. # 등록 안 된 alias 를 고르면 호출부가 그 자리에서 실패하므로, # 배선하는 쪽이 fail-closed 를 스스로 정해야 한다. _grok_model = (getattr(settings, "grok_judge_model", "") or "").strip() if settings.openrouter_api_key and _grok_model: model_list.append({ "model_name": "grok", "litellm_params": { "model": f"openrouter/{_grok_model}", "api_key": settings.openrouter_api_key, "api_base": settings.openrouter_base_url, # ★출력 상한을 **여기 적으면 안 된다.** `call_structured` 가 # 요청마다 `max_tokens` 를 넣고 Router 가 요청값을 배포값 # 위에 얹어, 여기 적은 값은 나가지 않는다(끝점 실측 # 2026-08-27: body 의 max_tokens = 65,536). # 상한은 [[MODEL_MAX_OUTPUT_TOKENS]] 가 잡는다. "drop_params": True, }, }) # Anthropic — 후보 선정 판정 전용 (2026-08-06 사용자 지시). # 키가 없으면 alias 를 등록하지 않는다 → 선정 판정은 `gemini-pro` 로 # 남는다(fail-open). 등록될 때만 `claude-opus` 가 실재한다. if settings.anthropic_api_key: for _alias, _mdl in ( ("claude-opus", settings.anthropic_judge_model), # 비교 실험용 — 판정 배선에는 쓰지 않는다. ("claude-fable", settings.anthropic_fable_model), ): model_list.append({ "model_name": _alias, "litellm_params": { "model": f"anthropic/{_mdl}", "api_key": settings.anthropic_api_key, # Claude 5 계열은 sampling 파라미터를 거부(400)한다 — # `_NO_TEMPERATURE_ALIASES` 와 이중 방어. "drop_params": True, }, }) # gpt-5* family는 temperature=1만 허용 등 모델별 spec 차이가 있다. # 글로벌 drop_params=True 설정으로 unsupported param을 자동 drop해 # 호출 측 코드를 단순화. (Router.__init__ 키워드는 미지원이라 글로벌로.) import litellm as _litellm _litellm.drop_params = True _router = Router( model_list=model_list, routing_strategy="simple-shuffle", num_retries=settings.llm_max_retries, timeout=settings.llm_timeout_text, retry_after=5, # 429 후 5초 대기 ) # 지은 슬롯을 Router 와 묶어 둔다 — 이후 실패는 이 슬롯의 실패다. _binding = _RouterBinding(slot=openai_slot, router=_router) # ★alias → 물리 모델. 출력 상한을 **모델 표에서** 읽으려면 이 짝이 있어야 # 한다. 손으로 적지 않는다 — Router 를 지은 그 목록이 곧 정본이다. _ALIAS_PHYSICAL.clear() for _d in model_list: _ALIAS_PHYSICAL.setdefault( str(_d["model_name"]), str(_d["litellm_params"].get("model") or "")) logger.info( "LiteLLM Router initialized: %d deployments (%d Gemini keys, " "OpenAI=%s slot=%s/%d)", len(model_list), len(gemini_keys), bool(openai_key), openai_slot or "없음", openai_keys.slot_count(), ) return _router # ── 파이프라인 단계 → 모델 매핑 (problems.md #5 통합) ── # # Single source of truth: ``app.core.step_manifest.STEP_MANIFEST``. # 이 파일의 ``PIPELINE_STEPS`` 는 manifest 의 default_model/provider/category/ # label 을 derived view 로 build 한 후, manifest 에 없는 sub_step / v2 legacy # extension 을 합친다. 이전에는 manifest 와 별도 dict 가 drift 가능했고 # (``planning_doc_analysis`` 의 alias mismatch), runtime 은 PIPELINE_STEPS 만 # 봐서 manifest 의 default_model 필드가 dead 였다. # # Extension table (manifest 미등록): # - scene_image_pipeline 의 sub_step 8종 (prompt_translation 등) # - v2 legacy step 7종 (entity_extract / entity_style / scene_dependency / # location_consistency / outlook_merge / webbook_gen / style_rules / # entity_detail_batch) # 이들은 step_runner 흐름과 별개로 LLM call routing 이 필요한 sub-routine. # manifest 미등록 step 의 model alias 매핑 (extension table). _PIPELINE_STEP_EXTENSIONS: Dict[str, Dict[str, str]] = { # scene_image_pipeline sub_steps "prompt_translation": {"label": "T2I 프롬프트 번역", "default": "gpt-mini", "category": "image_sub"}, "scene_t2i_gen": {"label": "T2I 이미지 생성", "default": "gemini-image", "category": "image_sub"}, "scene_t2i_validation": {"label": "이미지 검증", "default": "gpt", "category": "image_sub"}, "prompt_sanitize": {"label": "프롬프트 안전화", "default": "gpt", "category": "image_sub"}, "angle_recommend": {"label": "앵글 추천", "default": "gpt", "category": "image_sub"}, "fal_angle_apply": {"label": "fal.ai 앵글 적용", "default": "fal-ai", "category": "image_sub"}, "final_select": {"label": "최종 선택", "default": "gpt", "category": "image_sub"}, "t2i_translation": {"label": "T2I 편집 번역", "default": "gpt-mini", "category": "image_sub"}, # GROUNDING-V2 §2-2 — 고증 후보 분류. 초기 판정자는 **Sol** (사용자 지시). # ★등록하지 않으면 _resolve_model 이 gemini-pro 로 fallback 한다 — 이름만 # 적어 둔다고 Sol 로 가지 않는다 (Codex 실측). "grounding_classify": {"label": "고증 후보 분류", "default": "gpt", "category": "analysis"}, # ★A0(`grounding_a0`) 는 **여기 없다** — STEP_MANIFEST 의 정식 스텝이 됐고 # manifest 가 extension 보다 우선이라 여기 두면 죽은 정의가 된다. # 모델은 `step_manifest.py` 의 `default_model: "gpt"` 가 정한다. # G3.2 judge — scene_detail post-parse owned validation # round 4 MINOR 1 / round 5 M2: 등록 키는 'default' (NOT 'default_model'). # 1-call no retry (round 4 Q1=B / round 5 M1) — caller `run_owned_judge` 가 # violations 발견 시 contract_violation status 마킹만 수행. "scene_detail_owned_judge": {"label": "owned 객체 redraw 검사", "default": "gpt-mini", "category": "analysis_sub"}, # FINDING 7 (e2e-bughunt-v1) — owned redraw violation 1-call repair. # owned judge 와 동일 tier (analysis_sub). caller `_attempt_owned_redraw_repair` # 가 max 1 attempt 관리 + 결과를 기존 validator 로 재검증. "scene_detail_owned_repair": {"label": "owned 객체 redraw 수정", "default": "gpt-mini", "category": "analysis_sub"}, # W-M (2026-07-03) — 야외 같은 장소 그룹 plate 생성 순서 결정(단계화 체인). # caller = background_render 그룹 체인 lane. 검증(순열+depends_on)/재시도/ # fallback 은 location_aerial.decide_group_plate_order 가 담당. "background_plate_order": {"label": "야외 plate 생성 순서", "default": "gpt-mini", "category": "image_sub"}, # v2 legacy (기존 코드 호환용 — manifest 제외, runtime 라벨 유지) "entity_extract": {"label": "요소 추출 (3턴)", "default": "gpt", "category": "analysis"}, "entity_style": {"label": "요소 추출 — 스타일+이름", "default": "gpt", "category": "analysis"}, "entity_detail_batch": {"label": "요소 추출 — 상세", "default": "gpt", "category": "analysis"}, "webbook_gen": {"label": "웹북 패키지 생성", "default": "gpt", "category": "auxiliary"}, "style_rules": {"label": "스타일 규칙 생성", "default": "gpt", "category": "analysis"}, "outlook_merge": {"label": "아웃룩 병합", "default": "gpt", "category": "analysis"}, "location_consistency": {"label": "Location 외형 고정", "default": "gemini-pro", "category": "analysis"}, } def _build_pipeline_steps() -> Dict[str, Dict[str, str]]: """STEP_MANIFEST + extension 을 합쳐 ``PIPELINE_STEPS`` view 를 build. manifest 의 ``default_model`` / ``provider`` / ``category`` / ``label`` 을 꺼내 기존 PIPELINE_STEPS schema (``label`` / ``default`` / ``category``) 로 변환. 동일 step 이 양쪽에 있으면 manifest 우선 (problems.md #5: manifest is single source). """ from app.core.step_manifest import STEP_MANIFEST out: Dict[str, Dict[str, str]] = {} for sid, info in STEP_MANIFEST.items(): out[sid] = { "label": info.get("label", sid), "default": info.get("default_model", "gemini-pro"), "category": info.get("category", "analysis"), } # extension 은 manifest 에 없는 step 만 추가 (manifest 가 single source). # overlap 시 logger.warning + extension 무시 — 회귀 가드 보강 (review M1). for sid, info in _PIPELINE_STEP_EXTENSIONS.items(): if sid in out: logger.warning( "_PIPELINE_STEP_EXTENSIONS overlap with STEP_MANIFEST: %s — " "manifest wins (extension entry is dead and should be removed).", sid, ) continue out[sid] = dict(info) return out PIPELINE_STEPS = _build_pipeline_steps() # 사용 가능한 모델 별칭 (UI용) AVAILABLE_MODELS = [ # ★2026-09-08: 넷 다 물리 모델은 `gpt-6-astra` 하나고 **추론 강도만** 다르다 # (gpt=medium · terra/luna/nano=low). 라벨 괄호가 그 강도다. {"alias": "gpt", "label": "GPT-6 Astra (medium)", "provider": "openai"}, {"alias": "gpt-terra", "label": "GPT-6 Astra (low)", "provider": "openai"}, {"alias": "gpt-luna", "label": "GPT-6 Astra (low)", "provider": "openai"}, {"alias": "gpt-mini", "label": "Gemini 3.8 Flash", "provider": "gemini"}, {"alias": "gpt-nano", "label": "GPT-6 Astra (low)", "provider": "openai"}, {"alias": "gemini-pro", "label": "Gemini 3.1 Pro", "provider": "gemini"}, {"alias": "gemini-flash", "label": "Gemini 3.8 Flash", "provider": "gemini"}, {"alias": "gemini-lite", "label": "Gemini 3.8 Flash", "provider": "gemini"}, {"alias": "gpt-4.1", "label": "GPT-4.1", "provider": "openai"}, {"alias": "gpt-4.1-mini", "label": "GPT-4.1 Mini", "provider": "openai"}, # ANTHROPIC_API_KEY 가 있을 때만 Router 에 실재한다 (_build_router). {"alias": "claude-opus", "label": "Claude Opus 5", "provider": "anthropic"}, {"alias": "claude-fable", "label": "Claude Fable 5", "provider": "anthropic"}, # OPENROUTER_API_KEY + GROK_JUDGE_MODEL 이 있을 때만 Router 에 실재한다. # ★사람이 화면에서 고를 수는 **없다** — `LLMConfigPanel` 의 # `available_models` 타입이 `{openai, gemini}` 둘뿐이라 `xai` 묶음은 # select 에 안 뜬다(anthropic 도 같다). 이 줄이 있어야 하는 이유는 # 다른 데 있다: `projects.py:570` 이 이 목록을 **alias→provider 의 # SOT** 로 읽는다. 없으면 `startswith("gpt")` 추론으로 떨어져 grok 이 # 「gemini」로 잘못 표시된다. {"alias": "grok", "label": "Grok 4.6", "provider": "xai"}, ] _UNKNOWN_STEP_WARNED: set[str] = set() def _resolve_model(step: str, project_config: Optional[Dict] = None) -> str: """단계 + 프로젝트 설정 → LiteLLM Router 모델 별칭 반환. PIPELINE_STEPS (manifest+extension) 에 없는 step 은 ``gemini-pro`` 로 silent fallback. typo 등을 surface 하기 위해 process 당 1회 logger.warning emit (review I2). project_config 가 명시한 step 은 fallback 대상 아님. """ if project_config and step in project_config: return project_config[step].get("model", PIPELINE_STEPS.get(step, {}).get("default", "gemini-pro")) info = PIPELINE_STEPS.get(step) if info is None and step not in _UNKNOWN_STEP_WARNED: _UNKNOWN_STEP_WARNED.add(step) logger.warning( "_resolve_model: unknown step '%s' — falling back to 'gemini-pro'. " "If intentional, register in STEP_MANIFEST or _PIPELINE_STEP_EXTENSIONS.", step, ) return (info or {}).get("default", "gemini-pro") # temperature를 강제로 제거해야 하는 모델 alias. # - gpt-5* family: temperature=1만 허용 (그 외 값은 OpenAI/litellm이 거부) # - gemini-3 family (pro/flash/lite): temperature<1.0이면 "infinite loops, degraded reasoning, # failure on complex tasks" warning. 빈 응답으로 schema 검증 실패하는 사례 발생 # (예: scene_camera_flow S25에서 gemini-pro가 empty content 반환). # LiteLLM 경고 그대로 따라 temperature=1.0 default 사용. # - claude-opus (Claude Opus 5): `temperature`/`top_p`/`top_k` 를 **거부**한다 # (400 invalid_request_error). call_structured 의 기본 temperature=0.2 가 # 그대로 나가면 전 호출이 실패하므로 제거 대상 필수. _NO_TEMPERATURE_ALIASES = frozenset({ # ★`gpt-high` 를 빼먹었었다 (Codex NON-BLOCK 2026-09-08). Router 에만 # 등록하면 이 공통 처리들이 **따라오지 않는다** — 같은 물리 모델인데 # 한 별칭만 temperature 를 달고 나간다. "gpt", "gpt-high", "gpt-terra", "gpt-luna", "gpt-mini", "gpt-nano", "gemini-pro", "gemini-flash", "gemini-lite", "claude-opus", "claude-fable", }) # gpt-mini 는 Gemini 매핑이지만 temperature 제거 대상 유지 (양쪽 무해) def _sanitize_kwargs_for_model(model: str, kwargs: Dict[str, Any]) -> None: """모델별로 전달 불가 파라미터를 in-place 제거. drop_params 글로벌/deployment 설정이 일관되게 적용되지 않는 버전이 있어 명시 제거. """ if model in _NO_TEMPERATURE_ALIASES: kwargs.pop("temperature", None) cap = _provider_max_output(model) if cap is not None: want = kwargs.get("max_tokens") kwargs["max_tokens"] = cap if want is None else min(int(want), cap) # ★★★OpenAI 추론 모델은 `max_tokens` 를 **거부한다** (2026-09-08 실측, # `gpt-6-astra` 로 모델을 올린 뒤 `visual_world_rules` 가 400 으로 죽었다): # OpenAIException - Unsupported parameter: 'max_tokens' is not # supported with this model. Use 'max_completion_tokens' instead. # 앞 모델(`gpt-5.6-*`)은 받아 줘서 이 자리가 필요 없었다. 모델을 갈아 # 끼우면 **그 값을 옮기는 줄**도 같이 봐야 한다는 사례다. # ★gpt-4.1 계열은 `_OPENAI_ALIASES` 에 없으므로 그대로 `max_tokens` 를 # 쓴다. Gemini·Anthropic·grok 도 건드리지 않는다. if model in _OPENAI_ALIASES and "max_tokens" in kwargs: want = kwargs.pop("max_tokens") # 둘 다 실려 오면 provider 가 거부한다 — 이미 있으면 작은 쪽을 남긴다. have = kwargs.get("max_completion_tokens") kwargs["max_completion_tokens"] = ( want if have is None else min(int(have), int(want))) # alias 별 출력 상한 — 없으면 `llm_max_output_tokens`(65,536). # # ★**`litellm_params` 에 적은 `max_tokens` 는 실효 상한이 아니다** # (2026-08-27 Codex 지적, 끝점 실측으로 확인). `call_structured` 가 요청마다 # `max_tokens` 를 넣고(`:1058`) Router 가 요청값을 배포값 위에 얹는다 — # 나가는 body 에 실제로 실린 값은 **65,536** 이었다. # # ★왜 grok 만 상한을 두나: 추론 모델이라 출력 과금에 사고 토큰이 함께 # 들어간다. 실측 완성 토큰이 평균 3,532 · 최대 7,337 인데 상한이 65,536 # 이면 한 콜이 그 아홉 배까지 열려 있다(출력 $6/M 기준 콜당 최대 $0.39). # # ★재는 자리는 **나가는 kwargs** 다. Router 설정값을 재면 이 결함을 # 못 잡는다 — 그것이 이 상수가 생긴 이유다. MODEL_MAX_OUTPUT_TOKENS: Dict[str, int] = {"grok": 8000} def _provider_max_output(model: str) -> Optional[int]: """그 alias 의 **물리 모델이 실제로 허용하는** 출력 상한. ★[2026-09-09] 왜 필요한가 — 전역 하나(`llm_max_output_tokens`)로 모든 모델을 재던 판에서 두 방향으로 틀렸다: gpt-6-astra 최대 128,000 인데 **65,536 만** 요청했다 (절반을 버렸다) gpt-4.1 최대 32,768 인데 65,536 을 요청했다 (넘겨서 보냈다) 그래서 모델 표에서 읽어 **모델마다 제 최대**를 쓴다. 손으로 적으면 모델을 갈아 끼울 때 또 어긋난다 — 값의 주인은 provider 다. ★`MODEL_MAX_OUTPUT_TOKENS` 의 명시 값이 있으면 **그쪽이 이긴다.** grok 8,000 처럼 우리가 비용 때문에 일부러 좁힌 것이 있다. """ if model in MODEL_MAX_OUTPUT_TOKENS: return MODEL_MAX_OUTPUT_TOKENS[model] try: import litellm dep = _ALIAS_PHYSICAL.get(model) if not dep: return None info = litellm.get_model_info(dep) v = info.get("max_output_tokens") return int(v) if v else None except Exception: # 모르면 기존 동작 그대로 — 여기서 세우지 않는다 return None #: alias → 물리 모델. Router 를 지을 때 채운다(`_build_router`). _ALIAS_PHYSICAL: Dict[str, str] = {} # Gemini 모델 alias 집합 — safety_settings 강제 적용 대상. # gpt-mini: 2026-07-11 Gemini 원복 — flash 매핑 복귀로 재편입. _GEMINI_ALIASES = frozenset({"gemini-pro", "gemini-flash", "gemini-lite", "gpt-mini"}) # Gemini safety filter BLOCK_NONE (전 카테고리) — 픽션 시나리오(영화/드라마) 분석용. # 시나리오는 폭력/사망/성적 묘사 등 fictional 서사 요소를 포함하므로 # 기본 safety threshold(BLOCK_MEDIUM_AND_ABOVE)에서 content_filter trip 빈발. # entity_character_list (PID 0bb48ebf, 2026-05-01) 5회 연속 finish_reason='content_filter' 사례. _GEMINI_SAFETY_SETTINGS_OFF = [ {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"}, {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"}, {"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_NONE"}, {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_NONE"}, ] def _apply_gemini_safety(model: str, kwargs: Dict[str, Any]) -> None: """Gemini 모델 호출에 safety_settings BLOCK_NONE 강제 — 픽션 콘텐츠 분석 차단 회피.""" if model in _GEMINI_ALIASES: kwargs["safety_settings"] = _GEMINI_SAFETY_SETTINGS_OFF # OpenAI 계열 model alias (strict structured output 호출 대상). # 2026-07-11 Gemini 원복: gpt-mini 제외(Gemini flash 매핑 복귀). _OPENAI_ALIASES = frozenset({"gpt", "gpt-high", "gpt-terra", "gpt-luna", "gpt-nano"}) # ★`gpt-high` 는 `AVAILABLE_MODELS` 에는 **일부러 안 넣는다.** 그 목록은 # 화면에서 고를 수 있는 것과 alias→provider 의 SOT 를 겸하는데, 이 별칭은 # 단계 모델이 아니라 VLM 판정이 코드에서 직접 쓰는 것이다(`multiroll_gemini`). # `_resolve_model` 이 이 값을 돌려주는 길이 없어 SOT 로도 필요 없고, # 넣으면 사람이 고를 수 있는 단계 모델처럼 보인다. # OpenAI strict structured output (response_format json_schema) 이 거부하는 # jsonschema keyword. FINDING 10: scene_detail detail_schema 의 # t2i_variations[].reference_phrase_kinds.uniqueItems 가 GPT fallback 경로에서 # BadRequestError("'uniqueItems' is not permitted") 를 유발. scope = uniqueItems 한정. _OPENAI_UNSUPPORTED_SCHEMA_KEYS = frozenset({"uniqueItems"}) def _strip_schema_keys(node: Any, keys: frozenset) -> Any: """node 트리의 모든 중첩 dict 에서 keys 를 제거한 deep copy 반환. 원본 node 는 mutate 하지 않는다 — dict/list 를 새로 재구성한다. """ if isinstance(node, dict): return { k: _strip_schema_keys(v, keys) for k, v in node.items() if k not in keys } if isinstance(node, list): return [_strip_schema_keys(v, keys) for v in node] return node def _is_openai_strict_compatible(node: Any) -> bool: """OpenAI strict 의 known object-contract 호환성 재귀 검사 (완전 판정 아님). 검사 범위는 이번에 실측된 object 계약(required=전 property 키 + additionalProperties:false)에 한정 — strict 의 전체 keyword subset 을 판정하지 않는다 (uniqueItems 는 별도 strip). properties={} 빈 object 검사는 후속 hardening 항목 (Codex 비차단 권고 2026-07-10). strict=True 는 모든 object 노드에 대해 ①`required` 가 properties 의 전 키를 포함하고 ②`additionalProperties: false` 명시를 요구한다. Gemini 시절 저작 팩 스키마(optional 필드 관용)는 이 규칙을 어겨 BadRequestError ("'required' is required to be supplied and to be an array including every key in properties") 가 난다 — GPT-5.6 이관 2회차 E2E shot_extract 30/30 실측 (2026-07-10). """ if isinstance(node, dict): props = node.get("properties") if isinstance(props, dict) and props: req = node.get("required") if not isinstance(req, list) or set(req) != set(props.keys()): return False if node.get("additionalProperties") is not False: return False return all(_is_openai_strict_compatible(v) for v in node.values()) if isinstance(node, list): return all(_is_openai_strict_compatible(v) for v in node) return True _STRICT_DOWNGRADE_LOGGED: set[str] = set() def _drop_non_string_enums(node: Any, dropped: List[str]) -> Any: """`enum` 값이 문자열이 아닌 노드에서 `enum` 만 걷은 deep copy. ★Gemini `response_schema` 의 `enum` 은 **문자열 배열**이다. 정수 enum 을 보내면 400 이 난다 (2026-08-28 실측, `gemini-3.1-pro-preview`): Invalid value at '…properties[0].value.enum[0]' (TYPE_STRING), 1 정수 enum 을 주입하는 자리는 **셋**이다(`app/` 전 트리 AST 로 셌다): · `beat_shot_steps.py:226` — `beat_extract` 의 `scene_index` · `beat_shot_steps.py:529` — `shot_extract` 의 `scene_index` · `scene_camera_flow_step.py:240` — `scene_camera_flow` 의 `shot_index` ★셋째 자리가 더 고약하다 — 그 스텝은 예외를 **빈 flow 로 삼켜서** 실패가 안 보이고 품질만 조용히 내려앉는다. 앞 둘은 스텝이 죽어 주행이 멎으니 오히려 드러난다. 08-24 주행에는 통했는데 이번엔 400 이라 **provider 쪽이 조인 것**으로 본다(표본 하나라 단정은 아니다). ★왜 문자열로 바꾸지 않고 **걷나**: Gemini 의 enum 은 `type: STRING` 전용 이라 정수를 문자열로 바꿔 보내면 모델이 문자열로 돌려줄 수 있고, 그러면 `set(returned_indices) != set(expected_indices)` 대조가 통째로 어긋난다. enum 은 **안내**이고 계약은 뒤에서 따로 지킨다 — 반환 인덱스를 기대값과 대조해 순서로 재매핑하고, 안 맞으면 재시도하고, 빠진 씬은 빈 beat 로 채운다(`beat_shot_steps.py:248-280`). 안내 하나를 잃고 스텝을 살린다. `type` 은 그대로 둔다 — 정수를 달라는 요구는 남는다. """ if isinstance(node, dict): out = {} for k, v in node.items(): if (k == "enum" and isinstance(v, list) and v and not all(isinstance(x, str) for x in v)): dropped.append(str(node.get("type", "?"))) continue out[k] = _drop_non_string_enums(v, dropped) return out if isinstance(node, list): return [_drop_non_string_enums(v, dropped) for v in node] return node _GEMINI_ENUM_DROP_LOGGED: set[str] = set() def _sanitize_response_format_for_model(model: str, kwargs: Dict[str, Any]) -> None: """OpenAI 계열 model 호출 시 response_format schema 의 OpenAI 비호환 jsonschema keyword 를 제거 (FINDING 10 — provider-boundary fix). OpenAI strict structured output 은 `uniqueItems` 를 거부한다. Gemini 경로는 허용하므로 정상 동작하지만 GPT fallback 경로에서 BadRequestError 가 난다. `kwargs["response_format"]["json_schema"]["schema"]` 를 deep-copy + strip 한 새 객체로 교체한다 — caller 의 원본 response_schema 는 mutate 하지 않으므로 `_validate_local_schema` 의 local 검증은 원본 schema(uniqueItems 포함)로 그대로 수행된다. 비-OpenAI(Gemini 등) model 은 no-op. GPT-5.6 이관 (2026-07-10): strict 비호환 스키마(Gemini 팩 유래 — optional 필드/additionalProperties 미명시)는 strict=False 로 강등한다. 스키마 준수는 `_validate_local_schema` + call_structured retry 가 전 tier 에서 이미 보증(Gemini 경로와 동일한 enforcement 모델). strict 호환 스키마 (기존 gpt 스텝 팩)는 strict=True 그대로 — byte-identical. 팩 수정 없이 provider boundary 에서 해소 (프롬프트 팩 덮어쓰기 금지 준수). """ response_format = kwargs.get("response_format") if not isinstance(response_format, dict): return json_schema = response_format.get("json_schema") if not isinstance(json_schema, dict) or "schema" not in json_schema: return # Gemini: 문자열이 아닌 `enum` 은 400 을 낸다 — 경계에서 걷는다 # (사유는 `_drop_non_string_enums` 독스트링). caller 의 원본 스키마는 # 그대로라 `_validate_local_schema` 는 여전히 원본으로 검증한다. if model in _GEMINI_ALIASES: dropped: List[str] = [] json_schema["schema"] = _drop_non_string_enums( json_schema["schema"], dropped) name = str(json_schema.get("name", "?")) if dropped and name not in _GEMINI_ENUM_DROP_LOGGED: _GEMINI_ENUM_DROP_LOGGED.add(name) logger.info( "response_format '%s': Gemini 는 문자열 enum 만 받는다 — " "%s 형 enum %d개 걷음. 계약은 호출부 검증·재시도가 지킨다 " "(process 당 1회 로그)", name, ",".join(sorted(set(dropped))), len(dropped), ) return if model not in _OPENAI_ALIASES: return json_schema["schema"] = _strip_schema_keys( json_schema["schema"], _OPENAI_UNSUPPORTED_SCHEMA_KEYS, ) if json_schema.get("strict") and not _is_openai_strict_compatible( json_schema["schema"]): json_schema["strict"] = False name = str(json_schema.get("name", "?")) if name not in _STRICT_DOWNGRADE_LOGGED: _STRICT_DOWNGRADE_LOGGED.add(name) logger.info( "response_format '%s': OpenAI strict 비호환 스키마(Gemini 팩 " "유래) — strict=False 강등, 준수는 local validation+retry 가 " "보증 (process 당 1회 로그)", name, ) # ── 통합 호출 함수 ── def _fill_usage_sink( sink: Optional[Dict[str, Any]], response: Any, *, alias: str, max_tokens: Any = None, ) -> None: """토큰·비용·물리 모델을 호출자에게 돌려준다 — **주면 채우고 안 주면 안 한다.** ★왜 필요한가 (2026-08-27, #92). `call_structured` 는 payload 만 돌려줘 호출자가 **한 판정에 얼마를 썼는지 알 길이 없다.** 두 모델을 부르는 판에서는 모델별 비용을 나눠 적어야 하는데, 그러려면 응답이 들고 온 값을 그 자리에서 받아야 한다. ★**미보고를 0 으로 쓰지 않는다.** provider 가 usage 를 안 주면 키를 비워 둔다 — 0 으로 채우면 「안 썼다」로 읽힌다. 내가 그 함정을 이미 한 번 밟았다(`cost` 가 딕셔너리인데 스칼라로 읽어 0 이 나왔고 그것을 「기록이 없다」로 보고했다). Args: sink: 채울 딕셔너리. `None` 이면 아무것도 안 한다(기존과 동일). response: litellm 응답 객체. alias: Router alias (`grok`·`gemini-pro` …). max_tokens: 그 요청에 실제로 실린 상한 — 잘림을 나중에 가르려면 필요하다. """ if sink is None: return sink["alias"] = alias if max_tokens is not None: sink["max_tokens_sent"] = max_tokens # 물리 모델은 응답이 말하는 것이 진실이다 — 설정이 아니라. physical = getattr(response, "model", None) if physical: sink["physical_model"] = physical usage = getattr(response, "usage", None) if usage is not None: for key in ("prompt_tokens", "completion_tokens", "total_tokens"): v = getattr(usage, key, None) if v is None and isinstance(usage, dict): v = usage.get(key) if v is not None: sink[key] = v # litellm 은 계산 비용을 `_hidden_params` 에 넣는다. 없으면 **안 넣는다.** hidden = getattr(response, "_hidden_params", None) if isinstance(hidden, dict): cost = hidden.get("response_cost") if cost is not None: sink["estimated_cost_usd"] = cost def _add_fallback_tag(metadata: Dict[str, Any], tag: str) -> Dict[str, Any]: """Opik metadata에 fallback tag(sanitized/gpt_fallback)를 추가한 새 dict 반환. 원본 dict 변경 없음. opik.tags가 없으면 새로 만든다. ★v2 에서는 `status:` 축을 붙인다 (2026-08-24 Codex 재리뷰). 이 자리가 `_build_opik_metadata` 의 축 정규화 **뒤**라, 접두사 없이 붙이면 앞에서 판 축이 여기서 다시 섞인다 — 그리고 그 span 은 안전 sanitize·GPT fallback 이라 정작 훑을 값이 큰 쪽이다. 설정 OFF 면 맨 이름 그대로다. """ new_meta = dict(metadata) if metadata else {} opik_meta = dict(new_meta.get("opik") or {}) tags = list(opik_meta.get("tags") or []) try: from app.core.config import settings if getattr(settings, "opik_trace_v2_enabled", False): from app.modules.llm.opik_trace import is_axis_tag if not is_axis_tag(tag): tag = f"status:{tag}" except Exception as exc: # noqa: BLE001 — 기록은 본 작업을 안 막는다 logger.debug("_add_fallback_tag v2 축 부착 실패 (non-fatal): %s", exc) if tag not in tags: tags.append(tag) opik_meta["tags"] = tags new_meta["opik"] = opik_meta return new_meta def call_structured( step: str, system_prompt: str, user_prompt: "str | list", response_schema: Dict[str, Any], project_config: Optional[Dict] = None, schema_name: str = "response", opik_metadata: Optional[Dict] = None, temperature: float = 0.2, max_tokens: Optional[int] = None, *, enable_fallback: bool = True, validate_response: Optional[Callable[[Dict[str, Any]], bool]] = None, usage_sink: Optional[Dict[str, Any]] = None, num_retries: Optional[int] = None, timeout: Optional[float] = None, ) -> Dict[str, Any]: """Structured JSON output 호출 — LiteLLM Router 경유. ★``timeout`` — 이 호출만의 대기 상한(초). ``None`` 이면 Router 가 지어질 때 박힌 ``settings.llm_timeout_text`` 를 그대로 쓴다(기존 호출 불변). **전역을 올리지 않는다** — 긴 대본 한 자리 때문에 모든 글 스텝이 오래 기다리게 만들면 진짜로 걸린 호출도 그만큼 늦게 드러난다 (Codex 2026-09-09). 모든 provider에 대해 response_format으로 JSON schema 강제. LiteLLM이 provider별 변환 자동 처리. user_prompt: str 또는 multimodal content list (PDF/이미지 포함 시). 3-tier fallback (enable_fallback=True 시 자동, default): Tier 1: 기본 모델 (gemini-pro 등 step 기본). Tier 2: sanitize_for_safety + SAFETY_SYSTEM_SUFFIX prepend. Tier 3: GPT 강제 (project_config[step] = {"model": "gpt"}). enable_fallback=False (legacy 호환): 1차 호출만 수행, 실패 시 즉시 raise. `validate_response` (P2-3): Tier 1/2 응답이 valid JSON이지만 의미상 빈 결과 (예: 빈 list)일 때 caller가 fallback을 강제 트리거할 수 있는 callback. callback이 False 반환 → `EmptySemanticResponseError` raise → safety 분류 → Tier 2/3 진행. None이면 schema 검증만 통과해도 즉시 반환 (기본 동작). 이 callback 은 Tier 3에는 적용 안 함 (마지막 시도 보호). Local jsonschema 검증 (problems.md #13): ``_do_call`` 안에서 ``jsonschema. validate`` 가 **모든 tier 에서** 동작하여 provider strict mode 의 enforcement 약화를 보완한다. ``validate_response`` callback (의미적 빈 결과) 와 별개의 레이어 — Tier 3 도 schema 위반 시 ``SchemaValidationError`` 가 caller 까지 전파된다. ENV ``LLM_LOCAL_SCHEMA_VALIDATE=false`` 로 즉시 disable 가능. fallback 트리거 분류 (`is_safety_related_error`): - 콘텐츠 안전/모더레이션 신호 (PROHIBITED/content_filter/empty/...): 진행 - rate-limit/timeout/auth 등 명시적 transient: 즉시 raise (비용 보호) """ from app.modules.llm.safety import ( SAFETY_SYSTEM_SUFFIX, EmptySemanticResponseError, is_safety_related_error, sanitize_for_safety, ) def _do_call( sys_p: str, user_p: "str | list", cfg: Optional[Dict], suffix_tag: str = "", ) -> Dict[str, Any]: binding = _get_router_binding() model = _resolve_model(step, cfg) base_metadata = _build_opik_metadata(step, opik_metadata) metadata = _add_fallback_tag(base_metadata, suffix_tag) if suffix_tag else base_metadata kwargs: Dict[str, Any] = { "model": model, "messages": [ {"role": "system", "content": sys_p}, {"role": "user", "content": user_p}, ], "response_format": { "type": "json_schema", "json_schema": { "name": schema_name, "schema": response_schema, "strict": True, }, }, "temperature": temperature, "metadata": metadata, } from app.core.config import settings kwargs["max_tokens"] = max_tokens if max_tokens is not None else settings.llm_max_output_tokens # ★★**`enable_fallback=False` 는 Router 재시도를 안 막는다** # (2026-08-27 Codex BLOCK). 그 인자는 이 파일의 Tier 2/3 만 닫고, # Router 자체는 `num_retries=settings.llm_max_retries`(기본 3)로 # 지어진다(`:648`) — 한 alias 가 **최대 4번 전송**될 수 있다. # 게다가 `usage_sink` 는 **마지막 응답만** 보므로 앞선 과금 시도가 # 기록에서 사라진다. # 「모델마다 한 번씩」이 계약인 자리는 여기로 0 을 내려보낸다. if num_retries is not None: kwargs["num_retries"] = int(num_retries) # ★요청값이 Router 배포값을 덮는다. 안 실으면 배포에 박힌 값이 남는다. if timeout is not None: kwargs["timeout"] = float(timeout) _sanitize_kwargs_for_model(model, kwargs) _sanitize_response_format_for_model(model, kwargs) _apply_gemini_safety(model, kwargs) response = _completion(binding, model, kwargs) _fill_usage_sink(usage_sink, response, alias=model, max_tokens=kwargs.get("max_tokens")) content = response.choices[0].message.content if not content: # ★이유 칸을 같이 싣는다 (2026-09-19). 이 칸 없이 빈 응답을 # 「검열」로 읽은 적이 있다 — 확인되지 않은 원인이었다. raise EmptyLLMResponse( f"LLM returned empty response for step={step}, model={model}, " f"finish_reason=" f"{getattr(response.choices[0], 'finish_reason', None)!r}") payload = _unwrap_tool_envelope(json.loads(content), response_schema) _validate_local_schema( payload, response_schema, step=step, schema_name=schema_name, ) return payload def _validate_or_raise(result: Dict[str, Any], tier_label: str) -> Dict[str, Any]: """validate_response callback 적용. False 반환 시 EmptySemanticResponseError raise.""" if validate_response is not None and not validate_response(result): raise EmptySemanticResponseError( f"validate_response failed for step={step} tier={tier_label} — " f"semantic empty result" ) return result # Tier 1: 기본 try: result = _do_call(system_prompt, user_prompt, project_config) return _validate_or_raise(result, "tier1") except Exception as exc_t1: if not enable_fallback or not is_safety_related_error(exc_t1): raise logger.warning( "call_structured[%s] Tier 1 failed (%s), trying sanitized input", step, exc_t1, ) # Tier 2: sanitize + 영화 프레이밍 system suffix sanitized_user = sanitize_for_safety(user_prompt) safe_system = system_prompt + SAFETY_SYSTEM_SUFFIX try: result = _do_call(safe_system, sanitized_user, project_config, suffix_tag="sanitized") return _validate_or_raise(result, "tier2") except Exception as exc_t2: logger.warning( "call_structured[%s] Tier 2 sanitized failed (%s), trying GPT fallback", step, exc_t2, ) # Tier 3: GPT fallback (Gemini → GPT, sanitize 유지) # validate_response (의미적 빈 결과 callback) 는 Tier 3에 적용하지 않음 — # 마지막 시도이므로 결과를 보존. 단, _do_call 안의 local jsonschema validation # 은 Tier 3 에서도 동작하므로 schema 위반은 SchemaValidationError 로 raise. gpt_config = dict(project_config) if project_config else {} gpt_config[step] = {"model": "gpt"} return _do_call(safe_system, sanitized_user, gpt_config, suffix_tag="gpt_fallback") def call_text( step: str, system_prompt: str, user_prompt: str, project_config: Optional[Dict] = None, opik_metadata: Optional[Dict] = None, temperature: float = 0.2, *, enable_fallback: bool = True, ) -> str: """Free-text 호출 — LiteLLM Router 경유. 3-tier fallback (call_structured와 동일 정책)을 자동 적용한다. `enable_fallback=False` (legacy 호환): - Tier 1만 수행. 빈 응답은 `""` 반환 (raise 안 함) — pre-fallback 시기 동작 보존. - 그 외 예외는 그대로 raise. `enable_fallback=True` (default): - 빈 응답을 RuntimeError로 변환 → safety 분류 → Tier 2/3 진행. - rate-limit/timeout 등 transient는 즉시 raise (비용 보호). """ from app.modules.llm.safety import ( SAFETY_SYSTEM_SUFFIX, is_safety_related_error, sanitize_for_safety, ) def _do_call(sys_p: str, user_p: str, cfg: Optional[Dict], suffix_tag: str = "") -> str: """완료된 응답 content 반환. 빈 응답이라도 raise 없이 그대로 (caller가 처리).""" binding = _get_router_binding() model = _resolve_model(step, cfg) base_metadata = _build_opik_metadata(step, opik_metadata) metadata = _add_fallback_tag(base_metadata, suffix_tag) if suffix_tag else base_metadata from app.core.config import settings text_kwargs: Dict[str, Any] = { "model": model, "messages": [ {"role": "system", "content": sys_p}, {"role": "user", "content": user_p}, ], "temperature": temperature, "max_tokens": settings.llm_max_output_tokens, "metadata": metadata, } _sanitize_kwargs_for_model(model, text_kwargs) _apply_gemini_safety(model, text_kwargs) response = _completion(binding, model, text_kwargs) return response.choices[0].message.content or "" # Tier 1: 기본 try: content = _do_call(system_prompt, user_prompt, project_config) except Exception as exc_t1: if not enable_fallback or not is_safety_related_error(exc_t1): raise logger.warning( "call_text[%s] Tier 1 raised (%s), trying sanitized input", step, exc_t1, ) else: if content: return content # 빈 응답: enable_fallback=False면 legacy 동작 (빈 문자열) 보존. if not enable_fallback: return "" # enable_fallback=True: Tier 2 진행 (safety 가능성) logger.warning( "call_text[%s] Tier 1 returned empty, trying sanitized input", step, ) # Tier 2: sanitize + 영화 프레이밍 system suffix sanitized_user = sanitize_for_safety(user_prompt) safe_system = system_prompt + SAFETY_SYSTEM_SUFFIX try: content = _do_call(safe_system, sanitized_user, project_config, suffix_tag="sanitized") except Exception as exc_t2: logger.warning( "call_text[%s] Tier 2 sanitized failed (%s), trying GPT fallback", step, exc_t2, ) else: if content: return content logger.warning( "call_text[%s] Tier 2 returned empty, trying GPT fallback", step, ) # Tier 3: GPT fallback gpt_config = dict(project_config) if project_config else {} gpt_config[step] = {"model": "gpt"} content = _do_call(safe_system, sanitized_user, gpt_config, suffix_tag="gpt_fallback") if not content: # 마지막 시도까지 빈 응답 — 명시적으로 raise (caller가 인지) raise RuntimeError( f"LLM returned empty text response for step={step} after all 3 tiers" ) return content def call_multiturn( step: str, messages: List[Dict[str, str]], response_schema: Optional[Dict[str, Any]] = None, project_config: Optional[Dict] = None, opik_metadata: Optional[Dict] = None, schema_name: str = "response", temperature: float = 0.2, *, enable_fallback: bool = True, ) -> Any: """멀티턴 대화 호출 — entity_extractor Turn 0→1 등에 사용. messages: [{"role": "system", "content": ...}, {"role": "user", "content": ...}, ...] response_schema: 있으면 structured, 없으면 free text. 3-tier fallback (call_structured와 동일 정책)을 자동 적용한다. Tier 2/3에서 system 메시지에 SAFETY_SYSTEM_SUFFIX append + user/assistant content 일괄 sanitize. Tier 1 예외 분류 (`is_safety_related_error`): - 콘텐츠 안전 신호: Tier 2/3 진행 - 명시적 transient (timeout/rate-limit 등): 즉시 raise (비용 보호) """ from app.modules.llm.safety import ( SAFETY_SYSTEM_SUFFIX, is_safety_related_error, sanitize_messages, ) def _do_call(msgs: List[Dict[str, str]], cfg: Optional[Dict], suffix_tag: str = "") -> Any: binding = _get_router_binding() model = _resolve_model(step, cfg) base_metadata = _build_opik_metadata(step, opik_metadata) metadata = _add_fallback_tag(base_metadata, suffix_tag) if suffix_tag else base_metadata kwargs: Dict[str, Any] = { "model": model, "messages": msgs, "temperature": temperature, "metadata": metadata, } _sanitize_kwargs_for_model(model, kwargs) _apply_gemini_safety(model, kwargs) if response_schema: kwargs["response_format"] = { "type": "json_schema", "json_schema": { "name": schema_name, "schema": response_schema, "strict": True, }, } _sanitize_response_format_for_model(model, kwargs) response = _completion(binding, model, kwargs) content = response.choices[0].message.content or "" if response_schema: if not content: raise RuntimeError( f"LLM returned empty multiturn response for step={step}, model={model}" ) payload = _unwrap_tool_envelope( json.loads(content), response_schema) _validate_local_schema( payload, response_schema, step=step, schema_name=schema_name, ) return payload if not content: raise RuntimeError( f"LLM returned empty multiturn text for step={step}, model={model}" ) return content try: return _do_call(messages, project_config) except Exception as exc_t1: if not enable_fallback or not is_safety_related_error(exc_t1): raise logger.warning( "call_multiturn[%s] Tier 1 failed (%s), trying sanitized input", step, exc_t1, ) # Tier 2: system 메시지 첫 항목에 SAFETY_SYSTEM_SUFFIX append + content 전체 sanitize. safe_messages = sanitize_messages(messages) if safe_messages and isinstance(safe_messages[0], dict) and safe_messages[0].get("role") == "system": first = dict(safe_messages[0]) sys_content = first.get("content", "") if isinstance(sys_content, str): first["content"] = sys_content + SAFETY_SYSTEM_SUFFIX safe_messages = [first] + safe_messages[1:] try: return _do_call(safe_messages, project_config, suffix_tag="sanitized") except Exception as exc_t2: logger.warning( "call_multiturn[%s] Tier 2 sanitized failed (%s), trying GPT fallback", step, exc_t2, ) gpt_config = dict(project_config) if project_config else {} gpt_config[step] = {"model": "gpt"} return _do_call(safe_messages, gpt_config, suffix_tag="gpt_fallback")