"""Grok Imagine (xAI, OpenRouter 경유) 이미지 생성 클라이언트 (2026-08-13). 최종 스틸 생성 엔진 교체(#103, 사용자 확정 "grok 2 로 가자") 전용 — GeminiImageClient 를 상속해 컨텍스트·로깅·캡처 기계를 그대로 쓰고 운반층만 OpenRouter chat/completions(modalities=["image"]) 로 바꾼다. 운영 실측 계약 (artifact/20260813_MAI이미지25_실측/calls.json): - modalities 는 ["image"] 단독 — ["image","text"] 는 404. - 참조는 콘텐츠 parts 로 라벨 텍스트+이미지 쌍 다중 첨부 (5장 실측 통과). - 텍스트 총량(프롬프트+라벨) 8,000바이트 상한 — 초과는 xAI 400. 잘라 보내는 일은 없다: 여기서는 사전 검증으로 크게 실패시키고, 조립층(still_recipe v17 컴팩트 지도 절)이 예산을 맞춘다. - 응답 이미지: choices[0].message.images[].image_url.url (data URL). """ import base64 import concurrent.futures as _futures import json import logging import time import urllib.error import urllib.request from typing import List, Optional, Tuple from app.core.config import settings as _settings from app.core.image_call_budget import ( ImageCallBudgetExceeded, reserve_current_call, ) from app.modules.llm.gemini_image_client import GeminiImageClient from app.modules.llm.image_format import ensure_png_bytes from app.modules.llm.image_moderation import is_moderation_text from app.modules.llm.llm_logger import log_llm_call from app.services.image_capture.sink import capture_generated_image logger = logging.getLogger(__name__) OPENROUTER_CHAT_URL = "https://openrouter.ai/api/v1/chat/completions" # xAI 이미지 계열 텍스트 상한 실측 400 문언: "maximum allowed length of # 8000". 라벨 텍스트도 합산된다. # # 2026-08-18: 목표와 거절 상한을 나눈다(Codex BLOCK 수용). 하나로 쓰면 # 여유분 100B 안에 든 롤(실측 7,938B 1건)이 **보내 보지도 않고** 죽는데, # 모델이 실제로 거절하는 값은 8,000B 이고 거절은 400 이라 돈이 안 든다. # 즉 여유분을 "덜어내기 목표"로만 쓰고, 더 못 덜어도 진짜 상한 안이면 # 보낸다 — 못 보내는 것보다 낫고 잃는 것은 없다. GROK_TEXT_BYTE_LIMIT = 7900 # 덜어내기 목표(여유분 포함) GROK_TEXT_BYTE_HARD_LIMIT = 8000 # 모델이 선언한 실제 상한 — 넘으면 400 # 다시 보내면 될 수도 있는 상태 코드. HTTP 상태로 올 때는 아래 # HTTPError 갈래가 이미 다시 보내는데, **HTTP 200 본문 안에 담겨 오면** # (OpenRouter 관례) 한 번에 포기했다 — 2026-08-19 15:43 S88sh2 가 그렇게 # 죽었고 그 한 장 때문에 걷기 전체가 두 시간 다시 돌았다. # 실측 본문: {"message": "Image generation failed. Please try again # later.", "code": 502} # 검열(400)은 여기 없다 — 다시 보내도 같은 자리에서 막히고 거부당해도 # 요금이 나간다. GROK_RETRYABLE_BODY_CODES = frozenset({429, 500, 502, 503, 504}) def _body_error_code(err: object) -> int | None: """본문 오류의 상태 코드 — 정수로 실려 올 때만 읽는다.""" if not isinstance(err, dict): return None code = err.get("code") if isinstance(code, bool) or not isinstance(code, int): return None return code class GrokPromptOverBudget(RuntimeError): """텍스트 총량이 xAI 상한을 넘는다 — 조립층에서 고칠 문제라 크게 실패.""" class GrokImageClient(GeminiImageClient): """OpenRouter 경유 xAI Grok Imagine 클라이언트 — nb2 gen_fn 슬롯 호환. 상속 재사용: set_context/_ctx/_trace_step/_log_ctx/_opik_meta. 오버라이드: 키 소스(OPENROUTER_API_KEY)와 generate_image 운반층. """ def __init__(self, api_key: str = None, model: str = None) -> None: super().__init__(api_key=api_key, model=model or _settings.grok_image_model) def _get_api_key(self) -> str: if self._fixed_api_key: return self._fixed_api_key key = (getattr(_settings, "openrouter_api_key", "") or "").strip() if not key: raise RuntimeError("OPENROUTER_API_KEY 가 설정에 없다") return key def generate_image( self, prompt: str, reference_images: Optional[List[bytes]] = None, aspect_ratio: str = "16:9", labeled_references: Optional[List[Tuple[str, bytes]]] = None, ) -> Tuple[bytes, int]: """OpenRouter chat/completions 이미지 생성 — (PNG bytes, ms) 반환. aspect_ratio 는 API 파라미터가 없어 전송하지 않는다 — 참조 첨부 시 참조 비율을 상속하는 것이 실측 동작이고, 스틸 경로는 항상 참조를 첨부한다. (무참조 t2i 는 모델 기본 비율.) """ content: list = [{"type": "text", "text": prompt}] ref_image_ids = [] text_bytes = len(prompt.encode("utf-8")) if labeled_references: for i, (label, img_bytes) in enumerate(labeled_references, 1): lbl = label or f"Reference image {i}:" content.append({"type": "text", "text": lbl}) content.append({ "type": "image_url", "image_url": { "url": "data:image/png;base64," + base64.b64encode(img_bytes).decode("ascii"), }, }) text_bytes += len(lbl.encode("utf-8")) if label: ref_image_ids.append(label) elif reference_images: for img_bytes in reference_images: content.append({ "type": "image_url", "image_url": { "url": "data:image/png;base64," + base64.b64encode(img_bytes).decode("ascii"), }, }) shed: List[str] = [] if text_bytes > GROK_TEXT_BYTE_LIMIT: # 컴팩트 스템을 다 적용해도 긴 롤은 상한을 넘는다(직전 판 실측 # 378롤 중 61롤). 그대로 두면 롤이 죽고 롤이 죽으면 샷이 죽으므로 # **일반 규칙 절만** 정해진 순서로 덜어 맞춘다. 샷 고유 재료는 # 목록에 없다. 조립 권위가 무엇을 덜지 정하므로 헬퍼는 # still_recipe 에 있고 여기서는 상한을 아는 자리에서 부른다. from app.modules.pipeline.still_recipe import fit_grok_prompt label_bytes = text_bytes - len(prompt.encode("utf-8")) fitted, shed = fit_grok_prompt( prompt, label_bytes, GROK_TEXT_BYTE_LIMIT) if shed: before_bytes = text_bytes prompt = fitted content[0] = {"type": "text", "text": prompt} text_bytes = len(prompt.encode("utf-8")) + label_bytes # 덜어낸 내역의 durable 기록은 이 줄이 아니라 **실제로 보낸 # 프롬프트**다 — 아래 log_llm_call·Opik·capture 가 재대입된 # prompt 를 그대로 남긴다(records.json 은 조립 명목판). # 이 줄은 운전 중 눈으로 보기 위한 것이다. logger.warning( "grok 상한 대응 — 일반 규칙 절 %d개 덜어냄 %s " "(%dB → %dB, 목표 %dB, 모델 상한 %dB)", len(shed), shed, before_bytes, text_bytes, GROK_TEXT_BYTE_LIMIT, GROK_TEXT_BYTE_HARD_LIMIT, ) if text_bytes > GROK_TEXT_BYTE_HARD_LIMIT: raise GrokPromptOverBudget( f"grok 텍스트 총량 {text_bytes}B > " f"{GROK_TEXT_BYTE_HARD_LIMIT}B(모델 상한) — 컴팩트 조립" "(guidance v17)·일반 규칙 절 덜어내기로도 안 맞는다" ) body = { "model": self._model, "messages": [{"role": "user", "content": content}], "modalities": ["image"], } from app.modules.llm.image_tracer import get_image_tracer _tracer = get_image_tracer() req_bytes = json.dumps(body).encode("utf-8") headers = { "Authorization": f"Bearer {self._get_api_key()}", "Content-Type": "application/json", } start_time = time.monotonic() last_error: Exception | None = None timeout = _settings.llm_timeout_image_gen max_retries = _settings.llm_max_retries for attempt in range(1, max_retries + 2): try: reserve_current_call(source="grok_image_client.urlopen") req = urllib.request.Request( OPENROUTER_CHAT_URL, data=req_bytes, headers=headers, method="POST", ) # ★총 마감시간 (2026-08-14 카나리아 실측): urllib 의 # timeout 은 **개별 소켓 연산** 한도라, 서버가 응답을 # 찔끔찔끔 흘리며 연결을 붙잡으면 read 가 영원히 안 # 끝난다 — S3sh4 cine 호출이 ESTABLISHED 소켓 하나로 # 54분+ 매달려 llm_call_log 에 아무 기록도 없이 걷기 # 전체를 세웠다(lsof 로 확정). 왕복 전체를 timeout+60s # 벽 안에 가둔다 — 초과는 여느 네트워크 오류처럼 # 재시도 사다리를 탄다. def _roundtrip(): with urllib.request.urlopen( req, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) # ★shutdown(wait=False): 벽을 넘긴 좀비 스레드를 기다리면 # 행이 그대로 되돌아온다 — 스레드는 버리고 호출자만 푼다 # (소켓은 프로세스 수명/서버 종단에서 정리된다). _ex = _futures.ThreadPoolExecutor(max_workers=1) _fut = _ex.submit(_roundtrip) try: payload = _fut.result(timeout=timeout + 60) finally: _ex.shutdown(wait=False, cancel_futures=True) # ★HTTP 200 인데 본문에 오류가 담겨 오는 갈래를 여기서 # 가른다(2026-08-19). 루프 밖에서 다루면 이미 break 한 # 뒤라 다시 보낼 방법이 없다 — 그것이 S88sh2 를 죽인 # 모양이다. 검열은 다시 보내지 않는다(요금만 나간다). _body_err = ( payload.get("error") if isinstance(payload, dict) else None) if _body_err is not None: _text = json.dumps(_body_err, ensure_ascii=False) _code = _body_error_code(_body_err) if (_code in GROK_RETRYABLE_BODY_CODES and not is_moderation_text(_text) and attempt <= max_retries): last_error = RuntimeError( f"Grok image API error: {_text}") logger.warning( "grok 본문 오류 %s — 다시 보낸다 (%d/%d)", _code, attempt, max_retries) time.sleep(2 * attempt) continue break except _futures.TimeoutError as exc: last_error = TimeoutError( f"grok 왕복 총 마감시간 초과({timeout + 60}s) — " "서버가 연결을 붙잡은 채 응답을 안 끝냄") if attempt <= max_retries: time.sleep(2 * attempt) continue elapsed_ms = int((time.monotonic() - start_time) * 1000) log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=str(last_error), duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=str(last_error), duration_ms=elapsed_ms, extra_metadata=self._opik_meta, ) raise last_error from exc except ImageCallBudgetExceeded: raise except urllib.error.HTTPError as exc: error_text = exc.read().decode("utf-8", errors="replace") last_error = RuntimeError( f"Grok image API error {exc.code}: {error_text}") if exc.code in {429, 500, 502, 503, 504} and attempt <= max_retries: time.sleep(2 * attempt) continue elapsed_ms = int((time.monotonic() - start_time) * 1000) log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=str(last_error), duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=str(last_error), duration_ms=elapsed_ms, extra_metadata=self._opik_meta, ) raise last_error except Exception as exc: # noqa: BLE001 — 네트워크·타임아웃류 last_error = exc if attempt <= max_retries: time.sleep(2 * attempt) continue elapsed_ms = int((time.monotonic() - start_time) * 1000) log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=str(exc), duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=str(exc), duration_ms=elapsed_ms, extra_metadata=self._opik_meta, ) raise else: # pragma: no cover — break 없이 끝나는 일 없음 raise last_error or RuntimeError("unreachable") elapsed_ms = int((time.monotonic() - start_time) * 1000) if isinstance(payload, dict) and payload.get("error"): err = json.dumps(payload["error"], ensure_ascii=False) log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=err, duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=err, duration_ms=elapsed_ms, extra_metadata=self._opik_meta, ) if is_moderation_text(err): raise RuntimeError(f"moderation blocked: {err}") raise RuntimeError(f"Grok image API error: {err}") choices = (payload or {}).get("choices") or [] msg = (choices[0].get("message") or {}) if choices else {} for im in (msg.get("images") or []): url = ((im.get("image_url") or {}).get("url") if isinstance(im, dict) else "") or "" if not url.startswith("data:"): continue b64 = url.split(",", 1)[1] call_id = log_llm_call( model_name=self._model, user_prompt=prompt, output_text="[image generated]", status="success", duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], output_image_id=self._ctx.get("image_id"), duration_ms=elapsed_ms, params={"backend": "grok2"}, extra_metadata=self._opik_meta, ) # grok 은 jpg 로 준다 — 저장 규약(.png)과 lineage 정합을 위해 # nb2 경로와 같은 정규화를 거친다 (2026-08-06 교훈 동형). img_bytes = ensure_png_bytes( base64.b64decode(b64), context=f"{self._model}/" f"{self._ctx.get('operation_type') or '?'}", ) _capture_meta = { k: self._ctx.get(k) for k in ( "project_id", "episode_id", "operation_type", "step", "scene_index", "shot_index", "still_id", "entity_id", "image_id", ) if self._ctx.get(k) is not None } if ref_image_ids: _capture_meta["reference_labels"] = ref_image_ids capture_generated_image( img_bytes, role=self._ctx.get("operation_type") or "grok_image", prompt=prompt, generation_call_id=call_id, pipeline_metadata=_capture_meta, ) return img_bytes, elapsed_ms # 이미지 없는 200 — 텍스트/빈 응답은 실패로 기록하고 전파 err = f"no image in response (finish={choices[0].get('finish_reason') if choices else None})" log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=err, duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=err, duration_ms=elapsed_ms, extra_metadata=self._opik_meta, ) raise RuntimeError(f"Grok image API: {err}")