"""인물 + 변형 체이닝 추출 — 3000자 번들 + 2000자 참조 + 이전 결과 누적.""" import json, sys, time sys.path.insert(0, ".") from app.modules.llm.llm_client import call_structured from app.core.config import settings PID = "c0f47e88-b9f7-43a5-9e73-66b0f965b247" EID = "d2d203b8-4865-40f0-b5a6-e5990fa3effd" CP = f"{settings.projects_dir}/{PID}/checkpoints/episodes/{EID}" seg = json.load(open(f"{CP}/scene_save/manifest.json")) from app.core.database import SessionLocal from app.models.project import Episode from sqlalchemy.orm import undefer db = SessionLocal() ep = db.query(Episode).options(undefer(Episode.fulltext)).filter(Episode.id == EID).first() ft = ep.fulltext db.close() # 씬 텍스트 준비 scene_texts = [] for s in seg["data"]["segments"]: text = s.get("text") or ft[s["start_char"]:s["end_char"]] scene_texts.append({"idx": s["scene_index"], "text": text, "len": len(text)}) BUNDLE_TARGET = 3000 REF_MAX = 2000 schema = { "type": "object", "properties": { "characters": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string", "description": "인물 이름"}, "appearance_count": {"type": "integer", "description": "시각적 출현 횟수 (이번 씬 묶음 내에서)"}, "variants": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "variant_type": { "type": "string", "enum": ["face", "full"] } }, "required": ["name", "variant_type"], "additionalProperties": False } } }, "required": ["name", "appearance_count", "variants"], "additionalProperties": False } } }, "required": ["characters"], "additionalProperties": False } system = """시나리오에서 등장하는 인물을 나열하고, 변형이 있다면 그 형태를 따로 나열하세요. ## 인물의 정의 - 머리와 몸이 구분되는 모든 존재 (인간, 요괴, 동물, 로봇, 외계인 등) - 단, 서사가 있어야 함 (행동이나 대사가 있는 존재만) ## 변형의 정의 변형은 **인물에서 파생된 다른 외형**만 해당합니다: - 얼굴이 크게 달라지는 변화 → variant_type: "face" 예: 나이 변화(어린이↔성인), 변장/성형 전후, 빙의로 얼굴이 바뀜 - 시각적으로 얼굴과 몸을 구분할 수 없는 존재로의 변형, 구분 가능하지만 시나리오상에서 옷이 필요하지 않는 경우 → variant_type: "full" ## 변형이 아닌 것 (절대 포함 금지) - 일시적 상태: 부상, 출혈, 창백해짐 등 - 비유적 표현: "해골처럼", "짐승 같은" 등 ~처럼/~같은 - 복장/의상/감정 변화 ## 출력 규칙 - name: 인물의 기본 이름 - appearance_count: 이번 씬 묶음 내에서 시각적으로 출현한 횟수 - variants: 변형 목록 (이름 + variant_type) - 변형 이름은 시나리오에서 사용된 짧은 이름/호칭만 - 변형이 없는 인물은 variants를 빈 배열로 - 이미 추출된 인물 목록이 주어지면, 새로 등장하는 인물만 추가하고 기존 인물은 출력하지 말 것""" def build_bundles(scene_texts): """씬을 3000자 이하 번들로 묶기.""" bundles = [] i = 0 while i < len(scene_texts): bundle = [] bundle_len = 0 while i < len(scene_texts) and bundle_len + scene_texts[i]["len"] <= BUNDLE_TARGET: bundle.append(scene_texts[i]) bundle_len += scene_texts[i]["len"] i += 1 if not bundle and i < len(scene_texts): bundle.append(scene_texts[i]) i += 1 bundles.append(bundle) return bundles def build_ref_text(scenes, up_to_idx, max_chars): """앞쪽 씬 참조 텍스트 (max_chars 이내).""" ref_parts = [] ref_len = 0 for j in range(up_to_idx - 1, -1, -1): t = scenes[j]["text"] if ref_len + len(t) > max_chars: break ref_parts.insert(0, f"[씬 {scenes[j]['idx']}]\n{t}") ref_len += len(t) return "\n\n".join(ref_parts) if ref_parts else "" def run_chain(step_name, label): bundles = build_bundles(scene_texts) all_characters = {} # name → {appearance_count, variants} print(f"\n{'='*60}") print(f"=== {label} ({len(bundles)} bundles) ===") print(f"{'='*60}") scenes_so_far = 0 for bi, bundle in enumerate(bundles): # 현재 번들 텍스트 bundle_text = "\n\n".join(f"[씬 {s['idx']}]\n{s['text']}" for s in bundle) # 앞쪽 참조 ref_text = build_ref_text(scene_texts, scenes_so_far, REF_MAX) # 이미 추출된 인물 목록 existing_list = "" if all_characters: lines = [] for name, info in all_characters.items(): vstr = "" if info["variants"]: vstr = " → " + ", ".join(f"{v['name']}({v['variant_type']})" for v in info["variants"]) lines.append(f"- {name} (출현 {info['appearance_count']}회){vstr}") existing_list = "\n[이미 추출된 인물 — 중복 추출 금지]\n" + "\n".join(lines) # 프롬프트 조합 user_prompt = "" if ref_text: user_prompt += f"[앞쪽 씬 참조]\n{ref_text}\n\n" user_prompt += f"[분석 대상 씬]\n{bundle_text}" if existing_list: user_prompt += f"\n\n{existing_list}" scene_range = f"씬{bundle[0]['idx']}~{bundle[-1]['idx']}" try: result = call_structured( step=step_name, system_prompt=system, user_prompt=user_prompt, response_schema=schema, schema_name=f"char_chain_{step_name}_{bi}", ) new_chars = result.get("characters", []) # 누적 for c in new_chars: name = c["name"] if name in all_characters: all_characters[name]["appearance_count"] += c.get("appearance_count", 0) # 새 변형 추가 existing_vnames = {v["name"] for v in all_characters[name]["variants"]} for v in c.get("variants", []): if v["name"] not in existing_vnames: all_characters[name]["variants"].append(v) else: all_characters[name] = { "appearance_count": c.get("appearance_count", 0), "variants": c.get("variants", []), } print(f" 번들{bi+1} ({scene_range}): +{len(new_chars)}명 → 누적 {len(all_characters)}명") except Exception as e: print(f" 번들{bi+1} ({scene_range}): ERROR {e}") scenes_so_far += len(bundle) # 최종 결과 print(f"\n--- {label} 최종: {len(all_characters)}명 ---") for name, info in sorted(all_characters.items(), key=lambda x: -x[1]["appearance_count"]): if info["variants"]: vstr = ", ".join(f"{v['name']}({v['variant_type']})" for v in info["variants"]) print(f" {name:20s} [{info['appearance_count']:3d}회] → {vstr}") else: print(f" {name:20s} [{info['appearance_count']:3d}회]") return all_characters print(f"총 {len(scene_texts)}씬, {sum(s['len'] for s in scene_texts)}자") gemini_result = run_chain("beat_extract", "Gemini Pro") gpt_result = run_chain("visual_world_rules", "GPT")