03 파이프라인 그래프: 악역영애 말포이 영상 제작 (LangGraph 스타일)

• 대상: 10-07~10-08 실제 진행한 5편(1편 S-RANK PULL · 2편 AGENT SHOWCASE · 3편 로판 OP · 4편 HYPER EDIT · 5편 로판 OP 영어판)
• 근거: STORYBOARD.md · V1-EDL.md · V3-STORYBOARD.md · V4-BRIEF.md · 06_flow/FLOW-UI.md·각 LOG.md · 05_matte/README.md · 04_audio README 3종 · 03_motion README 3종 · _tools · 07_render/v1~v5 tools · 사용자 피드백 기록
• 읽는 법: 사각형 = 노드(일 하는 단계), 마름모 = 조건 분기(게이트), 원 = 병렬 분기·합류, 빨간 육각형 = 사용자 interrupt. 노드 둘째 줄 〔 〕 = 담당
• 담당 6종: 메인 Claude(오케스트레이터·판정) / Claude 서브에이전트(워커) / codex(이미지 생성) / Flow(Aside 브라우저로 조작) / ElevenLabs API / 로컬 GPU·CPU

① 전체 그래프

• 흐름 한 줄: 기획 → 병렬 분기 3갈래(이미지→Flow→매팅 / 오디오 / 코드 배경 블록) → 합류 → 편별 합성 → 드래프트 판정 루프 → 최종 렌더 → 사용자 interrupt → 승인 시 인계, 다음 편은 스토리보드로 복귀
• 루프 4개
• 이미지 이름 교정 루프: 파일명과 그림 내용이 어긋나면 교정 후 재대조
• Flow 재생성 루프: 클립 판정 실패 시 1회만 재생성(_r2), 그래도 흠이면 "채택(주의)"로 내려보내고 후반에서 가림
• 오디오 후보 루프: 기준 충족 후보가 없으면 요청 JSON 고쳐 다시 생성
• 컨택트시트 루프: 드래프트 540p에서 배치·타이밍 어긋나면 합성 단계로 복귀(4편은 r1→r2→r3 3회)
• 조건 분기: 360p 테스트 게이트 / 매팅 GPU 성공 여부 / 인월드 컷은 매팅 건너뛰고 바로 합류
• 오디오 → 배경 블록 점선: 워커 B 키네틱 월·비트 컷인이 beats.json 실측 박 시각을 상수로 박음(드롭 7.484)

사용자 interrupt 되돌림 지도

• 원칙: 반려는 "결과물 고치기"가 아니라 원인 노드로 되돌아가 거기서부터 다시 흐름
• 반영된 규칙은 State의 rules에 박혀 이후 편 전체에 적용(예: 노래는 영어 보컬 기본, 캐릭터물은 의상 여러 벌 기본)

② State 스키마 (그래프가 들고 다니는 상태)

• budget 실측 누계
• Flow: 테스트 10 + v1 88 + ref 88 + extra 71 + v3 59 + v4 70 = 386 크레딧 사용(1,050 → 664)
• ElevenLabs(character_count): 1편 3,101 + 3편 9,938 + 4편 1,324 + 영어판 5,112 = 19,475 사용
• 병렬 분기에서 여러 노드가 같은 키에 쓰는 건 리스트 누적(reducer)으로 합침: clips, mattes, blocks는 append

③ 노드 표

단계별 세부 워크플로

기획

• 리서치 3갈래: 밈 확산 경로(09-22 X 팬아트 → 10-02~06 유튜브), Flow 공식 문서(단가·워터마크·9:16), 핀터레스트 5분류(char·showcase·gacha·bg·romfan)
• 스토리보드가 정한 것: 컷 경계 = 곡 마디 격자, 레이어 3층(코드 배경 / 매팅 캐릭터 / 오버레이), 합성 컷은 블루 배경·드라마 컷은 인월드, 크레딧 예산(360p 테스트 ~100, 1편 ~150 …)
• 편별 결정표(EDL)에 레이어 순서·자막 위치(상단 y 120~260, 대사 y 1500~1650)·SFX 시각까지 고정 → 합성 워커는 이 표만 보고 조립

이미지

• imagegen_runner.py: codex exec 4개 병렬, jsonl 한 줄 = 한 장, 이미 있는 파일 건너뜀(재실행 안전)
• 1편 첫 프레임 8장 3.9분, 의상 7벌 5.0분, 3편 15장 6.6분(전부 실패 0)
• 회수 방식이 "공용 폴더에서 가장 최근·아직 안 가져간 PNG"라서 병렬일 때 이름이 엇갈릴 수 있음 → 스왑 교정 노드가 필요한 이유(아래 ⑥)

Flow

• 테스트: T1(8초 전신)·T2(4초 클로즈업) 360p, 10 크레딧. 교훈 = 프레임 모드 + 단색 블루 첫 프레임 + "background stays perfectly flat solid blue" + "One continuous shot, no scene cuts"
• 본편: 720p 프레임 모드 단가 4초 7 · 6초 10 · 8초 12. 소재 모드(@캐릭터 참조)는 동영상 참조 시 20, 블루가 그라데이션으로 무너지는 경향 → 합성 컷은 프레임 모드 우선
• 소요: 8초 영상 약 45초, 4초 약 28초. 동시 5~9개 제출해도 정책·레이트 경고 0
• 판정 기록: 컷 분할, 블루 유지, 문장(은색 뱀) 변형, 블루 위 VFX 잔여물(매팅 시 마스크 필요) 등을 LOG.md에 행 단위로
• 4편은 id 오배정 방지로 3중 대조(프롬프트 매칭 + 다운로드 제안 파일명 + 타일 의상), 오배정 0

매팅

• 하이브리드 = WSL 로컬 플레이트 색차 키(머리카락·망토 가장자리) + Windows GPU isnet-anime(가비지 매트·몸통 보호)
• GPU 분리 이유: WSL에 CUDA·ROCm 없음(GPU = AMD RX 6600). Windows 워커가 같은 영상을 직접 디코드해서 마스크만 돌려줌(WSL→Windows 파이프가 0.8MB/s라 프레임 전송이 병목)
• 속도: 하이브리드 1.25~1.46초/프레임(10초 클립 약 7분), GPU 추론 자체는 약 0.05초/장
• 큐 2라인: matte_queue.sh [SRC] [DST]를 배치별로 띄움(v1·ref·extra·v3·v4 로그 존재). 다운로드 중 파일은 20초 지나야 집음, 60분 새 파일 없으면 종료

오디오

• 곡: composition_plan 청크로 구조 지시 → 실측으로 검증. 모델이 브레이크·끝 히트 지시를 자주 무시 → 편집 마스터에서 직접 만듦
• 선택 기준 예: 1편 = BPM 128 근접 + 드롭 직전 숨 고르기, 3편 = 진짜 정지 브레이크 + 정시 훅 + 가사 인식, 4편 = 빌드업 대비 드롭 크기
• 3편 한국어 곡: 21후보 중 K9 선택(보컬 분리 후 인식 0.80/0.82) → 사용자 반려 → 영어 12후보 중 E11 → 5편
• VO: 1편 Emmaline(10대 음역 + 영국 억양), 3편 Chaemong·Taemin. "gems"를 "gens"로 읽는 테이크는 ASR로 걸러 버림

코드 배경 블록

• 워커 3기 분담: A(가챠·S랭크 카드·이름판·공용 mm-lib.js) / B(키네틱 월·비트 컷인) / C(엔딩 도장·클립월)
• 공통 규약: 1080x1920, 박 단위 타이밍, layer 변수(bg/fg)로 같은 블록을 캐릭터 아래·위 두 번 올림
• 근거 기법은 awesome-ai-motion slug로 README에 남김(kinetic-beats, glitch-rgb-split, band-slide 등)

합성·렌더

• 1편: HyperFrames 배경 패스(mp4+오디오) → 알파 webm 오버레이 → 전경 패스(알파 mov) → 줌 펀치·글리치·그레인·비네트 → -14 LUFS
• 3·5편: bgchar.mp4 위에 ffmpeg overlay로 캐릭터 컷만 불투명 mp4로 구운 뒤 HyperFrames 본 조립
• 4편: edl.py(박 격자 → 컷 슬롯·의상 순환·포니 약 40%) → comp4.py(프레임마다 배경 판·타이포·알파 캐릭터·가상 카메라를 numpy로 합성)
• 최종: finalize.sh = 2패스 loudnorm -14 LUFS / TP -1 + 1초 간격 컨택트시트

④ 시퀀스: 1편 S4 한 컷이 이미지에서 합성까지

• 실제 값: S4 첫 프레임 02_frames/S4_fullbody.png → Flow 8초 12 크레딧, 18:28 채택 → 05_matte/v1/V1_S4.webm → 본편 7.484초(드롭)부터 소스 0.2~7.7초 전신 배치(높이 1920의 86%, 하단 정렬)
• 메인 Claude가 직접 손대는 건 대조·판정 2지점(3번, 19번)뿐, 나머지는 워커·큐가 처리

⑤ 실제 LangGraph 코드로 옮긴다면 (의사코드, 실행 안 함)

# 의사코드: 실제 파이프라인은 LangGraph 없이 메인 Claude 오케스트레이션 + 셸 큐로 돌렸음
from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send, interrupt, Command

class S(TypedDict):
    storyboard: dict; rules: list; assets: dict; budget: dict; feedback: str
    clips: Annotated[list, operator.add]; mattes: Annotated[list, operator.add]
    blocks: Annotated[list, operator.add]; audio: dict; renders: dict

g = StateGraph(S)
for n in ["research", "storyboard", "design", "frames", "swap_fix", "flow_test",
          "flow_job", "matte", "audio", "block", "join", "composite", "draft", "final", "handoff"]:
    g.add_node(n, NODES[n])          # NODES = 각 노드 함수(codex·Aside·ElevenLabs·로컬 호출)

g.add_edge(START, "research"); g.add_edge("research", "storyboard")
# 병렬 분기: 이미지 갈래 1 + 오디오 1 + 블록 워커 3
g.add_conditional_edges("storyboard", lambda s: [Send("design", s), Send("audio", s)]
                        + [Send("block", {**s, "worker": w}) for w in "ABC"])
g.add_edge("design", "frames")
g.add_conditional_edges("frames", lambda s: "swap_fix" if names_mismatch(s) else "flow_test")
g.add_edge("swap_fix", "frames")      # 교정 후 재대조
g.add_conditional_edges("flow_test", lambda s: "frames" if not test_ok(s) else
                        [Send("flow_job", j) for j in s["storyboard"]["flow_jobs"]])   # 동시 5~6
g.add_conditional_edges("flow_job", route_clip)   # 재생성 1회 → flow_job / 블루 컷 → matte / 인월드 → join
g.add_edge(["matte", "audio", "block"], "join")   # 합류(fan-in)
g.add_edge("join", "composite"); g.add_edge("composite", "draft")
g.add_conditional_edges("draft", lambda s: "final" if contact_sheet_ok(s) else "composite")

def final(s):                         # heavy.sh 세마포어: 동시 2
    with HEAVY_SLOTS: render(s, workers=2)
    fb = interrupt({"upload": s["renders"]["upload"]})   # 사용자 피드백에서 멈춤
    if fb["approved"]: return Command(goto="handoff")
    return Command(goto=RETURN_MAP[fb["reason"]],        # 예: "korean_vocal" → "audio"
                   update={"rules": [fb["rule"]], "feedback": fb["text"]})

g.add_conditional_edges("handoff", lambda s: "storyboard" if s["storyboard"]["next"] else END)
app = g.compile(checkpointer=SqliteSaver("malfoy.db"))   # WSL 재부팅에도 재개

• 대응 관계: Send = 병렬 분기, Annotated[list, operator.add] = 합류 시 리스트 누적, interrupt = 사용자 피드백, Command(goto) = 되돌림 지도, checkpointer = 세션 끊김 후 재개(실제로는 산출 파일 존재 확인으로 대신함)

⑥ 실제로 막혔던 지점과 그래프상 대응 노드