scripts/build_flf25.py

"""Build LTX-2.5 first-and-last-frame graphs in ComfyUI API format.

WHY THIS EXISTS
LTX-2.5 has a stock "First & Last Frame to Video" workflow, but it is a UI
subgraph (awkward to queue programmatically) and it *generates* audio from an
empty latent. The image+audio path that exists for 2.3 was rebuilt for 2.5 as a
hybrid (first frame via LTXVImgToVideoInplace, last via LTXVAddGuide). This
unifies both into one API-format generator using the stock guide pattern —
BOTH frames through chained LTXVAddGuide at frame_idx 0 and -1 — and adds the
two things neither had: a LoRA slot and a separate audio save.

TWO MODES
  gen    LTXVEmptyLatentAudio — the model generates the audio along with the
         video. This is the mode for "write the line in the prompt, let the
         base model speak it", whose output you then feed to the AKUSPACE a2a
         pass. The base model's own reverb is inconsistent, which is exactly
         the before/after story.
  drive  Existing audio is VAE-encoded and pinned by a zero SolidMask, so the
         video denoises around fixed audio. Use when the audio is already final
         (e.g. an a2a pass has already roomed it).

TWO PRESETS — PICK THE ONE THAT MATCHES THE WEIGHTS
LTX-2.5 ships two transformers that need opposite sampling regimes, and mixing
them produces soft, washed-out frames that look like a broken model but are just
the wrong preset:

  distilled (default)  int8 distilled transformer + int8 gemma, ~8 steps with
                       explicit ManualSigmas, LTXVDualCFGGuider at 1/1,
                       SamplerEulerAncestral. This mirrors the shipped ComfyUI
                       preset and is the regime verified to produce clean output.
  dev                  bf16 dev transformer + bf16 gemma, ~24 steps at CFG 4
                       through LTXVScheduler and plain CFGGuider.

An earlier revision of this file argued for CFGGuider over LTXVDualCFGGuider on
the grounds that the latter was absent from a schema dump. That was wrong: the
dump had been taken before the node pack finished loading. Both are present on a
correctly installed pack.

Every node type and input name below is validated against a live object_info
dump before the graph is written — see validate(). Never ship a graph that has
not been through it; wrong input names are the trap that costs the most time.
"""

import argparse
import json
import sys
from pathlib import Path

W, H = 960, 544
TRANSFORMER = "ltx-2.5-22b-dev-transformer-bf16.safetensors"
TEXT_ENCODER = "gemma4-12b-with-proj-ltx-2.5-bf16.safetensors"
# The distilled pair the shipped preset uses. Different sampling regime — see
# WORKING_CONFIG.md. Mixing the two is what produces "distorted" output.
TRANSFORMER_DISTILLED = "ltx-2.5-22b-distilled-transformer-comfy-int8-convrot.safetensors"
TEXT_ENCODER_INT8 = "gemma4-12b-with-proj-ltx-2.5-comfy-int8-convrot.safetensors"
SIGMAS_DISTILLED = "1.0, 0.99375, 0.9875, 0.98125, 0.975, 0.909375, 0.725, 0.421875, 0.0"
VIDEO_VAE = "ltx-2.5-video-vae-bf16.safetensors"
AUDIO_VAE = "ltx-2.5-audio-vae-bf16.safetensors"
NEGATIVE = ("distorted, clipped, metallic, unintelligible speech, changed words, "
            "changed rhythm, deformed hands, extra fingers, watermark, text")


def build(a) -> dict:
    g: dict = {}

    def n(nid, cls, **inputs):
        g[str(nid)] = {"class_type": cls, "inputs": inputs}

    unet = TRANSFORMER_DISTILLED if a.preset == "distilled" else TRANSFORMER
    clip = TEXT_ENCODER_INT8 if a.preset == "distilled" else TEXT_ENCODER
    n(1, "UNETLoader", unet_name=unet, weight_dtype="default")
    model = ["1", 0]
    if a.lora:
        n(2, "LoraLoaderModelOnly", model=["1", 0], lora_name=a.lora, strength_model=a.lora_strength)
        model = ["2", 0]

    n(3, "CLIPLoader", clip_name=clip, type="ltxv")
    n(4, "CLIPTextEncode", clip=["3", 0], text=a.prompt)
    n(5, "CLIPTextEncode", clip=["3", 0], text=a.negative)
    n(6, "LTXVConditioning", positive=["4", 0], negative=["5", 0], frame_rate=float(a.fps))

    n(7, "VAELoader", vae_name=VIDEO_VAE)
    n(8, "VAELoader", vae_name=AUDIO_VAE)

    # first frame
    n(9, "LoadImage", image=a.first_image)
    n(10, "ImageScale", image=["9", 0], upscale_method="lanczos", width=a.width, height=a.height, crop="center")
    n(11, "LTXVPreprocess", image=["10", 0], img_compression=18)
    # last frame
    n(12, "LoadImage", image=a.last_image)
    n(13, "ImageScale", image=["12", 0], upscale_method="lanczos", width=a.width, height=a.height, crop="center")
    n(14, "LTXVPreprocess", image=["13", 0], img_compression=18)

    n(15, "EmptyLTXVLatentVideo", width=a.width, height=a.height, length=a.frames, batch_size=1)

    # Chained guides: the second consumes the first's conditioning AND latent.
    # frame_idx -1 is the stock idiom for "last"; it is what the shipped 2.5
    # flf2v workflow uses, so it is verified usage rather than an assumption.
    n(16, "LTXVAddGuide", positive=["6", 0], negative=["6", 1], vae=["7", 0],
      latent=["15", 0], image=["11", 0], frame_idx=0, strength=a.guide_strength)
    n(17, "LTXVAddGuide", positive=["16", 0], negative=["16", 1], vae=["7", 0],
      latent=["16", 2], image=["14", 0], frame_idx=-1, strength=a.guide_strength)

    if a.mode == "gen":
        n(18, "LTXVEmptyLatentAudio", frames_number=a.frames, frame_rate=float(a.fps),
          batch_size=1, audio_vae=["8", 0])
        audio_latent = ["18", 0]
    else:
        n(18, "LoadAudio", audio=a.audio)
        n(19, "TrimAudioDuration", audio=["18", 0], start_index=0.0,
          duration=round(a.frames / a.fps, 3))
        n(20, "LTXVAudioVAEEncode", audio=["19", 0], audio_vae=["8", 0])
        n(21, "SolidMask", value=0.0, width=1024, height=1024)
        n(22, "SetLatentNoiseMask", samples=["20", 0], mask=["21", 0])
        audio_latent = ["22", 0]

    n(30, "LTXVConcatAVLatent", video_latent=["17", 2], audio_latent=audio_latent)
    n(31, "RandomNoise", noise_seed=a.seed)

    if a.preset == "distilled":
        # The regime the shipped preset uses, and the only one confirmed to
        # produce clean frames. Distilled wants ~8 steps at CFG 1 with explicit
        # sigmas; feeding it dev-style 24 steps at CFG 4 yields soft, washed-out
        # output that looks like a broken model but is just the wrong preset.
        n(32, "LTXVDualCFGGuider", model=model, positive=["17", 0], negative=["17", 1],
          video_cfg=a.video_cfg, audio_cfg=a.audio_cfg)
        n(33, "SamplerEulerAncestral", eta=a.eta, s_noise=a.s_noise)
        n(34, "ManualSigmas", sigmas=a.sigmas)
    else:
        n(32, "CFGGuider", model=model, positive=["17", 0], negative=["17", 1], cfg=a.cfg)
        n(33, "KSamplerSelect", sampler_name="euler")
        n(34, "LTXVScheduler", latent=["30", 0], steps=a.steps, max_shift=2.05,
          base_shift=0.95, stretch=True, terminal=0.1)

    n(35, "SamplerCustomAdvanced", noise=["31", 0], guider=["32", 0], sampler=["33", 0],
      sigmas=["34", 0], latent_image=["30", 0])

    n(36, "LTXVSeparateAVLatent", av_latent=["35", 0])
    # Crop the guide frames out of the video latent before decoding.
    n(37, "LTXVCropGuides", positive=["17", 0], negative=["17", 1], latent=["36", 0])
    n(38, "VAEDecodeTiled", samples=["37", 2], vae=["7", 0], tile_size=512,
      overlap=64, temporal_size=64, temporal_overlap=8)
    n(39, "LTXVAudioVAEDecode", samples=["36", 1], audio_vae=["8", 0])

    n(40, "CreateVideo", images=["38", 0], audio=["39", 0], fps=float(a.fps))
    n(41, "SaveVideo", video=["40", 0], filename_prefix=a.out_prefix, format="auto", codec="auto")
    # Audio saved separately so it can be round-tripped through the AKUSPACE
    # a2a pass and recombined — that pair IS the before/after example.
    n(42, "SaveAudio", audio=["39", 0], filename_prefix=a.out_prefix + "_audio")
    return g


def validate(graph: dict, object_info: Path) -> list[str]:
    """Check every class and input name against a live object_info dump."""
    oi = json.loads(object_info.read_text())
    errs = []
    for nid, node in graph.items():
        cls = node["class_type"]
        if cls not in oi:
            errs.append(f"node {nid}: class '{cls}' not in object_info")
            continue
        spec = oi[cls]["input"]
        allowed = set(spec.get("required") or {}) | set(spec.get("optional") or {})
        for name in node["inputs"]:
            if name not in allowed:
                errs.append(f"node {nid} ({cls}): unknown input '{name}' — has {sorted(allowed)}")
        for name in (spec.get("required") or {}):
            if name not in node["inputs"]:
                errs.append(f"node {nid} ({cls}): MISSING required input '{name}'")
    # every reference must point at a node that exists
    for nid, node in graph.items():
        for name, val in node["inputs"].items():
            if isinstance(val, list) and len(val) == 2 and isinstance(val[0], str):
                if val[0] not in graph:
                    errs.append(f"node {nid}.{name} -> dangling reference to '{val[0]}'")
    return errs


def main() -> int:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--mode", choices=["gen", "drive"], default="gen")
    p.add_argument("--out", required=True)
    p.add_argument("--prompt", default="PLACEHOLDER scene description")
    p.add_argument("--negative", default=NEGATIVE)
    p.add_argument("--first-image", default="frame_first.png")
    p.add_argument("--last-image", default="frame_last.png")
    p.add_argument("--audio", default="track.wav", help="drive mode only")
    p.add_argument("--frames", type=int, default=241, help="241 @ 24fps = 10.04 s")
    p.add_argument("--fps", type=int, default=24)
    p.add_argument("--width", type=int, default=W)
    p.add_argument("--height", type=int, default=H)
    p.add_argument("--steps", type=int, default=24)
    p.add_argument("--cfg", type=float, default=4.0)
    p.add_argument("--seed", type=int, default=42)
    p.add_argument("--guide-strength", type=float, default=0.7,
                   help="0.7 is the stock flf2v value")
    p.add_argument("--lora", default=None, help="e.g. akuspace-v5/lora_weights_step_11500.safetensors")
    p.add_argument("--lora-strength", type=float, default=1.0)
    p.add_argument("--out-prefix", default="v5video/flf25")
    p.add_argument("--preset", choices=["distilled", "dev"], default="distilled",
                   help="distilled = the shipped, verified-good regime (default)")
    p.add_argument("--video-cfg", type=float, default=1.0)
    p.add_argument("--audio-cfg", type=float, default=1.0)
    p.add_argument("--eta", type=float, default=0.0)
    p.add_argument("--s-noise", type=float, default=1.0)
    p.add_argument("--sigmas", default=SIGMAS_DISTILLED)
    p.add_argument("--object-info", default=None, help="live object_info.json to validate against")
    a = p.parse_args()

    graph = build(a)

    if a.object_info:
        errs = validate(graph, Path(a.object_info))
        if errs:
            print(f"VALIDATION FAILED ({len(errs)}):", file=sys.stderr)
            for e in errs:
                print("  " + e, file=sys.stderr)
            return 1
        print(f"validated OK against {a.object_info}")
    else:
        print("WARNING: not validated — pass --object-info to check against a live install")

    Path(a.out).write_text(json.dumps(graph, indent=2))
    print(f"{a.out}: {len(graph)} nodes, mode={a.mode}, "
          f"{a.frames}f @ {a.fps}fps = {a.frames/a.fps:.2f}s, lora={a.lora or 'none'}")
    return 0


if __name__ == "__main__":
    sys.exit(main())