"""요소 추출 v3 — LiteLLM 기반 독립 4단계. 모든 LLM 호출이 독립적 (멀티턴 세션 없음). 이전 단계 결과는 프롬프트에 인라인으로 포함. Step 1: 스타일 + 요소 이름 (gemini-pro) Step 2a: GPT 리뷰/필터링 (gpt) Step 2b: GPT 상세 추출 (gpt) Step 3: T2I 프롬프트 생성 (gemini-flash, 병렬) """ import hashlib import json import logging import os import time from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Dict, List, Optional, Tuple from app.modules.llm.llm_client import call_structured, PIPELINE_STEPS from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) _MODULE = "entity_extractor_v2" def _load_prompt(name: str, **kwargs) -> str: return load_prompt(_MODULE, name, **kwargs) def _load_schema(name: str) -> Dict[str, Any]: stem = name.replace(".json", "") if name.endswith(".json") else name return load_schema(_MODULE, stem) def _load_system() -> str: return _load_prompt("system") # ── 스키마 ── _ENTITY_BRIEF_ITEM = { "type": "object", "properties": { "name": {"type": "string"}, "appearances": {"type": "integer"}, }, "required": ["name", "appearances"], "additionalProperties": False, } STEP1_SCHEMA = { "type": "object", "properties": { # 스타일 필드 (turn0_style.md 기준) "episode_summary": {"type": "string", "description": "에피소드 상세 요약 (5문장 이상)"}, "visual_world_rules": { "type": "array", "items": {"type": "string"}, "description": "인물의 물리적 존재 판단과 씬 간 시각적 일관성에 필요한 핵심 규칙만 (2~5개)", }, "era": {"type": "string"}, "region": {"type": "string"}, "genre": {"type": "string"}, "building_style": {"type": "string"}, "clothing_style": {"type": "string"}, "vehicle_style": {"type": "string"}, "color_tone": {"type": "string"}, "must_avoid": {"type": "string"}, # 요소 필드 (turn1.md 기준) "characters": {"type": "array", "items": _ENTITY_BRIEF_ITEM}, "locations": {"type": "array", "items": _ENTITY_BRIEF_ITEM}, "props": {"type": "array", "items": _ENTITY_BRIEF_ITEM}, }, "required": [ "episode_summary", "visual_world_rules", "era", "region", "genre", "building_style", "clothing_style", "vehicle_style", "color_tone", "must_avoid", "characters", "locations", "props", ], "additionalProperties": False, } ENTITY_DETAIL_SCHEMA = { "type": "object", "properties": { "name": {"type": "string"}, "entity_type": {"type": "string"}, "description": {"type": "string"}, "visual_traits": { "type": "array", "items": {"type": "string"}, }, "t2i_prompt": {"type": "string"}, # D6 T2 + Area B (2026-05-13): metadata_json 의 closed shape. # entity_type 별 instance: # - character: {"location": null, "visual_identity": null} # - location single_space: {"location": {"space_profile": # {"kind": "single_space", "allowed_space_keys": ["main"], # "default_space_key": null}}, "visual_identity": null} # - location multi_space: {"location": {"space_profile": {...}}, # "visual_identity": null} # - prop: {"location": null, # "visual_identity": {"reference_required": true|false}} # object only — null is OOS for prop # OpenAI structured outputs strict mode — additionalProperties=False + # 모든 nested object key required 의무. "metadata_json": { "type": "object", "properties": { "location": { "anyOf": [ {"type": "null"}, { "type": "object", "properties": { "space_profile": { "type": "object", "properties": { "kind": { "type": "string", "enum": ["single_space", "multi_space"], }, "allowed_space_keys": { "type": "array", "items": {"type": "string"}, }, "default_space_key": { "type": ["string", "null"], }, }, "required": ["kind", "allowed_space_keys", "default_space_key"], "additionalProperties": False, }, }, "required": ["space_profile"], "additionalProperties": False, }, ], }, "visual_identity": { "anyOf": [ {"type": "null"}, { "type": "object", "properties": { "reference_required": {"type": "boolean"}, }, "required": ["reference_required"], "additionalProperties": False, }, ], }, }, "required": ["location", "visual_identity"], "additionalProperties": False, }, }, # short_description 은 entity_t2i 후속 fix (Codex review MIN 1) 로 schema 에서 제거. # entity_detail step (Step 14) 자체는 별도 detail_schema 사용 + short_description # 생성 X. entity_t2i 에서도 source detail forward 정책상 빈 문자열로 고정됐음 → # consumer 0 (production code 에 ImageAsset / SceneStill / UI / export 모두 미사용, # experiment scripts 는 description fallback 패턴) → schema field 정리. # # metadata_json: T2-fix (I3) 로 required 추가. character/prop 은 prompt 가 `{}` # 출력 명시. location 은 space_profile shape — entity_t2i._gen_t2i 에서 post- # validate (entity_type==location 분기). "required": ["name", "entity_type", "description", "visual_traits", "t2i_prompt", "metadata_json"], "additionalProperties": False, } # ── 체크포인트 ── class _Checkpoint: def __init__(self, checkpoint_dir: Optional[str], fulltext: str): self._path = None self._data = {} if checkpoint_dir: os.makedirs(checkpoint_dir, exist_ok=True) ep_hash = hashlib.md5(fulltext.encode("utf-8")).hexdigest()[:8] self._path = os.path.join(checkpoint_dir, f"entity_extraction_{ep_hash}.json") if os.path.exists(self._path): try: self._data = json.loads(open(self._path, encoding="utf-8").read()) logger.info("Entity checkpoint loaded: step=%s", self._data.get("completed_step")) except Exception: self._data = {} def get(self, key: str, default=None): return self._data.get(key, default) @property def completed_step(self) -> Optional[str]: return self._data.get("completed_step") def save(self, step: str, **kwargs): if not self._path: return self._data.update(kwargs) self._data["completed_step"] = step tmp = self._path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(self._data, f, ensure_ascii=False, indent=2) os.replace(tmp, self._path) def delete(self): if self._path and os.path.exists(self._path): os.unlink(self._path) logger.info("Entity checkpoint deleted") # ── 메인 함수 ── def extract_entities( fulltext: str, checkpoint_dir: Optional[str] = None, project_llm_config: Optional[Dict] = None, prior_entities: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, Any]: """독립 4단계 요소 추출. .. deprecated:: v4 파이프라인 (entity_steps.py / entity_extractor_v4) 도입 후 모든 production 호출자가 이전됨. 이 legacy public function 은 여전히 ENTITY_DETAIL_SCHEMA 의 description / visual_traits / t2i_prompt 를 LLM 출력 그대로 사용하므로 entity_t2i 의 unsourced trait 차단 정책 (entity_steps.py 의 source-detail forward) 가 적용 안 됨. 신규 호출자 는 v4 경로 사용. 본 function 은 unit test 외 호출 금지 — production re-introduction 시 동일 forward 정책 또는 fail-fast 를 먼저 도입할 것. Args: fulltext: 시나리오 전문 checkpoint_dir: 체크포인트 저장 경로 project_llm_config: 프로젝트별 LLM 모델 설정 prior_entities: 이전 에피소드 기존 요소 목록 Returns: { "characters": [{name, description, visual_traits, t2i_prompt}], "locations": [{name, description, visual_traits, t2i_prompt}], "props": [{name, description, visual_traits, t2i_prompt}], "style": {era, region, genre, visual_tone}, } """ cp = _Checkpoint(checkpoint_dir, fulltext) system = _load_system() prior_block = "" if prior_entities: prior_block = ( "\n\n[이전 에피소드에서 추출된 기존 요소]\n" + json.dumps(prior_entities, ensure_ascii=False, indent=1) + "\n기존 요소가 이번 에피소드에도 등장하면 이름을 동일하게 유지하세요.\n" ) # ── Step 1: 스타일 + 요소 이름 (1회 호출) ── if cp.completed_step and cp.completed_step >= "step1": step1_result = cp.get("step1_result") logger.info("Step 1: restored from checkpoint") else: logger.info("Step 1: style + entity names") step1_prompt = _load_prompt("turn0_style", prior_block=prior_block, fulltext=fulltext) step1_prompt += "\n\n" + _load_prompt("turn1", prior_block=prior_block, fulltext=fulltext) step1_result = call_structured( step="entity_style", system_prompt=system, user_prompt=step1_prompt, response_schema=STEP1_SCHEMA, project_config=project_llm_config, schema_name="entity_step1", ) cp.save("step1", step1_result=step1_result) style = { "episode_summary": step1_result.get("episode_summary", ""), "era": step1_result.get("era", ""), "region": step1_result.get("region", ""), "genre": step1_result.get("genre", ""), "building_style": step1_result.get("building_style", ""), "clothing_style": step1_result.get("clothing_style", ""), "vehicle_style": step1_result.get("vehicle_style", ""), "color_tone": step1_result.get("color_tone", ""), "must_avoid": step1_result.get("must_avoid", ""), } logger.info("Step 1: %d chars, %d locs, %d props, style=%s/%s/%s", len(step1_result.get("characters", [])), len(step1_result.get("locations", [])), len(step1_result.get("props", [])), style["era"], style["region"], style["genre"]) # ── Step 2a: GPT 리뷰/필터링 (1회 호출) ── if cp.completed_step and cp.completed_step >= "step2a": review_result = cp.get("review_result") logger.info("Step 2a: restored from checkpoint") else: logger.info("Step 2a: GPT review") from app.modules.pipeline.ref_image_pipeline import _load_lvm_prompt def _fmt(items): return ", ".join(f"{e['name']}({e['appearances']}회)" for e in items) review_schema = _load_schema("turn1_review_schema.json") review_prompt = _load_lvm_prompt( "entity_list_review", screenplay_summary=fulltext, characters=_fmt(step1_result.get("characters", [])), locations=_fmt(step1_result.get("locations", [])), props=_fmt(step1_result.get("props", [])), ) try: review_result = call_structured( step="entity_review", system_prompt="시나리오 분석 전문가. 추출된 요소 목록의 정확성을 평가한다.", user_prompt=review_prompt, response_schema=review_schema, project_config=project_llm_config, schema_name="entity_review", ) except Exception as exc: logger.warning("Step 2a GPT review failed, skipping: %s", exc) review_result = {"entities": []} cp.save("step2a", step1_result=step1_result, review_result=review_result) # 리뷰 결과 적용 — 필터링 (importance: 0-100 정수, appearances: 정수) entities_review = review_result.get("entities", []) low_importance = { e["name"] for e in entities_review if int(e.get("importance", 50)) < 10 and int(e.get("appearances", 0)) < 2 } if low_importance: logger.info("Step 2a: filtering %d low-importance: %s", len(low_importance), low_importance) entity_queue: List[Tuple[str, str]] = [] for e in step1_result.get("characters", []): if e["name"] not in low_importance: entity_queue.append((e["name"], "character")) for e in step1_result.get("locations", []): if e["name"] not in low_importance: entity_queue.append((e["name"], "location")) for e in step1_result.get("props", []): if e["name"] not in low_importance: entity_queue.append((e["name"], "prop")) # ── Step 2b: GPT 상세 추출 (1회 호출) ── if cp.completed_step and cp.completed_step >= "step2b": gpt_details = cp.get("gpt_details", {}) logger.info("Step 2b: restored from checkpoint (%d entities)", len(gpt_details)) else: logger.info("Step 2b: GPT detail extraction (%d entities)", len(entity_queue)) gpt_details: Dict[str, Dict[str, Any]] = {} entity_list_text = "\n".join(f"- {name} ({etype})" for name, etype in entity_queue) try: detail_schema = _load_schema("turn1_7_detail_batch_schema.json") detail_prompt = _load_prompt( "turn1_7_detail_batch", entity_list=entity_list_text, fulltext=fulltext, ) batch_result = call_structured( step="entity_review", # GPT 사용 system_prompt="시나리오 분석 전문가. 요소별 시각적 상세 정보를 최대한 많이 추출한다.", user_prompt=detail_prompt, response_schema=detail_schema, project_config=project_llm_config, schema_name="entity_detail_batch", ) for ent in batch_result.get("entities", []): gpt_details[ent["name"]] = { "description": ent.get("description", ""), "visual_traits": ent.get("visual_traits", []), } logger.info("Step 2b: GPT returned %d/%d entities", len(gpt_details), len(entity_queue)) # 누락 재시도 missing = [(n, t) for n, t in entity_queue if n not in gpt_details] if missing: logger.warning("Step 2b: %d missing, retrying", len(missing)) missing_text = "\n".join(f"- {n} ({t})" for n, t in missing) retry_prompt = _load_prompt( "turn1_7_detail_batch", entity_list=missing_text, fulltext=fulltext, ) try: retry_result = call_structured( step="entity_review", system_prompt="시나리오 분석 전문가. 요소별 시각적 상세 정보를 최대한 많이 추출한다.", user_prompt=retry_prompt, response_schema=detail_schema, project_config=project_llm_config, schema_name="entity_detail_retry", ) for ent in retry_result.get("entities", []): gpt_details[ent["name"]] = { "description": ent.get("description", ""), "visual_traits": ent.get("visual_traits", []), } except Exception as retry_exc: logger.warning("Step 2b retry failed: %s", retry_exc) except Exception as exc: logger.warning("Step 2b GPT detail failed: %s", exc) cp.save("step2b", step1_result=step1_result, review_result=review_result, gpt_details=gpt_details) # ── Step 3: T2I 프롬프트 생성 (병렬) ── characters: List[Dict[str, Any]] = [] locations: List[Dict[str, Any]] = [] props: List[Dict[str, Any]] = [] total = len(entity_queue) if total == 0: logger.info("No entities for T2I generation") else: system_prompt = _load_system() def _gen_t2i(idx: int, ename: str, etype: str) -> Tuple[int, str, str, Dict[str, Any]]: gpt_detail = gpt_details.get(ename, {}) extra = "" if gpt_detail: desc = gpt_detail.get("description", "") traits = ", ".join(gpt_detail.get("visual_traits", [])) extra = f"\n\n[시나리오 기반 상세 정보]\n설명: {desc}\n시각적 특징: {traits}" turn_msg = _load_prompt( "turn_entity_detail", entity_name=ename, entity_type=etype, ) + extra for attempt in range(3): try: detail = call_structured( step="entity_t2i", system_prompt=system_prompt, user_prompt=turn_msg, response_schema=ENTITY_DETAIL_SCHEMA, project_config=project_llm_config, schema_name="entity_t2i", opik_metadata={"tags": ["entity_t2i", ename]}, ) return (idx, ename, etype, { "name": detail.get("name", ename), "description": detail.get("description", ""), "visual_traits": detail.get("visual_traits", []), "t2i_prompt": detail.get("t2i_prompt", ""), # D6 T2: LLM 출력의 metadata_json forward (location 만 의미 있음). "metadata_json": detail.get("metadata_json") or {}, }) except Exception as exc: logger.warning("Entity T2I %s attempt %d failed: %s", ename, attempt + 1, exc) if attempt < 2: time.sleep(2 * (attempt + 1)) return (idx, ename, etype, { "name": ename, "description": "", "visual_traits": [], "t2i_prompt": "", "metadata_json": {}, }) # Step 3 체크포인트에서 이미 완료된 엔티티 복원 step3_done: Dict[str, Dict[str, Any]] = cp.get("step3_done", {}) remaining_queue = [(i, n, t) for i, (n, t) in enumerate(entity_queue) if n not in step3_done] # 이미 완료된 것 복원 for ename, data in step3_done.items(): etype = data.get("entity_type", "character") if etype == "character": characters.append(data) elif etype == "location": locations.append(data) else: props.append(data) if remaining_queue: max_workers = min(10, len(remaining_queue)) logger.info("Step 3: T2I generation — %d entities (%d already done), %d workers", total, len(step3_done), max_workers) results: List[Tuple[int, str, str, Dict[str, Any]]] = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = {} for i, ename, etype in remaining_queue: if futures: time.sleep(1) futures[executor.submit(_gen_t2i, i, ename, etype)] = i for future in as_completed(futures): result = future.result() results.append(result) # 개별 체크포인트 저장 _, ename, etype, data = result data["entity_type"] = etype step3_done[ename] = data cp.save("step3", step1_result=step1_result, review_result=review_result, gpt_details=gpt_details, step3_done=step3_done) results.sort(key=lambda r: r[0]) for _idx, _ename, etype, data in results: if etype == "character": characters.append(data) elif etype == "location": locations.append(data) else: props.append(data) else: logger.info("Step 3: all %d entities restored from checkpoint", len(step3_done)) logger.info("Entity extraction complete: %d chars, %d locs, %d props", len(characters), len(locations), len(props)) cp.delete() return { "characters": characters, "locations": locations, "props": props, "style": style, }