"""이미지 생성 Opik 트레이서 — 메타데이터만 기록 (이미지 바이트 절대 첨부 안 함). 모든 이미지 생성 호출(Gemini T2I/I2I, fal.ai)을 감싸서 프롬프트, 참조 이미지 UUID, 생성 이미지 UUID, 파라미터, 에러를 Opik에 기록. thread-local context(StepRunner)가 있으면 자동으로 session 그룹핑. """ import logging import threading import time from contextlib import contextmanager from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) _opik_available = False try: import opik _opik_available = True except ImportError: # 의도적: opik은 optional 의존성. 미설치 환경에서도 이미지 파이프라인은 정상 동작. pass def _provider_of(model: str) -> str: """모델 이름 → provider. 모르는 이름을 fal.ai 로 몰지 않는다. ★전에는 gemini/imagen 이 아니면 **전부 fal.ai** 였다. gpt-image 계열이 Opik 에서 fal.ai 로 잘못 적혔다 (Codex 리뷰 지적, 2026-08-07). """ m = (model or "").lower() if "gemini" in m or "imagen" in m: return "google_ai" if m.startswith("gpt-image") or "gpt-image" in m or m.startswith("dall-e"): return "openai" if "qwen" in m or "flux" in m or "fal" in m: return "fal.ai" return "unknown" class ImageTracer: """이미지 생성 호출 래퍼 — Opik span 자동 기록. Usage: tracer = ImageTracer() with tracer.span("ref_image_gen", model="imagen-3.0-generate-002", prompt=prompt, ref_image_ids=["abc123"]) as s: img_bytes, elapsed = gemini_client.generate_image(...) s.set_output(image_id="def456", duration_ms=elapsed) # 또는 단순 호출 tracer.log("fal_angle", model="qwen-image-edit-2511", prompt="H=15 V=-10 Z=1.0", params={"horizontal": 15, "vertical": -10, "zoom": 1.0}, ref_image_ids=["abc123"], output_image_id="def456", duration_ms=320) """ def __init__(self): self._client = None self._explicit_context: Optional[Dict] = None if _opik_available: try: # ★project_name 을 명시한다 (2026-08-07 실측). 인자 없이 부르면 # 기본 프로젝트로 쌓여 **텍스트 호출과 다른 곳에 갈린다.** # 실제로 이미지 trace 는 기본 프로젝트에, 텍스트는 # theroad-scene-lab 에 있었고, 후자만 보고 "이미지 기록이 # 통째로 없다"고 잘못 판단했다. project = None try: import os as _os from app.core.config import settings project = getattr(settings, "opik_project_name", None) # 셀프 호스팅(2026-08-14): 이 클라이언트가 llm_client 의 # _init_opik 보다 먼저 만들어질 수 있다 — SDK 는 환경 # 변수만 읽으므로 주소·작업 공간을 생성 전에 올린다. if getattr(settings, "opik_url_override", ""): _os.environ.setdefault( "OPIK_URL_OVERRIDE", settings.opik_url_override) _os.environ.setdefault( "OPIK_WORKSPACE", settings.opik_workspace) except Exception: # noqa: BLE001 — 설정을 못 읽어도 추적은 산다 project = None self._client = (opik.Opik(project_name=project) if project else opik.Opik()) except Exception as exc: logger.warning("Opik client init failed — image tracing disabled: %s", exc) def set_context(self, ctx: Optional[Dict]) -> None: """명시적 context 설정 — thread pool worker에서 사용. StepRunner 메인 스레드에서 context를 캡처한 후 worker thread에 전달할 때 사용. """ self._explicit_context = ctx def _get_context(self) -> Dict: """context 조회 — **thread-local 우선**, 없으면 명시적, 없으면 빈 dict. ★순서를 뒤집었다 (Codex 리뷰 2026-08-07). `_explicit_context` 는 process 하나에 공유되고 **해제되지 않는다**(`image_steps` 가 설정만 한다). 그것을 먼저 읽으면, 스텝 이름은 맞게 나와도 Opik 의 thread_id·tags·프로젝트 묶음이 **직전 스텝 것으로** 붙는다. thread-local 이 있으면 그것이 지금 이 실행의 진실이고, worker thread 처럼 thread-local 이 빈 자리에서만 명시적 값이 쓰인다. """ try: from app.modules.llm.llm_client import _get_thread_opik_meta thread_meta = _get_thread_opik_meta() if thread_meta: return thread_meta except Exception: # noqa: BLE001 pass return self._explicit_context or {} def get_context(self) -> Dict: """현재 context 를 밖에서 읽는다 (gpt-image primitive 가 step 이름을 끌어올 때 사용). Opik client 가 없어도 context 는 있을 수 있으므로 `_client` 유무와 무관하게 돌려준다.""" return self._get_context() def log( self, step: str, model: str, prompt: str, ref_image_ids: Optional[List[str]] = None, output_image_id: Optional[str] = None, status: str = "success", error: Optional[str] = None, duration_ms: int = 0, params: Optional[Dict[str, Any]] = None, extra_metadata: Optional[Dict[str, Any]] = None, provider: Optional[str] = None, usage: Optional[Dict[str, Any]] = None, total_cost: Optional[float] = None, ) -> None: """이미지 생성 결과를 Opik trace로 기록. Args: extra_metadata (Phase 4 iter 7 review I1): trace_meta (project_id / episode_id / scene_index / shot_index / still_id / entity_id 등) 를 trace + span metadata 에 병합. DB log 와 별도로 Opik trace 에서도 fan-out 단위 추적이 가능하도록. usage / total_cost (2026-08-27, 감사 2-B·0-D): ★**litellm 을 우회하는 호출은 여기서 안 넘기면 토큰·비용이 통째로 사라진다.** 실측: `still_recipe_fix_rejudge_ openrouter:xai/grok4.6` 등 137 span 이 `usage={}` · `total_estimated_cost=None` 이었고, 그 때문에 ① 비용 집계에서 빠지고(0-D 잔여) ② 캐시 적중률을 못 재서 감사표가 「캐시 0%」로 읽었다. 「0」이 「없다」가 아니라 **재는 칸이 비어 있던 것**이다. """ if not self._client: return try: ctx = self._get_context() # v2 는 `session_id` 를 지우고 `thread_id` 를 넣는다(litellm 이 # 읽는 이름). 여기서 `session_id` 만 보면 부모 없이 홀로 서는 # trace 가 에피소드 묶음 밖으로 떨어진다 — 2026-08-24 Codex # BLOCK 3. OFF 에서는 `thread_id` 가 없어 지금과 동일하다. session_id = ctx.get("thread_id") or ctx.get("session_id") tags = [step, model] if ctx.get("tags"): tags.extend(t for t in ctx["tags"] if t not in tags) trace_input = {"prompt": prompt} if ref_image_ids: trace_input["ref_image_ids"] = ref_image_ids if params: trace_input["params"] = params trace_output: Dict[str, Any] = {"status": status} if output_image_id: trace_output["image_id"] = output_image_id error_info = None if error: trace_output["error"] = error[:500] error_info = {"exception_type": "ImageGenError", "message": error[:500], "traceback": ""} trace_metadata = { "model": model, "duration_ms": duration_ms, "step": step, } if extra_metadata: trace_metadata.update(extra_metadata) # ★span_metadata 를 trace 생성보다 앞으로 옮겼다 — v2 분기가 # 그것을 쓴다. 내용은 그대로다. span_metadata = {"duration_ms": duration_ms, **(params or {})} if extra_metadata: span_metadata.update(extra_metadata) # v1(OFF) 은 지금 그대로 둔다 — 「되돌리기는 한 줄」 안전판이라 # 설정을 끄면 이전 주행과 같은 payload 가 나가야 한다. span_tags: Optional[List[str]] = None # ── v2 (2026-08-23): 부모가 있으면 span 만 붙인다 ────────── 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) op = (params or {}).get("role") or step axis = build_axis_tags( step=step, op=op, model=model, provider=provider or _provider_of(model), status=None if status == "success" else status, ) parent = current_trace() if parent is not None: s = self._client.span( trace_id=parent.uid, name=f"{op} · {model}", type="llm", model=model, provider=provider or _provider_of(model), input=trace_input, output=trace_output, metadata=span_metadata, tags=axis, **_usage_kwargs(usage, total_cost), ) s.end() return # 부모가 없으면 이름 있는 trace 를 홀로 세운다 — 기록이 # 사라지는 것보다 낫다. tags = axis # ★span 에도 같은 축을 실는다. 감사 도구는 **span 만 센다** # (`fetch_spans`: trace 는 묶음·이름을 얻는 데만 쓴다). # trace 에만 달고 span 을 맨몸으로 두면 그 호출은 「어느 # 스텝인지 모름」으로 남는다 — 2026-08-25 실측에서 span # 26/252 가 스텝 축을 잃었고 그중 11 건이 이 자리였다. span_tags = axis trace = self._client.trace( name=f"{step}/{model}", input=trace_input, output=trace_output, metadata=trace_metadata, tags=tags, thread_id=session_id, error_info=error_info, ) span_kwargs: Dict[str, Any] = { "name": step, "type": "llm", "model": model, "provider": provider or _provider_of(model), "input": trace_input, "output": trace_output, "metadata": span_metadata, } if span_tags is not None: span_kwargs["tags"] = span_tags span_kwargs.update(_usage_kwargs(usage, total_cost)) s = trace.span(**span_kwargs) s.end() trace.end() except Exception as exc: logger.debug("ImageTracer.log failed (non-fatal): %s", exc) @contextmanager def span( self, step: str, model: str, prompt: str, ref_image_ids: Optional[List[str]] = None, params: Optional[Dict[str, Any]] = None, ): """Context manager — 생성 후 결과를 set_output으로 기록. with tracer.span("scene_image_gen", ...) as s: img, elapsed = generate(...) s.set_output(image_id="xxx", duration_ms=elapsed) """ ctx = _SpanContext() t0 = time.monotonic() try: yield ctx except Exception as exc: ctx._error = str(exc) ctx._status = "error" raise finally: if not ctx._duration_ms: ctx._duration_ms = int((time.monotonic() - t0) * 1000) self.log( step=step, model=model, prompt=prompt, ref_image_ids=ref_image_ids, output_image_id=ctx._image_id, status=ctx._status, error=ctx._error, duration_ms=ctx._duration_ms, params=params, ) class _SpanContext: """span() context manager 내부에서 결과 설정용.""" __slots__ = ("_image_id", "_status", "_error", "_duration_ms") def __init__(self): self._image_id: Optional[str] = None self._status: str = "success" self._error: Optional[str] = None self._duration_ms: int = 0 def set_output(self, image_id: Optional[str] = None, duration_ms: int = 0, status: str = "success", error: Optional[str] = None): self._image_id = image_id self._duration_ms = duration_ms self._status = status self._error = error #: 스텝 컨텍스트에서 끌어올 신원 칸 — **화이트리스트**다. thread-local 에는 #: Opik 묶음용 `tags`/`session_id`/`trace_name` 도 같이 들어 있어서, 통째로 #: 병합하면 그것이 DB 호출 기록의 metadata 로 새 나간다. _IDENTITY_KEYS = ("project_id", "episode_id") def ambient_call_meta() -> Dict[str, Any]: """기록용 신원 메타를 **주변 scope 에서 직접 읽는다**. 근거 두 겹을 이 순서로 본다: 1. **이미지 포착 scope** — `still_id`·`shot_index` 까지 아는 가장 정확한 근거. 이미지를 만드는 자리에만 열려 있다. 2. **스텝 컨텍스트**(StepRunner 의 thread-local) — 신원 두 칸만 메운다. 검색처럼 이미지를 포착하지 않는 호출은 1번 scope 를 열 이유가 없어, 그것만 보면 `project_id`/`episode_id` 가 빈 채로 남는다. 실제로 검색 호출 기록이 그래서 비었고, **프로젝트 단위로 "이 검색이 살아 있나"를 물을 수 없었다.** ★인자로 받지 않는 이유 (2026-08-07): 이미지 저수준 함수는 호출 사이트가 여럿이라, 인자로 만들면 배선을 빠뜨린 사이트가 **조용히 기록 없이** 돈다. 실제로 gpt-image 경로 전체가 Opik·llm_call_log 양쪽에 한 줄도 안 남아 "작업이 살아 있나"를 볼 근거가 없었다. StepRunner 는 **모든 스텝**에서 컨텍스트를 설정하므로, 호출 사이트를 하나씩 배선하는 것과 달리 2번은 빠뜨릴 자리가 없다. ★2번의 한계: thread-local 이라 **worker thread 에는 자동 전파되지 않는다.** 지금 검색 호출자는 한 곳이고 병렬이 아니라 닿지만, 나중에 ThreadPool 로 나누면 신원이 다시 빈다. 그때는 1번처럼 명시 전파가 필요하다 (`bind_current_generation_context` 가 그 자리의 본보기다). """ meta: Dict[str, Any] = {} try: from app.services.image_capture.context import current_context gctx = current_context() if gctx is not None: for key in ("project_id", "episode_id", "still_id", "scene_index", "shot_index", "entity_id", "stage"): value = getattr(gctx, key, None) if value is not None: meta[key] = value except Exception as exc: # noqa: BLE001 — 기록은 메인 흐름을 막지 않는다 logger.debug("ambient_call_meta 조회 실패: %s", exc) if not all(meta.get(key) for key in _IDENTITY_KEYS): try: from app.modules.llm.llm_client import _get_thread_opik_meta step_meta = _get_thread_opik_meta() or {} for key in _IDENTITY_KEYS: if not meta.get(key) and step_meta.get(key): meta[key] = step_meta[key] except Exception as exc: # noqa: BLE001 logger.debug("ambient_call_meta 스텝 컨텍스트 조회 실패: %s", exc) return meta def resolve_step_name(fallback: Optional[str], ambient: Dict[str, Any]) -> str: """스텝 이름 — capture scope 의 stage > thread-local Opik 태그 > fallback. ★ImageTracer 의 explicit context 는 **쓰지 않는다**. 그것은 process 하나에 공유되고 지워지지 않아서, 그 mixin 을 쓰지 않는 스텝의 호출이 **직전 스텝 이름으로** 기록된다 (Codex 리뷰 2026-08-07). ★2026-08-28 실측: thread 태그를 `tags[0]` 로 **그대로** 집어서 두 군데가 깨졌다. 그 태그는 이미 축 접두사가 붙은 `step:shot_conti_light` 라서 - DB `llm_call_log.step_name` 이 `step:shot_conti_light` 가 되고 (다른 행은 `scene_image_pipeline` 처럼 맨 이름이라 **같은 스텝이 두 이름으로 쪼개진다**) - `build_axis_tags(step=...)` 가 접두사를 한 번 더 붙여 **`step:step:shot_conti_light`** 가 된다 이번 완주 주행 Opik 실측 = 이중 접두사 span **6건** / 세 자리 (`shot_conti_light` 3 · `background_render` 2 · `floor_plan_render` 1). 그래서 **`step:` 축 태그를 골라 접두사를 벗겨서** 돌려준다. 순서에 기대지 않는다 — 종전에는 `_AXIS_ORDER` 가 step 을 맨 앞에 두는 덕에 우연히 맞고 있었고, 축 순서가 바뀌면 모델명·제공자명이 스텝 이름이 됐다. """ stage = ambient.get("stage") if stage: return str(stage) try: from app.modules.llm.llm_client import _get_thread_opik_meta from app.modules.llm.opik_trace import STEP_TAG_PREFIX tags = (_get_thread_opik_meta() or {}).get("tags") or [] for tag in tags: text = str(tag) if text.startswith(STEP_TAG_PREFIX): bare = text[len(STEP_TAG_PREFIX):].strip() if bare: return bare except Exception as exc: # noqa: BLE001 logger.debug("resolve_step_name 조회 실패: %s", exc) return fallback or "image_gen" def _usage_kwargs( usage: Optional[Dict[str, Any]], total_cost: Optional[float], ) -> Dict[str, Any]: """Opik span 이 받는 `usage` / `total_cost` 칸 — **있는 것만 넣는다.** ★0 을 채우지 않는다. provider 가 안 알려 준 값을 0 으로 적으면 「토큰을 안 썼다」로 읽히고, 그 0 이 합산에 섞이면 비용이 실제보다 작게 나온다. 없으면 **칸 자체를 안 만든다** — 그래야 나중에 「못 잰 것」과 「0 이었던 것」을 가를 수 있다. ★키 이름은 litellm 이 쓰는 것과 맞춘다(`prompt_tokens` / `completion_tokens` / `total_tokens`). 이름이 다르면 감사 도구가 같은 칸으로 못 읽고, 지금 openrouter 137 span 이 겪는 것과 같은 「기록은 있는데 안 세어지는」 자리가 또 생긴다. """ out: Dict[str, Any] = {} if isinstance(usage, dict): keep = {k: v for k, v in usage.items() if isinstance(v, (int, float)) and not isinstance(v, bool)} if keep: out["usage"] = keep if isinstance(total_cost, (int, float)) and not isinstance( total_cost, bool): out["total_cost"] = float(total_cost) return out def usage_of_openai_response(resp: Any) -> Dict[str, Any]: """OpenAI 계열 SDK 응답에서 토큰 칸을 꺼낸다 — **캐시 칸까지**. ★캐시 적중이 여기 들어 있는데 지금까지 아무도 안 꺼냈다. 그래서 감사표가 판정 루프 캐시를 「0%」로 적었고, 나는 그것을 「캐시가 안 붙는다」로 읽을 뻔했다. 실측하면 grok 은 2,304/2,326 = 99% 적중이다. 없는 칸은 **안 만든다**(`_usage_kwargs` 와 같은 규칙). """ u = getattr(resp, "usage", None) if u is None: return {} out: Dict[str, Any] = {} for name in ("prompt_tokens", "completion_tokens", "total_tokens"): v = getattr(u, name, None) if isinstance(v, (int, float)) and not isinstance(v, bool): out[name] = v # 캐시 칸은 provider 마다 이름이 다르고 중첩돼 있다. det = getattr(u, "prompt_tokens_details", None) cached = getattr(det, "cached_tokens", None) if det is not None else None if cached is None: cached = getattr(u, "cached_tokens", None) if isinstance(cached, (int, float)) and not isinstance(cached, bool): out["cached_tokens"] = cached return out def record_provider_call( *, step: str, model: str, prompt: str, status: str, duration_ms: int, meta: Dict[str, Any], ref_count: int = 0, operation: Optional[str] = None, ref_image_ids: Optional[List[str]] = None, error: Optional[str] = None, to_db: bool = True, output_text: Optional[str] = None, provider: Optional[str] = None, usage: Optional[Dict[str, Any]] = None, total_cost: Optional[float] = None, ) -> Optional[str]: """provider 호출 하나를 DB(llm_call_log)+Opik 양쪽에 남기고 **호출 ID 를 돌려준다**. 이미지·검색·텍스트 어느 쪽이든 **litellm 을 우회하는 직접 호출**은 여기를 거쳐야 Opik 에 남는다(litellm 경로는 콜백으로 자동이다). 성공도 실패도 남는다. 어느 쪽 기록이 터져도 본 작업은 계속되어야 하므로 전부 삼킨다. 돌려준 ID 는 capture 의 ``generation_call_id`` 로 넘겨 **DB 호출 기록과 포착 자산을 묶는다**. ``to_db=False`` — 이미 자기 자리에서 ``log_llm_call`` 을 부르는 호출자용 (Opik 만 보탠다). 두 번 세지 않기 위한 것이다. ``output_text`` — 성공 기록의 산출 표시. ★기본값을 이미지로 두지 않는다 (Codex 리뷰 2026-08-07): 전에는 모든 성공에 ``"[image generated]"`` 를 박아서 **검색 호출까지 DB 에서 이미지 생성으로 읽혔다.** 이미지 호출자가 자기 표식을 넘긴다. ``provider`` — Opik span 의 provider. 모델 이름으로 못 가르는 경우 (검색 orchestrator 는 ``gpt-5.6`` 이라 이미지 규칙에 안 걸린다) 호출자가 명시한다. """ call_id: Optional[str] = None if to_db: try: from app.modules.llm.llm_logger import log_llm_call call_id = log_llm_call( model_name=model, user_prompt=prompt, output_text=output_text if status == "success" else None, status=status, duration_ms=duration_ms, project_id=meta.get("project_id"), episode_id=meta.get("episode_id"), operation_type=operation, step_name=step, reference_image_ids=list(ref_image_ids) if ref_image_ids else None, # ★토큰 칸을 잇는다 (2026-08-29, Codex 가 원인을 짚었다). # # `usage_of_openai_response` 가 만든 값을 **Opik 에는 # 넘기면서 DB 에는 안 넘겼다.** `log_llm_call` 은 두 칸을 # 이미 받는데 호출부가 안 준 것이다 — 그래서 DB 에서 # `input_tokens`/`output_tokens` 가 전부 0 이었고, # 「판정 시간이 출력 토큰에 비례한다」를 재려고 Opik 을 # 따로 봐야 했다. # # ★없는 칸은 **안 만든다** — provider 가 usage 를 안 주면 # `usage` 가 빈 dict 이고 그때는 None 이 간다. 0 을 적으면 # 「토큰을 안 썼다」와 「모른다」가 구분되지 않는다. input_tokens=(usage or {}).get("prompt_tokens"), output_tokens=(usage or {}).get("completion_tokens"), error_message=error, metadata={k: v for k, v in meta.items() if k not in ("project_id", "episode_id")} or None, ) except Exception as exc: # noqa: BLE001 # ★debug 가 아니라 warning 이다 (2026-08-24). DB 기록이 조용히 # 사라지면 「이 호출이 무엇을 만들었나」를 DB 로 못 묻게 되는데, # 그 사실 자체가 안 보인다. 실제로 그런 결함을 찾겠다고 넉 달 # 뒤에 조사를 벌였다(그때 본 것은 시험 기록이었다 — 아래 참조). # Opik 쪽(:447)은 debug 로 둔다 — 자체 호스팅이 잠깐 죽으면 # 호출마다 warning 이 쏟아져 진짜 신호를 덮는다. logger.warning( "record_provider_call: llm_call_log 기록 실패 (non-fatal): %s", exc) try: get_image_tracer().log( step=step, model=model, prompt=prompt, ref_image_ids=list(ref_image_ids) if ref_image_ids else None, status=status, error=error, duration_ms=duration_ms, params={"ref_count": ref_count, "role": operation}, extra_metadata=meta, provider=provider, usage=usage, total_cost=total_cost, ) except Exception as exc: # noqa: BLE001 logger.debug("record_provider_call: Opik 기록 실패: %s", exc) return call_id def traced_call( *, operation: str, provider: Optional[str] = None, model_of=None, prompt_of=None, output_text: Optional[str] = None, ): """litellm 을 우회하는 provider 호출 함수를 감싸 **성공·실패를 다** 남긴다. urllib 로 provider 를 직접 치는 자리(vision 검증·변형 추천·PDF 검증· legacy 구조화 출력)는 호출 흐름이 저마다 달라 본문 안에 기록을 심으면 빠뜨리기 쉽다. 감싸면 return 경로가 몇 개든 한 번만 남는다. ``model_of`` / ``prompt_of`` 는 감싼 함수의 인자에서 모델명·프롬프트를 꺼내는 함수다(사이트마다 인자 모양이 달라 호출자가 준다). """ import functools def deco(fn): @functools.wraps(fn) def wrapper(*args, **kwargs): meta = ambient_call_meta() step = resolve_step_name(operation, meta) try: model = str(model_of(*args, **kwargs)) if model_of else "unknown" except Exception: # noqa: BLE001 model = "unknown" try: prompt = str(prompt_of(*args, **kwargs)) if prompt_of else "" except Exception: # noqa: BLE001 prompt = "" started = time.monotonic() try: result = fn(*args, **kwargs) except Exception as exc: record_provider_call( step=step, model=model, prompt=prompt, status="error", duration_ms=int((time.monotonic() - started) * 1000), meta=meta, operation=operation, provider=provider, error=str(exc)[:500]) raise record_provider_call( step=step, model=model, prompt=prompt, status="success", duration_ms=int((time.monotonic() - started) * 1000), meta=meta, operation=operation, provider=provider, output_text=output_text) return result return wrapper return deco # 모듈 레벨 싱글턴 (thread-safe) _tracer: Optional[ImageTracer] = None _tracer_lock = threading.Lock() def get_image_tracer() -> ImageTracer: """싱글턴 ImageTracer 반환 (thread-safe).""" global _tracer if _tracer is None: with _tracer_lock: if _tracer is None: _tracer = ImageTracer() return _tracer