ltx-2-video

LTX-2 Video

ワークフローで動画と音声の同時生成、最初／最後のフレームによる条件付け、音声駆動型動画、部分的な再生成、延長、または公式のセルフホスト可能な方法が必要な場合は、LTX-2.3 を使用します。ホスト型 API の仕様、ローカルの重みの仕様、研究上の主張は、それぞれ分けて扱ってください。

まず計画を立てる

1. 要求されたモード、尺、アスペクト比、解像度、フレームレート、音声の要件、納品コーデック、入力に人物や機微なデータが含まれるかどうかを明記します。
2. 以下のルーティング表に従って、hosted または local を選択します。「オープン」が無料、無制限、または運用が容易という意味だと推測してはいけません。
3. 有料リクエスト、アセットのアップロード、またはモデルのダウンロードを行う前に、ドライランの計画を作成します。正確なエンドポイント／モデルまたはチェックポイントのファイル名、ホスト型の料金見積もり、出力先、検証手順を含めます。
4. 最初の課金対象となる API 送信、または数 GB のダウンロードの前に、明示的な承認を得ます。ユーザーに代わって自動チャージを有効にしてはいけません。
5. 来歴情報を保持します。プロンプト、利用可能な場合はシード、モデル／チェックポイント、API リクエスト／ジョブ ID、ソースアセットのハッシュ、生成パラメーター、ライセンス確認日、出力ハッシュを記録します。

適切にルーティングする

必要なもの	推奨ルート	仕様
4K オプション付きのマネージド T2V/I2V	ホスト型 async V2	試行や長めの 1080p クリップには ltx-2-3-fast、忠実度を重視する場合は ltx-2-3-pro
既存の 2〜20 秒の音声から映像を生成	ホスト型 A2V	Pro のみ。プロンプトまたは画像が必須。入力音声の秒数に基づいて課金
動画の一部を置き換える	ホスト型 retake	Pro のみ。編集範囲だけでなく、入力動画の全尺に基づいて課金
冒頭／末尾に素材を追加する	ホスト型 extend	Pro のみ。延長分とコンテキストが課金対象で、合計 505 フレームが上限
HDR グレーディング用に SDR 動画を変換する	ホスト型 HDR	async のみ。完成した MP4 ではなく、線形 EXR フレームの ZIP を返す
データ管理、再現性、カスタム LoRA、オフライン作業	公式ローカルリポジトリ	GPU、依存関係、チェックポイント、Gemma の利用規約、出力パイプライン、ライセンス遵守を自分で管理
ノードベースのローカルワークフロー	公式 ComfyUI-LTXVideo	ワークフローとカスタムノードのリビジョンを固定し、別個の実行環境として扱う

本番環境では async V2 を優先してください。同期型の V1 エンドポイントは、1 回の長いリクエストで MP4 を返すため、接続を維持できる短い実験にのみ適しています。2026-07-10 時点で、async 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文字。モデル、再生時間、解像度が必須です
画像から動画を生成	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 です

すべての非同期ジョブで、返された id を保存してから、5秒ごとに GET /v2/{operation}/{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リトライには上限を設け、検証、認証、決済エラーは決してリトライしないでください。Bearerキーや署名付きアセットURLをログに記録せず、x-request-id を記録してください。

送信前に予算を確認する

ホスト型サービスの料金を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リテイク	.10	—	—	入力動画の全体の長さ
2.3 Pro延長	.10	—	—	延長分＋コンテキスト。最大505フレーム
2.3 Pro HDR	.20	.40	.80	入力の解像度区分における長さ

本番環境で使用する直前に、料金を再取得してください。入力を検査してリクエストが受理されるまでは、見積もりはあくまで概算です。プリペイド残高は少額に保ち、プロジェクト単位の上限を設定してください。

入力を慎重に扱う

• 有効期間の短いHTTPS URLまたは /v1/upload を優先してください。アップロードフローは有効期間が1時間の事前署名付きPUT URLを返します。required_headers はすべて転送してください。アップロードした storage_uri の有効期間は24時間で、ファイルサイズは100 MBまでです。
• 公開HTTPS入力は、リダイレクトやIPアドレスのホストを使わず、直接アクセスできる必要があります。公開されている上限は、画像が15 MB／取得時間10秒、音声・動画が32 MB／取得時間30秒です。データ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語未満の、時系列に沿った1つの文章をそのまま記述します。被写体と設定を説明してから、動作の順序、カメラの動き、照明と色、台詞、フォーリー、環境音、最後の瞬間を説明します。矛盾するカット、同時に起こる動作の詰め込みすぎ、曖昧な雰囲気だけの表現は避けてください。

例：

真夜中の静かなダイナーを捉えたミディアムワイドショット。雨が窓を伝っています。疲れた様子の自転車乗りが画面左から入り、黄色いヘルメットを脱いでカウンター席に座ります。サーバーが湯気の立つマグカップをその人の隣に置くと、カメラはゆっくりとドリーインします。自転車乗りは静かに「嵐が来る前に着けた」と言います。陶器が木に触れる音がし、雨音と遠くの冷蔵庫のうなりがその下で続きます。ネオンのシアンと暖かなタングステンの反射がカウンターに揺らめきます。最後の1秒は、ほっとした笑顔を映したままカットせずに保ちます。

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."

同一のリクエストを1回だけ実行します。

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美的品質／音声アップデートです。公式の蒸留済みパイプラインでは、第1段階で8ステップ、第2段階で4ステップのデノイズを使用します。
• ltx-2.3-22b-distilled-1.0.safetensors：旧リリースです。互換性上の理由がない限り、新しい作業では選択しないでください。
• 蒸留済み LoRA 1.1（約7.61 GB）：選択したパイプラインが蒸留済みアダプターを想定している場合に、dev チェックポイントへ適用します。ベースチェックポイントの代わりにはなりません。
• ltx-2.3-spatial-upscaler-x2-1.1.safetensors（約996 MB）または x1.5（約1.09 GB）：現在の2段階実装で必要です。時間方向の x2 アップスケーラーは、現行パイプライン全般に必須ではなく、対応予定／将来のパイプライン向けの素材として説明されています。
• google/gemma-3-12b-it-qat-q4_0-unquantized の Gemma 3 アセット：テキストエンコーダーディレクトリとして必要です。Google の個別のアクセス条件に同意し、そのリビジョンを固定してください。

現在の公式オープンソースのシステム要件には、NVIDIA GPU（VRAM 32 GB 以上）、RAM 32 GB、空きストレージ100 GB、CUDA 11.8 以上、Python 3.10 以上と記載されていますが、リポジトリ／モデルカードには、より新しいテスト済みバージョンが記載されている場合があります。ドキュメントの要件は最低基準であり、動作の保証ではないと考えてください。公式トレーナーは、標準的な学習には VRAM 80 GB 以上、32 GB のカードには低 VRAM の INT8 構成を推奨しています。推論の最低要件を、学習の保証として扱わないでください。

公式トレーナーには、クリーンな Linux／CUDA 環境を使用してください。GPU のモデル／VRAM、ドライバー、CUDA、PyTorch、Python、リポジトリのコミット、パッケージロック、アテンションバックエンド、量子化、オフロードモードを記録してください。ダウンロード前に、チェックポイント1つ分、Gemma アセット、アップスケーラー、環境、出力を保存できるよう、約100 GB の空き容量を確保してください。キャッシュを大きくする場合や複数のバリアントを使う場合は、さらに多くの容量が必要です。

公式パイプラインを選択する

パイプライン	用途
TI2VidTwoStagesPipeline	空間アップサンプリングを含む、標準の本番向け T2V／I2V
TI2VidTwoStagesHQPipeline	品質重視の2次サンプリング
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 のトランスフォーマーの線形重みを低精度化します。fp8-scaled-mm は TensorRT-LLM を使用する FP8 チェックポイント向けであり、Hopper で推奨されます。BF16 の fp8-cast と互換性はありません。リポジトリでは、FlashAttention 4 の対応対象は検証済みの B200／PyTorch の組み合わせに限られています。Hopper を含む、それ以外の CUDA GPU では xFormers を使用してください。オフロードは VRAM を削減できますが、RAM、ディスク I/O、レイテンシが増加する可能性があります。また、一部のパイプラインでは、オフロード中に量子化が明示的に無効化されます。全般的な前提に頼らず、そのパイプラインのヘルプ／README に従ってください。

潜在表現のジオメトリを守ってください。公式のローカルパイプラインでは、幅と高さは 32 で割り切れる値、フレーム数は 8n+1 である必要があります。フラグは変わるため、固定したリビジョンで --help を使い、実際の CLI を確認してください。

ライセンスとポリシーの確認

モデルの重みが Apache ライセンスであるとは記載しないでください。公式のコード／リポジトリには寛容なライセンスのコンポーネントが含まれている場合があり、LTX Desktop は Apache-2.0 ですが、LTX-2 のモデル関連資料には、2026-01-05 付の LTX-2 Community License Agreement が適用されます。

ローカルでの使用や配布の前に、重要な場合は弁護士に相談して、最新の法的文書を確認してください。

• 合算した年間収益が USD 10M 未満の事業体にはコミュニティ助成が適用されます。法的文書では、USD 10M 以上の事業体に有料の商用利用ライセンスの取得を義務付けています。関連会社および共通の支配下にある事業体も合算してください。
• 「Derivative Models」は広義で、ファインチューニング／適応済みの重み、アーキテクチャの変更、蒸留、ならびに出力を使って類似の動作をするように訓練された特定のモデルを含みます。配布には、ライセンス、通知、下流での利用に関する義務が伴います。
• ライセンスは、直接競合する製品／サービス、出力をモデル訓練に使う特定の行為、禁止用途、制裁／輸出規制への違反、同意のないなりすましを制限しています。
• 機械生成コンテンツには、理解できる形で生成物であることを明示する必要があります。本人と認識できる人物のコンテンツを作成する際は同意を得て、誤解を招く身元の主張は決して作成しないでください。
• ライセンサーは通常、制限の適用を条件として、出力に対する権利を主張しません。ただし、これは著作権、権利処理、非侵害を保証するものではありません。

ホスト型 API の利用には、最新の API 契約および Acceptable Use Policy が別途適用されます。コミュニティモデルライセンスがホスト型の利用を規定すると想定しないでください。

納品物をすべて検証する

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 利用規定

モデルメニュー、料金、制限、ベータ版のステータス、廃止日、法的条件、チェックポイントの名前やサイズ、ハードウェアに関する推奨事項は変更される可能性があります。実行時に再確認し、確認日を記録してください。