# 파이프라인 리팩토링 v3 — DAG 기반 세부 단계 독립 실행 ## 이전 리뷰 반영 사항 - v2의 선형 체인 → DAG 의존성 - 누락 LLM 호출 추가 (GPT Vision 검증, outlook_merger 등) - Steps 14-17 → compound step으로 묶기 - 체크포인트 에피소드 스코핑 + 무효화 모델 - run_analysis 유지 (호환 래퍼) - Phase 0 추가 (manifest + DB schema) --- ## 단계 정의 — 20개 사용자 단계 + 내부 서브스텝 ### 분석 Phase (10단계) | # | step_id | 이름 | 기본 모델 | provider | depends_on | |---|---------|------|-----------|----------|------------| | 1 | entity_style | 스타일+요소이름 | gemini-pro | gemini | (없음 — fulltext만) | | 2 | entity_review | GPT 리뷰/필터 | gpt | openai | entity_style | | 3 | entity_detail_batch | GPT 상세 추출 | gpt | openai | entity_review | | 4 | entity_t2i | T2I 프롬프트 (병렬) | gemini-flash | gemini | entity_detail_batch | | 5 | scene_segmentation | 씬 세그먼테이션 | gemini-lite | gemini | (없음 — fulltext만) | | 6 | scene_split | 큰 씬 분할 | gemini-flash | gemini | scene_segmentation | | 7 | scene_dependency | 씬 연관 분석 | gemini-pro | gemini | scene_split | | 8 | outlook_extraction | 아웃룩 추출 | gemini-pro | gemini | scene_split, entity_t2i | | 9 | scene_detail | 씬 상세 분석 (병렬) | gpt | openai | scene_dependency, outlook_extraction, entity_t2i | | 10 | scene_verify | 교차 검증 (병렬) | gemini-pro | gemini | scene_detail | ### 이미지 Phase (7단계) | # | step_id | 이름 | 기본 모델 | provider | depends_on | |---|---------|------|-----------|----------|------------| | 11 | world_guide | 월드 가이드 | gpt | openai | entity_t2i, scene_detail | | 12 | ref_image_gen | 요소 참조 이미지 | gemini-image | gemini | entity_t2i | | 13 | composite_image_gen | 합성 이미지 | gemini-image | gemini | ref_image_gen, outlook_extraction | | 14 | scene_image_pipeline | 씬 이미지 생성 (compound) | mixed | mixed | composite_image_gen, scene_verify, world_guide | Step 14 내부 서브스텝 (per-scene, 분리 불가): | sub | 이름 | 모델 | |-----|------|------| | 14a | prompt_translation | gemini-flash | | 14b | scene_t2i_gen | gemini-image | | 14c | scene_t2i_validation | gpt-vision | | 14d | prompt_sanitize (조건부) | gpt | | 14e | angle_recommend | gpt-vision | | 14f | fal_angle_apply | fal.ai | | 14g | final_select | gpt-vision | ### 보조 단계 (on-demand, 독립 실행) | step_id | 이름 | 기본 모델 | depends_on | |---------|------|-----------|------------| | outlook_dedup | 아웃룩 중복 판별 | gpt | outlook_extraction | | outlook_merge | 아웃룩 병합 (DB) | gpt | outlook_extraction | | project_summary | 프로젝트 요약 | gemini-pro | entity_style | | style_rules | 스타일 규칙 생성 | gpt | entity_style | --- ## DAG 의존성 그래프 ``` ┌─ entity_style ─── entity_review ─── entity_detail_batch ─── entity_t2i ─┐ │ │ fulltext ───────────┤ ├── scene_detail ── scene_verify │ │ │ └─ scene_segmentation ── scene_split ─── scene_dependency ─────────────────┘ │ │ │ └──── outlook_extraction ──────────────────────┐ │ │ │ entity_t2i ──────────────── ref_image_gen ─┐ │ │ ├── composite_image_gen ─┐ │ │ outlook_extraction ────────────────────────┘ │ │ │ ├──── scene_image_pipeline scene_verify ──────────────────────────────────────────────────────┘ │ world_guide (entity_t2i + scene_detail) ─────────────────────────────────┘ ``` --- ## Step 상태 모델 ``` pending — 아직 실행 안 됨 blocked — 의존 단계 미완료 runnable — 의존 단계 완료, 실행 가능 running — 실행 중 (progress 포함) completed — 완료 partial — 일부 완료 (fan-out: 32/50) failed — 실패 (에러 메시지 포함) not_applicable — 해당 없음 (예: scene_split 할 큰 씬 없음, fal_key 미설정) stale — 업스트림 재실행으로 무효화됨 cancelled — 사용자 취소 ``` --- ## DB Schema — step_run 테이블 (PipelineProgress 대체) ```sql CREATE TABLE step_run ( id TEXT PRIMARY KEY, project_id TEXT NOT NULL REFERENCES project_registry(id), episode_id TEXT NOT NULL REFERENCES episode(id), step_id TEXT NOT NULL, -- 'entity_style', 'scene_detail', etc. status TEXT NOT NULL DEFAULT 'pending', -- 위 상태 모델 run_id TEXT, -- 실행 식별자 (UUID) resolved_model TEXT, -- 실제 사용된 모델 input_hash TEXT, -- 입력 데이터 해시 (무효화용) upstream_revision TEXT, -- 의존 단계 최종 완료 시각 해시 prompt_version TEXT, -- 사용된 프롬프트 버전 applicable_count INTEGER, -- fan-out: 총 항목 수 completed_count INTEGER DEFAULT 0, failed_count INTEGER DEFAULT 0, error_message TEXT, started_at TEXT, completed_at TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, UNIQUE(project_id, episode_id, step_id) ); ``` --- ## 체크포인트 구조 ``` projects/{pid}/checkpoints/episodes/{eid}/ entity_style/ manifest.json -- {status, run_id, resolved_model, input_hash, result} entity_review/ manifest.json entity_detail_batch/ manifest.json entity_t2i/ manifest.json -- {status, completed: {name: data}, failed: {name: err}} scene_segmentation/ manifest.json -- {status, segments: [...]} scene_split/ manifest.json -- {status, segments_final: [...]} scene_dependency/ manifest.json outlook_extraction/ manifest.json scene_detail/ manifest.json -- {status, completed: {scene_index: data}, failed: {}} scene_verify/ manifest.json ref_image_gen/ manifest.json -- {status, completed: {entity_id: {path, asset_id}}, failed: {}} composite_image_gen/ manifest.json scene_image_pipeline/ manifest.json -- {status, completed: {still_id: {paths, primary_id}}, failed: {}} ``` 무효화 규칙: - 업스트림 step 재실행 시 → 다운스트림 step status를 `stale`로 변경 - `input_hash` = md5(입력 데이터 요약 + resolved_model + prompt_version) - resume 시 `input_hash` 비교 → 다르면 재실행 --- ## API 설계 ``` # 단계별 실행 POST /api/v1/projects/{pid}/episodes/{eid}/steps/{step_id} ?mode=resume (기본: 완료면 스킵, partial이면 이어서) ?mode=force (기존 결과 stale 처리 + 다운스트림도 stale + 재실행) Response: 202: {"ok": true, "run_id": "...", "status": "started"} 200: {"ok": true, "status": "skipped", "reason": "already completed"} 200: {"ok": true, "status": "not_applicable", "reason": "..."} 400: {"error": "gate.blocked", "blocked_by": ["entity_review"], "message": "..."} 409: {"error": "step.already_running"} # 전체 상태 조회 (DAG + 각 단계 상태) GET /api/v1/projects/{pid}/episodes/{eid}/steps Response: { "steps": [ {"step_id": "entity_style", "status": "completed", "model": "gemini-pro", "completed_count": 1, "total_count": 1, "completed_at": "...", "depends_on": [], "can_run": true}, {"step_id": "entity_review", "status": "runnable", "model": "gpt", "depends_on": ["entity_style"], "can_run": true}, ... ] } # 개별 단계 결과 조회 GET /api/v1/projects/{pid}/episodes/{eid}/steps/{step_id}/result # fan-out 단계의 개별 항목 재실행 POST /api/v1/projects/{pid}/episodes/{eid}/steps/{step_id}/items/{item_id} # 전체 실행 (기존 호환 — 내부적으로 step 순차 호출) POST /api/v1/projects/{pid}/episodes/{eid}/analyze -- steps 1-10 순차 POST /api/v1/projects/{pid}/episodes/{eid}/generate-images -- steps 11-14 순차 # 모델 설정 (기존 확장) GET /api/v1/projects/{pid}/llm-config -- 20개 단계 + 서브스텝 전부 표시 PUT /api/v1/projects/{pid}/llm-config -- 개별 모델 변경 ``` --- ## Step Manifest (코드 내 정의) ```python STEP_MANIFEST = { "entity_style": { "label": "스타일+요소이름", "category": "analysis", "default_model": "gemini-pro", "provider": "gemini", "depends_on": [], "fan_out": False, # 단일 호출 "checkpoint_writer": "entity_style_checkpoint", "applicability": "always", # 항상 실행 }, "entity_t2i": { "label": "T2I 프롬프트 생성", "category": "analysis", "default_model": "gemini-flash", "provider": "gemini", "depends_on": ["entity_detail_batch"], "fan_out": True, # entity별 병렬 "checkpoint_writer": "entity_t2i_checkpoint", "applicability": "always", }, "scene_split": { "label": "큰 씬 분할", "category": "analysis", "default_model": "gemini-flash", "provider": "gemini", "depends_on": ["scene_segmentation"], "fan_out": True, "applicability": "if_large_scenes", # 큰 씬 없으면 not_applicable }, "scene_verify": { "label": "교차 검증", "category": "analysis", "default_model": "gemini-pro", "provider": "gemini", "depends_on": ["scene_detail"], "fan_out": True, "applicability": "if_multi_char_scenes", # 캐릭터 2+ 씬만 }, "scene_image_pipeline": { "label": "씬 이미지 생성", "category": "image", "default_model": "mixed", "provider": "mixed", "depends_on": ["composite_image_gen", "scene_verify", "world_guide"], "fan_out": True, # scene별, 배치 순서 보장 "sub_steps": ["prompt_translation", "scene_t2i_gen", "scene_t2i_validation", "prompt_sanitize", "angle_recommend", "fal_angle_apply", "final_select"], "applicability": "always", }, # ... 나머지 동일 패턴 } ``` --- ## StepRunner 베이스 클래스 ```python class StepRunner: def __init__(self, step_id, project_id, episode_id, db, project_config=None): self.step_id = step_id self.manifest = STEP_MANIFEST[step_id] self.db = db self.project_id = project_id self.episode_id = episode_id self.project_config = project_config def check_gate(self): """의존 단계 완료 확인. blocked이면 AppError.""" for dep in self.manifest["depends_on"]: dep_run = self._get_step_run(dep) if not dep_run or dep_run.status not in ("completed", "not_applicable"): raise AppError(f"gate.blocked", f"{dep} 미완료") def check_applicability(self): """이 단계가 적용 가능한지. not_applicable이면 스킵.""" # 서브클래스에서 구현 def load_checkpoint(self): """체크포인트 로드. completed/partial/None.""" def save_checkpoint(self, data): """체크포인트 저장 (원자적).""" def invalidate_downstream(self): """이 단계 재실행 시 다운스트림 stale 처리.""" for step_id, info in STEP_MANIFEST.items(): if self.step_id in info.get("depends_on", []): self._set_step_status(step_id, "stale") def run(self, mode="resume"): """실행. 서브클래스에서 _execute() 구현.""" self.check_gate() if mode == "resume": cp = self.load_checkpoint() if cp and cp["status"] == "completed": return {"status": "skipped"} elif mode == "force": self.invalidate_downstream() self._update_status("running") try: result = self._execute() self.save_checkpoint(result) self._update_status("completed") return result except Exception as exc: self._update_status("failed", error=str(exc)) raise def _execute(self): """서브클래스에서 구현 — 실제 LLM 호출.""" raise NotImplementedError ``` --- ## 구현 순서 ### Phase 0: 기반 (2-3일) - step_run DB 테이블 생성 - STEP_MANIFEST 정의 (의존성 DAG) - StepRunner 베이스 클래스 - 체크포인트 manifest 구조 - 무효화 로직 - GET /steps API (상태 조회) ### Phase 1: 분석 단계 (5-7일) - Steps 1-10 각각 StepRunner 서브클래스 - POST /steps/{step_id} API 라우터 - 기존 run_analysis → StepRunner 순차 호출 래퍼 - fan-out 단계 (entity_t2i, scene_detail, scene_verify) 병렬 + 체크포인트 ### Phase 2: 이미지 단계 (5-7일) - Steps 11-14 StepRunner 서브클래스 - Step 14 compound step (서브스텝 포함) - GPT Vision 호출 LiteLLM 전환 - fal.ai 호출 Opik 수동 span ### Phase 3: 프론트엔드 (3-4일) - PipelineStepsPanel (DAG 시각화 + 상태 + 모델 + 재실행) - LLMConfigPanel 확장 (20단계 + 서브스텝) - per-item 상태 표시 ### Phase 4: 정리 (2-3일) - 기존 모놀리식 코드 → StepRunner 래퍼로 대체 (삭제 안 함) - config key 마이그레이션 - 테스트: crash-resume, force 재실행 무효화, concurrent 실행 차단 --- ## 변경하지 않는 것 - 기존 run_analysis / generate_images API 유지 (내부를 StepRunner로 위임) - 기존 이미지 파일 삭제 안 함 - 기존 DB 데이터 유지