ltx-2-video

LTX-2 Video

如果工作流需要联合生成视频和音频、首帧／尾帧条件控制、音频驱动视频、局部重拍、视频扩展或官方自托管方案，请使用 LTX-2.3。请将托管 API 契约、本地权重契约和研究声明区分开来。

先制定计划

1. 说明所需模式、时长、宽高比、分辨率、帧率、音频要求、交付编解码器，以及输入是否包含人物或敏感数据。
2. 根据下方路由表选择 hosted 或 local。不要认为“open”就意味着免费、不受限制或易于运行。
3. 在发起任何付费请求、上传素材或下载模型前，先编写试运行计划。计划中应包含确切的端点／模型或检查点文件名、预估托管费用、输出路径和验证步骤。
4. 在首次计费 API 提交或下载多 GB 文件前，必须获得明确批准。切勿代替用户启用自动充值。
5. 保留溯源信息：提示词、可用时的种子、模型／检查点、API 请求／任务 ID、源素材哈希、生成参数、许可证审查日期和输出哈希。

有策略地选择路由

需求	首选路由	契约
托管 T2V/I2V，支持 4K 选项	托管异步 V2	ltx-2-3-fast 用于迭代或较长的 1080p 视频片段；ltx-2-3-pro 用于追求高保真
现有 2–20 s 音频驱动视觉内容	托管 A2V	仅限 Pro；必须提供提示词或图像；按输入音频秒数计费
替换视频中的某一段	托管重拍	仅限 Pro；按整个输入视频时长计费，而非仅按编辑片段时长计费
在开头／结尾添加内容	托管扩展	仅限 Pro；扩展部分和上下文均计费，且总帧数上限为 505
将 SDR 视频转换为 HDR 调色素材	托管 HDR	仅支持异步；返回包含线性 EXR 帧的 ZIP 文件，而非成品 MP4
数据控制、可复现性、自定义 LoRA、离线工作	官方本地仓库	自行负责 GPU、依赖项、检查点、Gemma 条款、输出流程和许可证合规
基于节点的本地工作流	官方 ComfyUI-LTXVideo	固定工作流和自定义节点版本；将其视为独立的执行环境

生产环境优先使用异步 V2。同步 V1 端点会在一个长请求中返回 MP4，仅适用于连接持续时间可接受的短时实验。截至 2026-07-10，官方变更日志仍将异步 V2 标记为 Beta；请隔离其适配器，并以稳健方式测试状态解析。

固定当前模型系列

使用托管模型 ID ltx-2-3-fast 和 ltx-2-3-pro。根据 2026-07-02 的公告，ltx-2-fast 和 ltx-2-pro 将于 2026-07-15 弃用，目前暂时由 2.3 按旧价格提供服务，并于 2026-08-15 移除。新工作中应拒绝使用这些旧 ID，并在该日期后重新查看弃用页面。

当前托管生成范围（于 2026-07-10 验证）：

• T2V/I2V Fast：1080p、24/25 fps 支持 6、8、10、12、14、16、18 或 20 秒；1080p、48/50 fps 以及所有 1440p/4K 组合支持 6、8 或 10 秒。
• T2V/I2V Pro：1080p、1440p 或 4K，在 24/25/48/50 fps 下支持 6、8 或 10 秒。
• 托管宽高比为 16:9 或 9:16。请使用文档中列出的确切尺寸，不要自行指定任意尺寸。
• generate_audio 默认值为 true。需要静音时，请明确设置。
• last_frame_uri 仅适用于 2.3 I2V 模型 ID。

不要将原始 LTX-2 论文中的 19B 架构套用到所有 2.3 产物上。2026 年 1 月的技术报告描述的是一个 14B 视频流和一个 5B 音频流。当前 Hugging Face 文件名对应的是 22B LTX-2.3 检查点。请注明每项声明对应的版本，不要跨版本进行参数量计算。

选择托管操作

使用 https://api.ltx.video 和 Authorization: Bearer $LTXV_API_KEY。

操作	提交	主要限制
文本生成视频	POST /v2/text-to-video	提示词不超过 5000 个字符；必须指定 model、duration 和 resolution
图像生成视频	POST /v2/image-to-video	需要首帧 image_uri；可选，仅 2.3 支持的 last_frame_uri
音频生成视频	POST /v2/audio-to-video	Pro；音频时长为 2–20 秒；必须提供图像或提示词；输出为 25 fps
重拍	POST /v2/retake	Pro；输入至少 73 帧，最高 4K；编辑时长至少 2 秒；可替换音频、视频或两者
扩展	POST /v2/extend	Pro；在 start 或 end 处增加 2–20 秒；输入为 16:9/9:16；保留输入分辨率
SDR 转 HDR	POST /v2/video-to-video-hdr	2.3 Pro；仅异步；结果键为 exr_frames_url

对于所有异步任务，在轮询 GET /v2/{operation}/{id}（每五秒一次）之前，先保存返回的 id。只接受 pending、processing、completed 或 failed 状态。任务完成后，读取 result.video_url；HDR 则读取 result.exr_frames_url。任务状态和输出 URL 会在终止状态后 24 小时过期。请尽快下载，在本地计算哈希，然后进行验证。

公共 API 未说明幂等键。如果提交请求超时且没有收到响应，请勿盲目再次执行 POST：检查开发者控制台，或向支持团队提供时间戳和任何 x-request-id。可能无法确认首次请求是否创建了计费任务。

遵守 429 响应中的 Retry-After，并区分 concurrency_limit_error 和 rate_limit_error。公布的默认并发数为 2，但配额因套餐而异。对 GET 请求的重试次数设置上限并加入抖动；验证、身份验证或支付错误绝不重试。记录 x-request-id，但不要记录 bearer 密钥或签名资源 URL。

提交前估算预算

托管服务定价已于 2026-07-10 核实，单位为每秒美元：

服务	1080p	1440p	4K	计费依据
2.3 Fast T2V/I2V	.06	.12	.24	输出时长
2.3 Pro T2V/I2V	.08	.16	.32	输出时长
2.3 Pro A2V	.10	—	—	输入音频时长
2.3 Pro retake	.10	—	—	完整输入视频时长
2.3 Pro extend	.10	—	—	扩展时长 + 上下文，最多计费 505 帧
2.3 Pro HDR	.20	.40	.80	按输入分辨率档位计费的输入时长

生产使用前，请重新获取定价。报价只是估算，需等输入经过探测且请求被接受后才能确定。预付余额应保持较低，并设置项目级上限。

谨慎处理输入

• 优先使用短期有效的 HTTPS URL 或 /v1/upload。上传流程会返回有效期为一小时的预签名 PUT URL；请转发所有 required_headers。上传文件的 storage_uri 有效期为 24 小时，文件大小上限为 100 MB。
• 公共 HTTPS 输入必须可直接访问，不能经过重定向，也不能使用 IP 地址作为主机。公布的限制为：图像 15 MB/获取超时 10 秒；音频/视频 32 MB/获取超时 30 秒。Data URI 支持 7 MB 的图像或 15 MB 的音频/视频。
• 图像：PNG、JPEG 或 WebP。视频：采用 H.264 或 H.265 编码的 MP4/MOV/MKV。音频：采用受支持编解码器的 WAV/MP3/M4A/OGG；不接受 PCM 和 HE-AAC，AAC 必须为 AAC-LC。
• 上传前使用 ffprobe 在本地探测视频/音频。检查时长、尺寸、帧率、帧数、流、旋转信息、编解码器和音频声道布局。
• 不要声称数据零保留或不会用于训练。当前 API 协议允许在汇总和匿名化后，为分析、产品改进、研究和开发处理使用数据及文本提示词。任务/输出的 24 小时有效期并不代表完整的数据保留政策。对于个人数据、受监管内容、数据驻留要求或机密影像，请获取最新的 DPA 和合同答复，或自行托管。

编写联合音视频提示词

使用一段按时间顺序描述的文字，通常控制在 200 词以内。先描述主体和场景，再按顺序描述动作、镜头运动、光线/色彩、对白、拟音、环境音和最后的画面。避免相互矛盾的剪辑、过多同时发生的动作，以及只描述模糊情绪的措辞。

示例：

安静的午夜餐馆内，中远景，雨水划过窗户。一名疲惫的骑行者从画面左侧进入，摘下黄色头盔，在柜台前坐下。镜头缓缓向前推进，服务员将一杯冒着热气的饮品放在骑行者身旁。骑行者轻声说道：“我赶在暴风雨前到了。”陶瓷杯轻敲木面；雨声和远处冰箱的嗡鸣持续在背景中。霓虹青色和温暖的钨丝灯反光在柜台上荡漾。最后一秒定格在骑行者如释重负的微笑上，不切镜头。

对于 I2V，描述运动和变化，而不是重新描述每个静态像素。对于 A2V，根据可听到的节拍来安排说话者、表演和镜头行为。将准确台词、身份、手部细节和文字渲染视为 QA 风险，而非保证能够实现的效果。

运行托管式客户端，先进行干跑

这个完整的标准库客户端默认会规划当前的 T2V 请求。只有提供 --execute 和审批摘要后才会提交；该摘要会绑定确切请求、有限的费用上限、最新定价证据、尝试记录、输出策略以及确切的交付主机。它绝不会重试 POST，会以原子方式持久化作业 ID，只通过 GET 恢复，不携带 API 凭据下载内容，而是通过固定的公开地址下载，并在验证后以支持崩溃恢复的原子方式发布。

#!/usr/bin/env python3
import argparse, hashlib, http.client, ipaddress, json, os, random, socket, ssl
import subprocess, sys, tempfile, threading, time, uuid
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

API = "https://api.ltx.video"
RATES = {
    ("ltx-2-3-fast", "1920x1080"): Decimal("0.06"),
    ("ltx-2-3-fast", "2560x1440"): Decimal("0.12"),
    ("ltx-2-3-fast", "3840x2160"): Decimal("0.24"),
    ("ltx-2-3-pro", "1920x1080"): Decimal("0.08"),
    ("ltx-2-3-pro", "2560x1440"): Decimal("0.16"),
    ("ltx-2-3-pro", "3840x2160"): Decimal("0.32"),
}
MAX_JSON = 1024 * 1024
MAX_VIDEO = 2_000_000_000
MAX_TOOL = 1024 * 1024

class APIError(RuntimeError):
    def __init__(self, message, status=None, retry_after=None, body_sha256=None, request_id=None):
        super().__init__(message); self.status = status; self.retry_after = retry_after
        self.body_sha256 = body_sha256; self.request_id = request_id

def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()

def fsync_directory(path):
    if os.name == "nt": return
    fd = os.open(path, os.O_RDONLY)
    try: os.fsync(fd)
    finally: os.close(fd)

def atomic_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(value, f, ensure_ascii=False, indent=2, sort_keys=True)
            f.write("\n"); f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path); fsync_directory(path.parent)
    finally:
        if os.path.exists(tmp): os.unlink(tmp)

def claim_once(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        json.dump(value, f, ensure_ascii=False, indent=2, sort_keys=True)
        f.write("\n"); f.flush(); os.fsync(f.fileno())
    fsync_directory(path.parent)

def read_record(path):
    if path.stat().st_size > MAX_JSON: raise RuntimeError("attempt record exceeds cap")
    value = json.loads(path.read_bytes())
    if not isinstance(value, dict): raise RuntimeError("attempt record must be an object")
    return value

def call_json(req, timeout=45):
    try:
        with urlopen(req, timeout=timeout) as r:
            body = r.read(MAX_JSON + 1)
            if len(body) > MAX_JSON: raise RuntimeError("API response exceeds cap")
            try: value = json.loads(body)
            except json.JSONDecodeError as exc:
                raise APIError("API returned invalid JSON", r.status, body_sha256=hashlib.sha256(body).hexdigest(),
                               request_id=r.headers.get("x-request-id")) from exc
            if not isinstance(value, dict): raise RuntimeError("API response is not an object")
            return value, r.headers.get("x-request-id")
    except HTTPError as e:
        body = e.read(MAX_JSON + 1)
        if len(body) > MAX_JSON: body = body[:MAX_JSON]
        try: delay = min(30.0, max(0.0, float(e.headers.get("Retry-After"))))
        except (TypeError, ValueError): delay = None
        raise APIError(f"API HTTP {e.code}", e.code, delay, hashlib.sha256(body).hexdigest(),
                       e.headers.get("x-request-id")) from e

def safe_get(req, deadline, attempts=5):
    last = None
    for attempt in range(attempts):
        try: return call_json(req(), timeout=45)
        except APIError as exc:
            if exc.status not in {429, 500, 503, 504, 529}: raise
            last = exc
            delay = exc.retry_after or min(20, 2 ** attempt) * (1 + random.random() * 0.25)
        except URLError as exc:
            last = exc
            delay = min(20, 2 ** attempt) * (1 + random.random() * 0.25)
        if attempt + 1 == attempts or time.monotonic() + delay >= deadline: raise last
        time.sleep(delay)

def run_bounded(argv, timeout):
    process = subprocess.Popen(argv, shell=False, stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err, total, lock, overflow = bytearray(), bytearray(), [0], threading.Lock(), threading.Event()
    def drain(stream, bucket):
        while True:
            chunk = stream.read(65536)
            if not chunk: break
            with lock:
                room = max(0, MAX_TOOL - total[0]); bucket.extend(chunk[:room]); total[0] += len(chunk)
                if total[0] > MAX_TOOL: overflow.set()
            if overflow.is_set():
                try: process.kill()
                except OSError: pass
    threads = [threading.Thread(target=drain, args=(process.stdout, out), daemon=True),
               threading.Thread(target=drain, args=(process.stderr, err), daemon=True)]
    for thread in threads: thread.start()
    timed_out = False
    try: code = process.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True; process.kill(); code = process.wait()
    for thread in threads: thread.join()
    return code, bytes(out), bytes(err), timed_out, overflow.is_set()

class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host, address):
        super().__init__(host, 443, timeout=120, context=ssl.create_default_context()); self.address = address
    def connect(self):
        sock = socket.create_connection((self.address, 443), self.timeout)
        try: self.sock = self._context.wrap_socket(sock, server_hostname=self.host)
        except BaseException:
            sock.close(); raise

def download_video(url, stage_path, allowed_hosts):
    parsed = urlsplit(url); host = (parsed.hostname or "").lower()
    if (parsed.scheme != "https" or parsed.username or parsed.password or parsed.fragment
            or parsed.port not in (None, 443) or not parsed.path or host not in allowed_hosts):
        raise RuntimeError("result URL is not an exact approved HTTPS host")
    try: ipaddress.ip_address(host); raise RuntimeError("IP-literal result host rejected")
    except ValueError: pass
    addresses = sorted({item[4][0] for item in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM)})
    if not addresses or any(not ipaddress.ip_address(address).is_global for address in addresses):
        raise RuntimeError("result host did not resolve only to public addresses")
    connection = PinnedHTTPSConnection(host, addresses[0]); created = False
    try:
        target = parsed.path + (("?" + parsed.query) if parsed.query else "")
        connection.request("GET", target, headers={"Accept": "video/mp4,application/octet-stream"})
        response = connection.getresponse()
        if 300 <= response.status < 400: raise RuntimeError("result redirect rejected")
        if response.status != 200: raise RuntimeError(f"result HTTP {response.status}")
        mime = response.getheader("Content-Type", "").split(";", 1)[0].lower()
        if mime not in {"video/mp4", "application/octet-stream"}: raise RuntimeError("result MIME rejected")
        length = response.getheader("Content-Length")
        if length and (not length.isdigit() or int(length) > MAX_VIDEO): raise RuntimeError("declared result size rejected")
        fd = os.open(stage_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600); created = True; total = 0
        with os.fdopen(fd, "wb") as f:
            while True:
                chunk = response.read(1024 * 1024)
                if not chunk: break
                total += len(chunk)
                if total > MAX_VIDEO: raise RuntimeError("result crossed byte cap")
                f.write(chunk)
            f.flush(); os.fsync(f.fileno())
        return stage_path
    except BaseException:
        if created and stage_path.exists(): stage_path.unlink()
        raise
    finally: connection.close()

def inspect_mp4(path, output, resolution, duration, fps, silent):
    with path.open("rb") as f: head = f.read(12)
    if len(head) < 12 or head[4:8] != b"ftyp": raise RuntimeError("MP4 signature rejected")
    args = ["ffprobe", "-v", "error", "-show_entries",
            "format=duration,size:stream=codec_type,codec_name,width,height,r_frame_rate", "-of", "json", str(path)]
    code, out, _, timed_out, overflow = run_bounded(args, 30)
    if code or timed_out or overflow: raise RuntimeError("bounded ffprobe failed")
    probe = json.loads(out); streams = probe.get("streams", []); videos = [s for s in streams if s.get("codec_type") == "video"]
    if len(videos) != 1: raise RuntimeError("expected one video stream")
    width, height = videos[0].get("width"), videos[0].get("height")
    expected = tuple(int(item) for item in resolution.split("x"))
    if (width, height) != expected: raise RuntimeError("dimensions do not match approval")
    audio_count = sum(s.get("codec_type") == "audio" for s in streams)
    if (not silent and audio_count < 1) or (silent and audio_count): raise RuntimeError("audio streams do not match approval")
    try: actual = Decimal(str(probe.get("format", {}).get("duration", "")))
    except InvalidOperation as exc: raise RuntimeError("invalid media duration") from exc
    if not actual.is_finite() or abs(actual - Decimal(duration)) > Decimal("0.75"):
        raise RuntimeError("media duration outside tolerance")
    rate = str(videos[0].get("r_frame_rate", "0/1")).split("/", 1)
    try: actual_fps = Decimal(rate[0]) / Decimal(rate[1])
    except (InvalidOperation, ZeroDivisionError, IndexError) as exc: raise RuntimeError("invalid frame rate") from exc
    if not actual_fps.is_finite() or abs(actual_fps - Decimal(fps)) > Decimal("0.01"):
        raise RuntimeError("frame rate does not match approval")
    code, _, _, timed_out, overflow = run_bounded(["ffmpeg", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
    if code or timed_out or overflow: raise RuntimeError("bounded full decode failed")
    digest = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""): digest.update(chunk)
    return {"path": str(output), "bytes": path.stat().st_size, "sha256": digest.hexdigest(),
            "media": {"duration": str(actual), "width": width, "height": height, "fps": str(actual_fps),
                      "videoCodec": videos[0].get("codec_name"), "audioStreams": audio_count}}

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--prompt", required=True)
    p.add_argument("--model", choices=["ltx-2-3-fast", "ltx-2-3-pro"], default="ltx-2-3-fast")
    p.add_argument("--duration", type=int, choices=[6,8,10,12,14,16,18,20], default=6)
    p.add_argument("--resolution", choices=["1920x1080","1080x1920","2560x1440","1440x2560","3840x2160","2160x3840"], default="1920x1080")
    p.add_argument("--fps", type=int, choices=[24,25,48,50], default=24)
    p.add_argument("--silent", action="store_true")
    p.add_argument("--attempt-id", required=True); p.add_argument("--attempt-record", type=Path, default=Path("ltx-attempt.json"))
    p.add_argument("--max-usd", required=True); p.add_argument("--approval-sha256")
    p.add_argument("--pricing-checked-at", required=True); p.add_argument("--pricing-evidence-sha256", required=True)
    p.add_argument("--delivery-host", action="append", required=True, help="exact approved result host; repeat as needed")
    p.add_argument("--execute", action="store_true"); p.add_argument("--resume", action="store_true")
    p.add_argument("--max-wait", type=int, default=1800)
    p.add_argument("--output", type=Path, default=Path("ltx-output.mp4"))
    p.add_argument("--show-full-plan", action="store_true")
    a = p.parse_args()
    if not 1 <= len(a.prompt) <= 5000 or len(a.prompt.encode()) > 20000:
        p.error("prompt must be 1..5000 characters and at most 20000 UTF-8 bytes")
    if not 60 <= a.max_wait <= 7200: p.error("max wait must be 60..7200 seconds")
    if a.model.endswith("pro") and a.duration not in (6,8,10): p.error("Pro supports only 6/8/10 s")
    if (a.duration > 10) and not (a.model.endswith("fast") and a.resolution in ("1920x1080","1080x1920") and a.fps in (24,25)):
        p.error("12–20 s requires Fast, 1080p, and 24/25 fps")
    if a.duration > 10 or a.resolution not in ("1920x1080","1080x1920") or a.fps not in (24,25):
        if a.duration not in (6,8,10): p.error("this model/resolution/fps combination supports only 6/8/10 s")
    landscape = a.resolution if int(a.resolution.split("x")[0]) > int(a.resolution.split("x")[1]) else "x".join(reversed(a.resolution.split("x")))
    estimate = RATES[(a.model, landscape)] * Decimal(a.duration)
    try: attempt = uuid.UUID(a.attempt_id); ceiling = Decimal(a.max_usd)
    except (ValueError, InvalidOperation) as exc: p.error(f"attempt ID or maximum invalid: {exc}")
    if attempt.version != 4 or str(attempt) != a.attempt_id.lower(): p.error("attempt ID must be canonical UUIDv4")
    if not ceiling.is_finite() or ceiling <= 0 or ceiling < estimate:
        p.error("maximum must be finite, positive, and cover the estimate")
    if (len(a.pricing_evidence_sha256) != 64
            or any(c not in "0123456789abcdef" for c in a.pricing_evidence_sha256)):
        p.error("pricing evidence must be a lowercase SHA-256")
    try: checked_at = datetime.fromisoformat(a.pricing_checked_at.replace("Z", "+00:00"))
    except ValueError as exc: p.error(f"pricing timestamp must be ISO-8601: {exc}")
    if checked_at.tzinfo is None: p.error("pricing timestamp must include a timezone")
    allowed_hosts = sorted(set(host.lower().rstrip(".") for host in a.delivery_host))
    if (not allowed_hosts or len(allowed_hosts) > 8
            or any(not host or len(host) > 253
                   or any(c not in "abcdefghijklmnopqrstuvwxyz0123456789-." for c in host) for host in allowed_hosts)):
        p.error("delivery hosts must be 1..8 exact DNS names")
    for host in allowed_hosts:
        try: ipaddress.ip_address(host); p.error("delivery hosts cannot be IP literals")
        except ValueError: pass
    output, record_path = a.output.resolve(), a.attempt_record.resolve()
    stage_path = output.with_name("." + output.name + ".ltx-stage")
    if output.suffix.lower() != ".mp4" or record_path in {output, stage_path}: p.error("output/record path rejected")
    payload = {"prompt":a.prompt,"model":a.model,"duration":a.duration,"resolution":a.resolution,"fps":a.fps,"generate_audio":not a.silent}
    request_hash = hashlib.sha256(canonical(payload)).hexdigest()
    endpoint = f"{API}/v2/text-to-video"
    envelope = {"backend":"ltx-hosted-async-v2", "endpoint":endpoint, "payload":payload,
        "attemptId":str(attempt), "attemptRecord":str(record_path),
        "outputPolicy":{"path":str(output), "stagePath":str(stage_path), "exactDeliveryHosts":allowed_hosts,
                        "maxBytes":MAX_VIDEO, "mime":["video/mp4","application/octet-stream"],
                        "noOverwrite":True, "fullDecode":True},
        "cost":{"pricingCheckedAt":checked_at.isoformat(), "pricingEvidenceSha256":a.pricing_evidence_sha256,
                "rateUsdPerSecond":str(RATES[(a.model, landscape)]), "seconds":a.duration,
                "creates":1, "estimatedMaxUsd":str(estimate), "approvedMaxUsd":str(ceiling)}}
    approval = hashlib.sha256(canonical(envelope)).hexdigest()
    safe_payload = {**payload, "prompt":"sha256:" + hashlib.sha256(a.prompt.encode()).hexdigest()}
    print(json.dumps({"dryRun":not a.execute, "approvalSha256":approval,
        "plan":{**envelope,"payload":safe_payload}}, indent=2, ensure_ascii=False))
    if a.show_full_plan: print(json.dumps({"protectedFullPlan":envelope}, indent=2, ensure_ascii=False))
    if not a.execute: return 0
    if a.approval_sha256 != approval: raise SystemExit("exact approval digest mismatch")
    age = datetime.now(timezone.utc) - checked_at.astimezone(timezone.utc)
    if age.total_seconds() < -300 or age.total_seconds() > 86400:
        raise SystemExit("pricing evidence must be rechecked within 24 hours of execution")
    if a.resume:
        if not record_path.exists(): raise SystemExit("resume record missing")
        record = read_record(record_path)
        if record.get("approval_sha256") != approval or record.get("request_sha256") != request_hash:
            raise SystemExit("resume does not match exact approved request")
        job_id = record.get("job_id")
        if not isinstance(job_id, str) or not job_id: raise SystemExit("no known job ID; never replay create")
        expected_artifact = record.get("artifact_staged") or record.get("artifact")
        candidate = output if output.exists() else stage_path if stage_path.exists() else None
        if candidate is not None:
            if not isinstance(expected_artifact, dict):
                if record.get("status") == "downloading" and candidate == stage_path:
                    stage_path.unlink(); fsync_directory(stage_path.parent)
                else: raise SystemExit("unclaimed output/stage exists; refusing overwrite or adoption")
            else:
                artifact = inspect_mp4(candidate, output, a.resolution, a.duration, a.fps, a.silent)
                if (artifact.get("sha256") != expected_artifact.get("sha256")
                        or artifact.get("bytes") != expected_artifact.get("bytes")):
                    raise SystemExit("recovered artifact does not match durable staged evidence")
                if candidate == stage_path:
                    if output.exists(): raise SystemExit("output appeared during recovery")
                    os.replace(stage_path, output); fsync_directory(output.parent)
                record.update(status="artifact_saved", artifact=artifact, updated_unix=int(time.time()))
                record.pop("artifact_staged", None); atomic_json(record_path, record)
                print(json.dumps(record, indent=2)); return 0
        key = os.environ.get("LTXV_API_KEY")
        if not key: raise SystemExit("LTXV_API_KEY is required to resume polling")
    else:
        if output.exists() or stage_path.exists(): raise SystemExit("output or stage already exists")
        key = os.environ.get("LTXV_API_KEY")
        if not key: raise SystemExit("LTXV_API_KEY is required")
        output.parent.mkdir(parents=True, exist_ok=True)
        record = {"attempt_id":str(attempt), "approval_sha256":approval, "request_sha256":request_hash,
                  "status":"posting", "job_id":None, "created_unix":int(time.time())}
        try: claim_once(record_path, record)
        except FileExistsError as exc: raise SystemExit("attempt already claimed; use resume") from exc
    if not a.resume:
        req = Request(endpoint, data=canonical(payload), method="POST",
                      headers={"Authorization":f"Bearer {key}","Content-Type":"application/json"})
        try: job, request_id = call_json(req, timeout=120)
        except Exception as exc:
            known_rejection = (isinstance(exc, APIError) and exc.status in {400,401,402,404,422})
            record.update(status="create_rejected" if known_rejection else "create_outcome_unknown",
                          error_type=type(exc).__name__, http_status=getattr(exc,"status",None),
                          error_body_sha256=getattr(exc,"body_sha256",None), request_id=getattr(exc,"request_id",None))
            atomic_json(record_path, record)
            if known_rejection: raise RuntimeError("provider rejected create before acceptance") from exc
            raise RuntimeError("create outcome unknown; reconcile console/support, never replay") from exc
        job_id = job.get("id")
        if (not isinstance(job_id, str) or not 1 <= len(job_id) <= 200
                or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_." for c in job_id)):
            record.update(status="create_outcome_unknown", error_type="MissingOrInvalidJobId",
                          create_response_sha256=hashlib.sha256(canonical(job)).hexdigest())
            atomic_json(record_path, record); raise RuntimeError("create returned no safe job ID; reconcile")
        record.update(status="job_id_saved", job_id=job_id, created_at=job.get("created_at"), request_id=request_id,
                      create_response_sha256=hashlib.sha256(canonical(job)).hexdigest(), updated_unix=int(time.time()))
        atomic_json(record_path, record)
    deadline = time.monotonic() + a.max_wait
    state = None
    while time.monotonic() < deadline:
        state, poll_request_id = safe_get(lambda: Request(f"{API}/v2/text-to-video/{job_id}",
            headers={"Authorization":f"Bearer {key}"}), deadline)
        status = state.get("status")
        if status == "failed":
            error = state.get("error") or {}; message = str(error.get("message", "")) if isinstance(error, dict) else str(error)
            record.update(status="generation_failed", error_type=error.get("type") if isinstance(error,dict) else None,
                          error_message_sha256=hashlib.sha256(message.encode()).hexdigest(), request_id=poll_request_id)
            atomic_json(record_path, record); raise RuntimeError("generation failed")
        if status == "completed": break
        if status not in ("pending","processing"): raise RuntimeError("unknown job status")
        if time.monotonic() + 5 >= deadline: break
        time.sleep(5 + random.random())
    if state is None or state.get("status") in {"pending","processing"}:
        record.update(status="polling_paused", updated_unix=int(time.time())); atomic_json(record_path, record)
        raise TimeoutError(f"poll timeout; resume job {job_id}, do not resubmit")
    video_url = state.get("result",{}).get("video_url")
    if not isinstance(video_url, str) or not video_url: raise RuntimeError("completed job omitted result URL")
    record.update(status="downloading", result_response_sha256=hashlib.sha256(canonical(state)).hexdigest(),
                  result_url_sha256=hashlib.sha256(video_url.encode()).hexdigest(), request_id=poll_request_id,
                  actual_cost_estimate_usd=str(estimate), billing_reconciliation="console_required",
                  stage_path=str(stage_path), updated_unix=int(time.time()))
    atomic_json(record_path, record)
    try:
        output.parent.mkdir(parents=True, exist_ok=True)
        download_video(video_url, stage_path, set(allowed_hosts))
        artifact = inspect_mp4(stage_path, output, a.resolution, a.duration, a.fps, a.silent)
    except BaseException as exc:
        record.update(status="artifact_validation_failed", error_type=type(exc).__name__)
        atomic_json(record_path, record); raise
    record.update(status="artifact_staged", artifact_staged=artifact, updated_unix=int(time.time()))
    atomic_json(record_path, record)
    if output.exists(): raise RuntimeError("output appeared before publication")
    os.replace(stage_path, output); fsync_directory(output.parent)
    record.update(status="artifact_saved", artifact=artifact, rights_review="required", moderation_review="required",
                  creative_qa="pending", ai_disclosure="required", provenance_credentials="not_documented",
                  updated_unix=int(time.time()))
    record.pop("artifact_staged", None); atomic_json(record_path, record)
    print(json.dumps(record, indent=2)); return 0

if __name__ == "__main__": main()

先规划，并使用账户交付路径文档中明确列出的当前结果主机：

python ltx_hosted.py --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.36 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$LTX_RESULT_HOST" --prompt "A single six-second chronological audiovisual shot of a red kite rising over a salt flat at dawn."

使用完全相同的请求执行一次：

LTXV_API_KEY=... python ltx_hosted.py --execute --approval-sha256 "<exact digest>" --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.36 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$LTX_RESULT_HOST" --prompt "A single six-second chronological audiovisual shot of a red kite rising over a salt flat at dawn."

BILLING_SHA256 是实际审阅过的带日期一手定价证据的摘要。LTX_RESULT_HOST 是当前 API 交付约定授权的确切主机名，绝不能根据返回的 URL 推断出通配符。定价证据超过 24 小时后，执行将被拒绝。只有在持久化的尝试记录包含作业 ID 时，才使用未更改的参数加上 --execute --resume；posting 或 create_outcome_unknown 状态必须在开发者控制台中核查或联系支持人员处理，绝不能重放。

对于其他操作，只调整端点和载荷；保留审批、持久化、禁止重新提交、轮询、下载和解码防护。对于 HDR，将 exr_frames_url 下载为 ZIP，解压时拒绝路径遍历，验证帧数/尺寸/通道，并在交付前进行显式色彩管理。

单独运行本地／开放权重

除非用户明确批准使用第三方运行时，否则仅使用官方 Lightricks/LTX-2 仓库和 Lightricks/LTX-2.3 模型仓库。固定经过审核的 Git 提交以及 Hugging Face 修订版／提交。可复现任务中不要使用会移动的 main，也不要使用未固定版本的 huggingface-cli download。

当前官方检查点用途（已于 2026-07-10 验证）：

• ltx-2.3-22b-dev.safetensors（约 46.1 GB）：BF16 开发检查点；灵活且可训练；用于 LoRA／微调和注重质量的工作流。
• ltx-2.3-22b-distilled-1.1.safetensors（约 46.1 GB）：步数蒸馏推理检查点；当前 1.1 美学／音频更新版；官方蒸馏工作流使用第一阶段 8 步、第二阶段 4 步去噪。
• ltx-2.3-22b-distilled-1.0.safetensors：较早版本；除非有兼容性原因，否则新任务不要选用。
• Distilled LoRA 1.1（约 7.61 GB）：当所选工作流需要蒸馏适配器时，将其应用于开发检查点；不能用它替代基础检查点。
• ltx-2.3-spatial-upscaler-x2-1.1.safetensors（约 996 MB）或 x1.5（约 1.09 GB）：当前两阶段实现所需。时间 x2 升频器被描述为受支持／未来工作流素材，而非当前通用必需项。
• 来自 google/gemma-3-12b-it-qat-q4_0-unquantized 的 Gemma 3 资源：必需的文本编码器目录；请接受 Google 的单独访问条款，并固定其修订版。

当前官方开源系统要求为：NVIDIA GPU，显存 32 GB 以上，内存 32 GB，空闲存储空间 100 GB，CUDA 11.8 以上，以及 Python 3.10 以上；而仓库／模型卡可能会列出更新的已测试版本。请将文档要求视为最低门槛，而非保证。官方训练器建议标准训练使用 80 GB 以上显存，并为 32 GB 显卡提供低显存 INT8 配置。不要把推理最低要求当作训练承诺。

请使用干净的 Linux／CUDA 环境运行官方训练器。记录 GPU 型号／显存、驱动、CUDA、PyTorch、Python、仓库提交、软件包锁定信息、注意力后端、量化方式和卸载模式。下载前，请确保有约 100 GB 可用空间，以容纳一个完整检查点、Gemma 资源、升频器、环境和输出；更大的缓存或多个变体需要更多空间。

选择官方工作流

工作流	用途
TI2VidTwoStagesPipeline	默认的生产级 T2V／I2V 加空间升频
TI2VidTwoStagesHQPipeline	注重质量的二阶采样
TI2VidOneStagePipeline	快速生成较低分辨率原型
DistilledPipeline	使用蒸馏检查点的官方最低步数路径
ICLoraPipeline	使用精确 IC-LoRA 进行兼容的图像／视频转换
KeyframeInterpolation	插值受控关键帧
A2VidPipelineTwoStage	本地音频驱动视频
RetakePipeline	本地视听重拍
LipDubPipeline	使用指定 IC-LoRA 的蒸馏口型同步路径
HDRICLoraPipeline	仅视频线性浮点帧，用于 EXR／色调映射；色彩工作流由调用方负责

确认适配器与基础模型／版本兼容。仓库中既有 2.3 22B 适配器，也有 LTX-2-19b-* 等较早的文件名；不要仅因功能名称相同，就将 19B LoRA 加载到 22B 2.3 上。

审核、批准，然后运行

先执行仅包含元数据的计划：

git ls-remote https://github.com/Lightricks/LTX-2.git HEAD
curl -fsS https://huggingface.co/api/models/Lightricks/LTX-2.3 > ltx-2.3-metadata.json
# Review commits, filenames, sizes, license, Gemma access, disk, and GPU before approval.

获得明确批准后，替换为经过审核的不可变修订版本：

git clone https://github.com/Lightricks/LTX-2.git
cd LTX-2 && git checkout <reviewed-git-commit> && uv sync --frozen
huggingface-cli download Lightricks/LTX-2.3 \
  ltx-2.3-22b-distilled-1.1.safetensors \
  ltx-2.3-spatial-upscaler-x2-1.1.safetensors \
  --revision <reviewed-hf-commit> --local-dir ./models/ltx-2.3
huggingface-cli download google/gemma-3-12b-it-qat-q4_0-unquantized \
  --revision <reviewed-gemma-commit> --local-dir ./models/gemma-3
sha256sum models/ltx-2.3/* > models/SHA256SUMS
uv run python -m ltx_pipelines.distilled \
  --distilled-checkpoint-path models/ltx-2.3/ltx-2.3-22b-distilled-1.1.safetensors \
  --spatial-upsampler-path models/ltx-2.3/ltx-2.3-spatial-upscaler-x2-1.1.safetensors \
  --gemma-root models/gemma-3 --seed 42 --output-path output.mp4 \
  --prompt 'A single chronological audiovisual paragraph here.'
ffprobe -v error -show_streams -show_format output.mp4
ffmpeg -v error -i output.mp4 -f null -

仅当所选 CLI 同时支持 --quantization fp8-cast --offload cpu，且通过小规模验收测试后，才添加这些参数。fp8-cast 会在加载期间将 BF16 transformer 线性权重降精度。fp8-scaled-mm 适用于使用 TensorRT-LLM 的 FP8 检查点，并且推荐用于 Hopper；它不能与 BF16 fp8-cast 互换。仓库仅针对经过验证的确切 B200/PyTorch 组合说明了 FlashAttention 4；其他 CUDA GPU（包括 Hopper）请使用 xFormers。Offload 可以减少 VRAM 占用，但可能增加 RAM 占用、磁盘 I/O 和延迟；某些流水线在启用 offload 时会明确禁用量化。请遵循该流水线的帮助文档/README，不要一概而论。

遵循潜空间几何要求：宽度和高度必须能被 32 整除，官方本地流水线的帧数必须为 8n+1。由于标志会不断变化，请在固定的修订版本上使用 --help 验证确切的 CLI。

许可证与政策关卡

不要称模型权重采用 Apache 许可证。官方代码/仓库可能包含采用宽松许可证的组件，LTX Desktop 采用 Apache-2.0 许可证，但 LTX-2 模型材料使用日期为 2026-01-05 的 LTX-2 Community License Agreement。

在本地使用或分发前，如涉及重大事项，请咨询法律顾问并审阅当前法律文本：

• 年度合并收入低于 10M USD 的实体可获得社区授权；法律文本要求年收入达到或超过 10M USD 的实体获取付费商业使用许可证。请合并计算关联公司及受共同控制的实体。
• “衍生模型”的定义很宽泛，包括微调/改编后的权重、架构修改、蒸馏，以及某些使用输出进行训练以实现类似功能的模型。分发时须履行许可证、声明和下游使用方面的义务。
• 许可证限制直接竞争的产品/服务、某些使用输出训练模型的行为、被禁止的用途、违反制裁/出口规定的行为，以及未经同意的冒充行为。
• 机器生成的内容必须附有清晰易懂的声明。对于可识别的人物，应取得其同意，且绝不可捏造具有欺骗性的身份声明。
• 许可方通常声明对输出不主张权利，但仍受相关限制；这不构成对版权、授权或不侵权的保证。

托管 API 的使用受当前 API 协议和可接受使用政策单独约束。不要假设社区模型许可证适用于托管服务。

验证每次交付

1. 端到端验证容器解码、时长容差、尺寸、帧率、视频/音频编解码器、音轨是否可听或是否有意静音、音画同步，以及是否存在损坏帧。
2. 检查首帧、中间帧、末帧和镜头边界；完整试听音频。检查身份、人体结构、唇形、语音、文本、时间连续性、闪烁、削波和环境音突变。
3. 对于有条件生成的作品，对比首帧/末帧和源身份。对于重拍，检查两个编辑接缝并确认未修改区域保持不变。对于延长，检查衔接处和音频连续性。
4. 对于 HDR，验证 EXR 序列完整性、线性光解释、元数据、色域、母版制作/色调映射变换，以及 SDR 观看变换。绝不要将 H.264 预览重新标记为 HDR 母版。
5. 扫描政策和声明要求，确认人物/品牌/音乐/素材的使用权，并保留不含机密信息或会过期的签名 URL 的清单。
6. 技术或政策关卡未通过时，应拒绝或重新生成；不要仅靠转码掩盖失败。

需要重新核查的来源

• LTX 官方 API 文档
• 支持的模型
• 异步作业
• 定价
• 输入格式
• API 更新日志
• LTX-2 官方代码仓库
• LTX-2.3 官方模型仓库
• 官方开源系统要求
• LTX-2 技术报告
• LTX-2 社区许可证
• LTX API 许可协议
• LTX 可接受使用政策

模型菜单、定价、限制、Beta 状态、弃用日期、法律条款、检查点名称和大小以及硬件指南都可能随时变化。请在执行时重新核查，并记录核查日期。