kling-kolors-image

Kling と Kolors で静止画を生成する

まず、名称を整理します。「Kling image」「Kolors」「可图」は現在、関連はありますが、相互に置き換えられない別々のサービスを指します。

サービス	内容	適切な用途
Kling AI IMAGE 3.0	複数の参照画像に対応した、現在提供中のホスト型静止画生成および自然言語による編集機能です	1枚の静止画、製品やキャラクターの一貫性、スタイル変更、部分的な編集
Kling AI IMAGE 3.0 Omni	画像シリーズのワークフローと、2K／4Kでの直接生成オプションを備えた、現在提供中のホスト型ナラティブ画像モデルです	ストーリーボード、連続性のある画像シーケンス、高解像度のマスター画像
Kling IMAGE O1／2.1 および旧 Open Platform ページ	アカウントや API の利用状況に依存する場合がある、以前のホスト型ワークフローです	ライブ環境で機能を確認したうえで、既存の承認済みパイプラインを再現する場合に限ります
オープン Kolors チェックポイント	2024 年に公開された Kuaishou の潜在拡散モデルで、重みとコードとして配布されています	ローカルでのバイリンガルなテキスト画像生成、および別途公開された img2img、IP-Adapter、ControlNet、インペインティング、FaceID、DreamBooth-LoRA の各ワークフロー

文書化された事実、2026-07-10 に確認済み： Kling の概要では、ホスト型画像機能が Kuaishou の Kolors 技術を基盤としていると説明されていますが、IMAGE 3.0、IMAGE 3.0 Omni、O1 が Kwai-Kolors/Kolors-diffusers と同じチェックポイントまたは API 契約であることを示す根拠は公開されていません。直接的な裏付けがない限り、これらのサービス間でパラメーター、ライセンス、ベンチマークの結論を流用しないでください。

制作上の判断を整理する

生成サービスを開く前に、簡潔な概要を記録します。

• 成果物、配置先、トリミング／アスペクト比、必要なピクセル数、最終画像の枚数、納期。
• 被写体、動作、環境、構図、照明、素材、配色、正確な表示テキストの要否。
• 参照画像が、アイデンティティ、製品の形状、ポーズ／構図、環境、スタイルのどれを左右するか。
• 変えてはならない要素と、変更してもよい要素。
• すべての参照画像の権利、商標／ロゴに関する制約、肖像の同意、プライバシー分類。
• ホスト型かローカル実行か、地域／アカウント、保持要件、コスト上限。
• フル解像度での合格／不合格基準。

バージョン番号ではなく、必要な成果に基づいて方法を選びます。

1. 1枚だけの作成、複数の参照画像を使った構図、スタイル変換、または平易な文章での編集には IMAGE 3.0 を使用します。
2. 成果物にシリーズモードや、2K／4K 画像の直接生成が必要な場合は IMAGE 3.0 Omni を使用します。「ネイティブ」や品質に関する説明は、返された画像のサイズとファイルを確認するまでは、プロバイダーによる説明として扱います。
3. データを管理下のインフラストラクチャ内にとどめる必要がある場合、決定論的なシード／スケジューラーが重要な場合、またはオープンなアダプターが必要な場合は、オープン Kolors をローカルで使用します。重みをダウンロードする前に、カスタムモデルのライセンスを確認してください。
4. who_am_i、ライブコンソール、または最新の公式 API ページで利用資格とパラメーターを確認できる場合に限り、旧ホスト型モデルを使用します。記憶を頼りにレガシーモデルを選ばないでください。

デフォルトでは、許容できる最低解像度で画像を1枚だけ生成します。コストを再見積もりし、承認を得ずに、サンプルを9枚のバッチにしたり、4K を有効にしたり、参照画像を追加したり、モデルを変更したりしないでください。

実際に文書化されている内容を把握する

以下のホスト上で確認できる情報は、2026-07-10 時点のものです。

• Kling は 2026 年 2 月に IMAGE 3.0 と IMAGE 3.0 Omni を世界展開しました。公式ガイドでは、現在の 3.0 作成ワークフローで最大 10 枚の参照画像を使用できると説明されています。
• IMAGE 3.0 は、テキストからの作成、参照画像をもとにした作成、スタイル／キャラクター／ポートレートのガイダンス、複数画像のブレンド、自然言語による部分編集に対応しています。
• IMAGE 3.0 Omni は、テキストから画像、画像から画像、複数画像のシリーズ作成のワークフローに加え、2K／4K の直接出力に対応しています。
• 2026 年 6 月のファーストパーティのマーケティングページでは、2K／4K の直接出力が Omni 独自なのか、IMAGE 3.0 ファミリー全体に適用されるのかについて記述が一致していません。モデル比較ガイドではこの差別化要素を Omni に割り当てていますが、ほかの投稿ではより広く説明されています。ファミリー名だけを根拠に対応を保証せず、現在のモデル選択画面と返されたピクセル数を確認してください。
• 一般向けのウェブ画像ガイドでは、中国語と英語のプロンプト、各辺が 300 px 以上の最大 10 MB の JPG／PNG 参照画像、記載されている 7 種類の 1K アスペクト比、最大 9 枚の出力に対応しています。これらのウェブ版の制限が API や CLI にも自動的に適用されるわけではありません。
• 現在のグローバル Open Platform では、中国国外のサーバー向けに https://api-singapore.klingai.com を使用します。新しい API キーはすべてのモデルで使用できます。認証ページによると、従来の AccessKey／SecretKey JWT 認証情報は 3.0 以前のモデルのみを対象としています。
• 公式 Kling MCP／CLI は、who_am_i、text_to_image、image_to_image、query_tasks、ファイルアップロード、クレジット／メンバーシップ情報の取得を提供します。FAQ によると、結果 URL は 24 時間有効で、送信済みタスクはキャンセルできません。アシスタント向けインターフェースの上限は 5 QPS で、現在ここで使用できるのは有料の Personal ワークスペースのクレジットのみです。ボーナスクレジットと Team ワークスペースのクレジットは対象外です。

現在、複数世代の画像 API の公開ページがありますが、今回の確認時には、権限のあるセッションなしではクライアント側でレンダリングされるリクエスト表を確実に読み取れませんでした。そのため、次の点に従ってください。

• CLI 呼び出しを組み立てる直前に、kling who_am_i、kling text_to_image --help、kling image_to_image --help を実行してください。
• REST フィールドとエンドポイント固有の料金については、ログイン済みの Open Platform ページ、または公式のエクスポート済みサンプルを使用してください。
• fal.ai、Replicate、Kie、PiAPI、Atlas、その他のゲートウェイのスキーマを Kling ネイティブのスキーマの代わりに使用しないでください。
• 画像モデルのトークン、料金、同時実行数、バッチの最大数、参照画像のエンコード形式、タスク結果の形式、出力ホスト名は、ライブ情報で確認できない限り、変動するものとして扱ってください。

旧スタンドアロンの作成サービス kolors.kuaishou.com は 2024-08-23 に終了し、ユーザーには中国の Kling プラットフォームが案内されました。その告知では、過去の作品は移行されないとも説明されていました。中国のプラットフォームとグローバル版 kling.ai の間で、アカウント、クレジット、アセット、認証情報が移行できると保証しないでください。一般用途に適した、現在の中国本土向けファーストパーティ API のホスト名は確認できていません。推測せず、中国のアカウント契約から確認してください。

Kling の語彙で画像を指示する

Kling の初心者向けガイドでは、5W1H を説明の手段として使っています。キーワードを機械的に並べるのではなく、必要な要素を網羅できているか確認するために使いましょう。

Subject and defining features; action or state; environment and time.
Framing, viewpoint, and spatial relationships. Lighting and palette.
Materials or rendering medium. Exact visible text: “...”.
Preserve: [invariants]. Change: [target and desired result].

制作上のヒューリスティック：

• 最初に伝わるべき被写体と関係性を先に書きます。カメラ、光、仕上げは、シーンの構成の後に置きます。
• 観察できる形状を具体的に指定します。「ダイナミックな広告向けアングル」ではなく、「3/4 の視点、右側にハンドルが見える」とします。
• テキストの場合は、正確な文字列を引用し、それが載る面と階層を指定して、生成するバリエーションを少数に絞ったうえで、すべての字形を確認します。プロバイダーがテキスト描画の改善をうたっていても、スペルが保証されるわけではありません。
• 見た目の特性を指定しない「8K、傑作、超高精細」など、箔をつけるだけの言葉を並べるのは避けます。
• 制約は一度に 1 つだけ変更します。プロンプト、選択したサーフェス／モデルのラベル、アカウントで確認したパラメーター、参照画像とハッシュ、出力 ID、寸法、クレジット、選択理由を保存します。

部分的な編集では、プロバイダーが直接示しているパターンを使います。

Keep everything else unchanged. Change [specific object/property in a named region]
from [current state] to [desired state]. Preserve [identity, logo geometry,
composition, lighting, shadows, material texture, and background as applicable].

複数の参照画像を使う場合は、それぞれの役割を平易な言葉で指定します。「画像 1 は製品の形状とラベル、画像 2 は大理石の表面、画像 3 は涼しげな窓辺の光の雰囲気だけを提供します」のようにします。「参照」が同一性の固定を意味すると決めつけないでください。一般的なウェブのワークフローでは、参照強度を高くすると、結果はプロンプトから離れて参照画像に近づきます。プロンプトとシード／その他のコントロールを一定に保って調整してください。

シリーズを作る場合は、生成前に一貫性のためのシートを作成します。

• 変更できないキャラクター／製品の情報
• 環境マップ、時刻、パレット、レンズ表現
• 各フレームにつき 1 つの動作と構図
• 隣接するフレーム間で許容される一貫性の変化
• 最終的なシーケンスの順序と、トリミングしても問題のない領域

長いシリーズを作る前に、少数のフレームで一貫性を検証します。隣接するフレームを並べて確認してください。単体では優れた画像でも、シーケンスとしては失敗することがあります。

スキーマを捏造せずにホスト型サービスを運用する

ターミナルでのワークフローには公式 CLI、会話型のワークフローには公式 MCP を使います。同じアシスタントには両方ではなく、どちらか一方だけをインストールして、ルーティングを確定させてください。CLI にはグローバルサイトと中国サイトのアカウント向けに別々のパッケージがあります。どちらも kling をインストールしますが、認証情報とアカウントの地域は相互に使えません。2026-07-10 時点で公式に文書化されているセットアップは次のとおりです。

npm install -g @klingai/[email protected] --registry=https://registry.npmjs.org
# China-site account only: npm install -g @klingai/[email protected] --registry=https://registry.npmjs.org
kling login
kling who_am_i
kling text_to_image --help
kling image_to_image --help

kling login は OAuth トークンを CLI の認証情報ストアに保存します。そのストアをリポジトリ、スクリーンショット、サポート用バンドル、タスクログに含めないでください。本番環境では、インストールしたパッケージのバージョンと完全性を固定して記録してください。REST を直接使う場合は、API キーまたは AK/SK をサーバー上だけに保持します。ブラウザーやモバイルのコードに埋め込んだり、出力先ホストに送信したりしないでください。

有料のリクエストを送信する前に：

1. 対象のツールについて、who_am_i によるモデルとパラメーターの情報を取得します。
2. 現行の公式コンソール／料金情報で正確なクレジット見積もりを確認します。画像の料金は、ここで固定して記載できるほど安定していませんでした。
3. モデルトークン／ラベル、モード、アスペクト比／解像度、出力数、参照画像、正確な生成コマンド／リクエストを明記します。
4. 有限の正の見積もりと、それをカバーする有限の正の上限を設定します。
5. その計画とすべての参照画像の SHA-256 値からハッシュを生成し、ユーザーがそのハッシュを承認するまで待ちます。
6. 送信前に計画を永続化します。CLI がタイムアウトした場合や送信後の応答が不正な形式だった場合は、結果が不確定です。自動的に再送せず、kling query_tasks、アカウント履歴、クレジットを照合してください。

完全な例：公式 CLI による単回の送信と取得

最新の @klingai/cli-global または @klingai/cli-cn の README に加え、実行時の who_am_i と各コマンドのヘルプを使用します。このガードは、ドキュメントに記載された静止画用の引数ベクトルを自ら構築します。位置引数のプロンプト、明示的な --model、--aspectRatio、--imgResolution、固定値の --imageCount 1 です。画像から画像への生成では、ローカルの --image スナップショットを1つ以上追加します。オペレーターは、その引数ベクトルに対する最新のコンソール見積もりを指定します。スクリプトがプロバイダーの料金を算出することはありません。デフォルトではドライランです。

ドライランの前に kling who_am_i --quiet を取得し、その正確な JSON を --who-am-i-json に渡します。新しい UUIDv4 の試行 ID、オペレーターがレビューした結果ホストの許可リスト、および {path} プレースホルダーをちょうど1つ含むスキャナーの引数ベクトルを使用します。表示されたハッシュを一度だけ承認します。実行時には CLI のパス、ハッシュ、バージョン、および who_am_i ドキュメント全体が再検証され、O_EXCL で試行が確保されます。同じ試行 ID で再実行したり、並行して実行したりすることはできません。

#!/usr/bin/env python3
import argparse, hashlib, http.client, io, ipaddress, json, math, os, shutil
import socket, ssl, stat, subprocess, tempfile, threading, time, uuid
from pathlib import Path
from urllib.parse import urlsplit
from PIL import Image, UnidentifiedImageError

MAX_REFERENCE_BYTES = 10 * 1024 * 1024
MAX_PIXELS = 40_000_000
MAX_CLI_BYTES = 1024 * 1024
Image.MAX_IMAGE_PIXELS = MAX_PIXELS

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

def atomic_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = None
    try:
        with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent,
                prefix=path.name, suffix=".tmp", delete=False) as f:
            tmp = Path(f.name); os.chmod(tmp, 0o600)
            json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path); tmp = None
    finally:
        if tmp is not None and tmp.exists(): tmp.unlink()

def claim_once(path, value):
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    fd = os.open(path, flags, 0o600)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            fd = None
            json.dump(value, f, sort_keys=True); f.flush(); os.fsync(f.fileno())
    finally:
        if fd is not None: os.close(fd)

def read_json(path, cap=MAX_CLI_BYTES):
    raw = path.read_bytes()
    if not raw or len(raw) > cap: raise ValueError("ledger JSON size rejected")
    try: value = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError("ledger JSON invalid") from exc
    if not isinstance(value, dict): raise ValueError("ledger JSON must be an object")
    return value

class LaunchError(Exception): pass

def run_bounded(argv, timeout, cap=MAX_CLI_BYTES, stdout_json_callback=None):
    try:
        proc = subprocess.Popen(argv, shell=False, stdin=subprocess.DEVNULL,
                                stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    except OSError as exc:
        raise LaunchError(str(exc)) from exc
    out, err, lock, overflow = bytearray(), bytearray(), threading.Lock(), threading.Event()
    callback_done, callback_error = False, []
    total = 0
    def drain(stream, bucket, inspect_json=False):
        nonlocal total, callback_done
        while True:
            chunk = stream.read1(65536) if hasattr(stream, "read1") else stream.read(65536)
            if not chunk: break
            candidate = None
            with lock:
                room = max(0, cap - total); bucket.extend(chunk[:room]); total += len(chunk)
                if total > cap: overflow.set()
                if inspect_json and stdout_json_callback is not None and not callback_done:
                    candidate = bytes(out)
            if candidate is not None:
                try: text = candidate.decode("utf-8-sig")
                except UnicodeDecodeError: text = ""
                candidates = [text] + [line for line in text.splitlines() if line.strip()]
                for item in candidates:
                    try: value = json.loads(item)
                    except json.JSONDecodeError: continue
                    try: callback_done = bool(stdout_json_callback(value))
                    except BaseException as exc:
                        callback_error.append(exc); overflow.set(); break
                    if callback_done: break
            if overflow.is_set():
                try: proc.kill()
                except OSError: pass
    threads = [threading.Thread(target=drain, args=(proc.stdout, out, True), daemon=True),
               threading.Thread(target=drain, args=(proc.stderr, err, False), daemon=True)]
    for thread in threads: thread.start()
    timed_out = False
    try: code = proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True; proc.kill(); code = proc.wait()
    for thread in threads: thread.join()
    if callback_error: raise RuntimeError("could not persist generation ID from stdout") from callback_error[0]
    return {"launched": True, "code": code, "stdout": bytes(out), "stderr": bytes(err),
            "timeout": timed_out, "overflow": overflow.is_set()}

def require_good(run, stage):
    if run["timeout"] or run["overflow"] or run["code"] != 0:
        raise RuntimeError(f"{stage} failed or exceeded a local bound")
    try: return json.loads(run["stdout"].decode("utf-8-sig"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise RuntimeError(f"{stage} returned invalid JSON") from exc

def cli_fingerprint():
    found = shutil.which("kling")
    if not found: raise ValueError("kling is not on PATH")
    path = Path(found).resolve(strict=True)
    if not path.is_file() or not 0 < path.stat().st_size <= 16 * 1024 * 1024:
        raise ValueError("unexpected CLI launcher size")
    data = path.read_bytes()
    run = run_bounded([str(path), "--version"], 30, 65536)
    if run["timeout"] or run["overflow"] or run["code"] != 0: raise ValueError("cannot verify CLI version")
    version = run["stdout"].decode("utf-8", "strict").strip()
    if not version or len(version) > 200: raise ValueError("invalid CLI version output")
    return {"path": str(path), "sha256": hashlib.sha256(data).hexdigest(), "version": version}

def load_reference(path):
    if path.is_symlink(): raise ValueError("reference symlinks are rejected")
    resolved = path.resolve(strict=True)
    with resolved.open("rb") as f:
        info = os.fstat(f.fileno())
        if not stat.S_ISREG(info.st_mode) or not 0 < info.st_size <= MAX_REFERENCE_BYTES:
            raise ValueError("reference must be a bounded regular file")
        data = f.read(MAX_REFERENCE_BYTES + 1)
    if len(data) != info.st_size: raise ValueError("reference changed while reading")
    head = data[:12]
    if head.startswith(b"\xff\xd8\xff"): fmt, exts, mime = "JPEG", {"jpg", "jpeg"}, "image/jpeg"
    elif head.startswith(b"\x89PNG\r\n\x1a\n"): fmt, exts, mime = "PNG", {"png"}, "image/png"
    else: raise ValueError("reference must be JPEG or PNG by signature")
    if resolved.suffix.lower().lstrip(".") not in exts: raise ValueError("extension/signature mismatch")
    try:
        with Image.open(io.BytesIO(data)) as im:
            dims, decoded = im.size, im.format
            if min(dims) < 300 or dims[0] * dims[1] > MAX_PIXELS: raise ValueError("reference dimensions rejected")
            im.verify()
        with Image.open(io.BytesIO(data)) as im: im.load()
    except (UnidentifiedImageError, OSError, Image.DecompressionBombError) as exc:
        raise ValueError("reference did not fully decode") from exc
    if decoded != fmt: raise ValueError("decoder/signature mismatch")
    return resolved, data, {"sha256": hashlib.sha256(data).hexdigest(), "bytes": len(data),
                            "dimensions": list(dims), "mime": mime}

def build_argv(cli, operation, model, prompt, ratio, resolution, references):
    argv = [cli, operation, "--model", model]
    for reference in references: argv += ["--image", reference]
    return argv + [prompt, "--aspectRatio", ratio, "--imgResolution", resolution,
                   "--imageCount", "1", "--quiet"]

def find_first(value, keys):
    if isinstance(value, dict):
        for key in keys:
            if key in value: return value[key]
        for child in value.values():
            found = find_first(child, keys)
            if found is not None: return found
    elif isinstance(value, list):
        for child in value:
            found = find_first(child, keys)
            if found is not None: return found
    return None

def generation_id_from(value):
    generation_id = find_first(value, ("generation_id", "generationId"))
    if generation_id is None: return None
    if not isinstance(generation_id, str) or not generation_id.strip() or len(generation_id) > 512:
        raise ValueError("generation_id invalid")
    return generation_id.strip()

def task_state(value):
    raw = find_first(value, ("task_status", "taskStatus", "status", "state"))
    status = str(raw).strip().lower() if raw is not None else "unknown"
    if status in {"success", "succeed", "succeeded", "complete", "completed"}: return "success"
    if status in {"failed", "failure", "rejected", "cancelled", "canceled"}: return "failure"
    if status in {"submitted", "queued", "pending", "processing", "running"}: return "nonterminal"
    return "unknown"

def result_url(value):
    works = find_first(value, ("works",))
    if not isinstance(works, list) or len(works) != 1 or not isinstance(works[0], dict):
        raise ValueError("success must contain exactly one work")
    url = works[0].get("url")
    if not isinstance(url, str) or not url: raise ValueError("work URL missing")
    return url, works[0].get("work_id") or works[0].get("id")

def allowed_host(host, rules):
    for rule in rules:
        rule = rule.lower().rstrip(".")
        if rule.startswith(".") and (host == rule[1:] or host.endswith(rule)): return True
        if host == rule: return True
    return False

class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host, ip, timeout=30):
        super().__init__(host, 443, timeout=timeout, context=ssl.create_default_context()); self.ip = ip
    def connect(self):
        sock = socket.create_connection((self.ip, self.port), self.timeout)
        self.sock = self._context.wrap_socket(sock, server_hostname=self.host)

def acquire(url, attempt_dir, rules, byte_cap, scanner_template):
    parsed = urlsplit(url)
    if parsed.scheme != "https" or parsed.username or parsed.password or parsed.fragment or parsed.port not in (None, 443):
        raise ValueError("result URL policy violation")
    host = (parsed.hostname or "").encode("idna").decode("ascii").lower().rstrip(".")
    if not host or not allowed_host(host, rules): raise ValueError("result host is not allowlisted")
    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(item).is_global for item in addresses):
        raise ValueError("result DNS is not globally routable")
    conn, tmp = PinnedHTTPSConnection(host, addresses[0]), None
    try:
        target = parsed.path or "/"
        if parsed.query: target += "?" + parsed.query
        conn.request("GET", target, headers={"Accept": "image/png,image/jpeg", "User-Agent": "kling-artifact-guard/1"})
        response = conn.getresponse()
        if 300 <= response.status < 400: raise ValueError("redirects are rejected")
        if response.status != 200: raise ValueError("artifact request failed")
        mime = response.getheader("Content-Type", "").split(";", 1)[0].lower()
        if mime not in {"image/png", "image/jpeg"}: raise ValueError("artifact MIME rejected")
        length = response.getheader("Content-Length")
        if length is not None and (not length.isdigit() or int(length) > byte_cap): raise ValueError("declared size rejected")
        digest, total = hashlib.sha256(), 0
        with tempfile.NamedTemporaryFile("wb", dir=attempt_dir, prefix="artifact-", suffix=".part", delete=False) as f:
            tmp = Path(f.name); os.chmod(tmp, 0o600)
            while True:
                chunk = response.read(65536)
                if not chunk: break
                total += len(chunk)
                if total > byte_cap: raise ValueError("streamed size rejected")
                digest.update(chunk); f.write(chunk)
            f.flush(); os.fsync(f.fileno())
        head = tmp.read_bytes()[:12]
        if head.startswith(b"\xff\xd8\xff"): fmt, expected_mime, suffix = "JPEG", "image/jpeg", ".jpg"
        elif head.startswith(b"\x89PNG\r\n\x1a\n"): fmt, expected_mime, suffix = "PNG", "image/png", ".png"
        else: raise ValueError("artifact signature rejected")
        if mime != expected_mime: raise ValueError("MIME/signature mismatch")
        with Image.open(tmp) as image:
            dims, decoded = image.size, image.format
            if dims[0] * dims[1] > MAX_PIXELS: raise ValueError("artifact pixel cap exceeded")
            image.verify()
        with Image.open(tmp) as image: image.load()
        if decoded != fmt: raise ValueError("artifact decoder/signature mismatch")
        scanner_argv = [str(tmp) if item == "{path}" else item for item in scanner_template]
        scan = run_bounded(scanner_argv, 300, MAX_CLI_BYTES)
        if scan["timeout"] or scan["overflow"] or scan["code"] != 0: raise ValueError("scanner rejected artifact")
        destination = attempt_dir / ("artifact" + suffix)
        if destination.exists(): raise ValueError("artifact destination already exists")
        os.replace(tmp, destination); tmp = None
        return {"path": str(destination), "sha256": digest.hexdigest(), "bytes": total,
                "dimensions": list(dims), "mime": expected_mime, "sourceHost": host,
                "connectedIp": addresses[0], "scannerOutputSha256":
                hashlib.sha256(scan["stdout"] + scan["stderr"]).hexdigest()}
    finally:
        conn.close()
        if tmp is not None and tmp.exists(): tmp.unlink()

parser = argparse.ArgumentParser()
parser.add_argument("--operation", choices=("text_to_image", "image_to_image"), required=True)
parser.add_argument("--region", choices=("global", "china"), required=True)
parser.add_argument("--attempt-id", required=True)
parser.add_argument("--who-am-i-json", required=True)
parser.add_argument("--model", required=True); parser.add_argument("--prompt", required=True)
parser.add_argument("--aspect-ratio", required=True); parser.add_argument("--img-resolution", required=True)
parser.add_argument("--reference", action="append", type=Path, default=[])
parser.add_argument("--estimated-credits", type=float, required=True)
parser.add_argument("--max-credits", type=float, required=True)
parser.add_argument("--allowed-result-host", action="append", default=[])
parser.add_argument("--max-artifact-bytes", type=int, default=50 * 1024 * 1024)
parser.add_argument("--scanner-argv-json", required=True)
parser.add_argument("--query-timeout", type=int, default=1800)
parser.add_argument("--run-dir", type=Path, default=Path("kling-runs"))
parser.add_argument("--execute", action="store_true")
parser.add_argument("--resume", action="store_true", help="query only an already claimed attempt with a recorded generation ID")
args = parser.parse_args()
try: attempt = uuid.UUID(args.attempt_id); who_am_i = json.loads(args.who_am_i_json); scanner = json.loads(args.scanner_argv_json)
except (ValueError, json.JSONDecodeError) as exc: raise SystemExit("attempt/identity/scanner input invalid") from exc
if attempt.version != 4 or str(attempt) != args.attempt_id.lower(): raise SystemExit("attempt-id must be canonical UUIDv4")
if not isinstance(who_am_i, dict): raise SystemExit("who-am-i-json must be the exact JSON object")
if not isinstance(scanner, list) or not scanner or not all(isinstance(x, str) and x for x in scanner) or scanner.count("{path}") != 1:
    raise SystemExit("scanner argv must be a string array with one {path}")
if not math.isfinite(args.estimated_credits) or args.estimated_credits <= 0: raise SystemExit("estimate must be finite and positive")
if not math.isfinite(args.max_credits) or args.max_credits <= 0 or args.estimated_credits > args.max_credits:
    raise SystemExit("covering ceiling must be finite and positive")
if not 0 < args.max_artifact_bytes <= 250 * 1024 * 1024 or not 1 <= args.query_timeout <= 7200:
    raise SystemExit("output policy bounds rejected")
if args.resume and not args.execute: raise SystemExit("--resume requires --execute")
if args.operation == "text_to_image" and args.reference: raise SystemExit("text_to_image cannot have references")
if args.operation == "image_to_image" and not args.reference: raise SystemExit("image_to_image requires references")
cli = cli_fingerprint(); loaded = [load_reference(path) for path in args.reference]
run_root = args.run_dir.resolve()
attempt_dir = run_root / str(attempt)
placeholders = [f"<reference:{item[2]['sha256']}>" for item in loaded]
argv = build_argv(cli["path"], args.operation, args.model, args.prompt, args.aspect_ratio, args.img_resolution, placeholders)
policy = {"imageCount": 1, "urlField": "works[0].url", "allowedHosts": sorted(set(args.allowed_result_host)),
          "httpsOnly": True, "redirects": False, "maxBytes": args.max_artifact_bytes,
          "maxPixels": MAX_PIXELS, "scannerArgv": scanner}
plan = {"attemptId": str(attempt), "region": args.region, "cli": cli, "whoAmI": who_am_i,
        "runRoot": str(run_root), "attemptDirectory": str(attempt_dir),
        "argv": argv, "settings": {"model": args.model, "aspectRatio": args.aspect_ratio,
        "imgResolution": args.img_resolution, "imageCount": 1},
        "promptSha256": hashlib.sha256(args.prompt.encode()).hexdigest(),
        "references": [item[2] for item in loaded], "estimatedCredits": args.estimated_credits,
        "maxCredits": args.max_credits, "costBasis": "operator estimate from current console for exact argv",
        "outputPolicy": policy}
approval = hashlib.sha256(canonical(plan)).hexdigest()
print(json.dumps({"dryRun": not args.execute, "approvalSha256": approval, "plan": plan}, indent=2))
if not args.execute: raise SystemExit(0)
if not policy["allowedHosts"]: raise SystemExit("execution requires an operator-reviewed result-host allowlist")
if os.environ.get("KLING_APPROVED_SHA256") != approval: raise SystemExit("approval hash mismatch")
if cli_fingerprint() != cli: raise SystemExit("CLI identity changed after approval")
live_identity_run = run_bounded([cli["path"], "who_am_i", "--quiet"], 60)
live_identity = require_good(live_identity_run, "who_am_i")
if canonical(live_identity) != canonical(who_am_i): raise SystemExit("account/workspace/capabilities changed after approval")
run_root.mkdir(parents=True, exist_ok=True)
if run_root.resolve(strict=True) != Path(plan["runRoot"]): raise SystemExit("approved run root changed")
if attempt_dir.parent.resolve(strict=True) != run_root.resolve(strict=True): raise SystemExit("attempt directory escaped approved run root")
ledger = attempt_dir / "submission.json"
def ledger_value(state, **extra):
    value = {"state": state, "approvalSha256": approval, "attemptId": str(attempt),
             "runRoot": str(run_root), "updatedUnix": time.time()}
    value.update(extra); return value

if args.resume:
    if not attempt_dir.is_dir(): raise SystemExit("approved attempt directory does not exist")
    claim = read_json(attempt_dir / "claim.json"); current = read_json(ledger)
    for record in (claim, current):
        if record.get("approvalSha256") != approval or record.get("attemptId") != str(attempt) or record.get("runRoot") != str(run_root):
            raise SystemExit("attempt record does not match this approval and run root")
    generation_id = generation_id_from(current)
    if generation_id is None: raise SystemExit("ambiguous attempt has no recoverable generation ID; do not create")
else:
    attempt_dir.mkdir(mode=0o700, parents=False, exist_ok=True)
    claim_once(attempt_dir / "claim.json", ledger_value("claimed"))
    atomic_json(ledger, ledger_value("claimed", plan=plan))
    observed = {"generationId": None}
    def persist_generation_id(value):
        generation_id = generation_id_from(value)
        if generation_id is None: return False
        if observed["generationId"] not in (None, generation_id): raise RuntimeError("conflicting generation IDs")
        observed["generationId"] = generation_id
        atomic_json(ledger, ledger_value("generation_id_observed", generationId=generation_id,
                    note="Create is consumed; query this ID only"))
        return True
    try:
        with tempfile.TemporaryDirectory(prefix="kling-approved-") as td:
            private = Path(td); os.chmod(private, 0o700); snapshot_paths = []
            for index, (_, data, evidence) in enumerate(loaded):
                snapshot = private / f"reference-{index}.img"
                with snapshot.open("xb") as f:
                    os.chmod(snapshot, 0o600); f.write(data); f.flush(); os.fsync(f.fileno())
                if hashlib.sha256(snapshot.read_bytes()).hexdigest() != evidence["sha256"]: raise RuntimeError("snapshot mismatch")
                snapshot_paths.append(str(snapshot))
            exec_argv = build_argv(cli["path"], args.operation, args.model, args.prompt,
                                   args.aspect_ratio, args.img_resolution, snapshot_paths)
            atomic_json(ledger, ledger_value("submitting"))
            try: submitted = run_bounded(exec_argv, 1800, stdout_json_callback=persist_generation_id)
            except LaunchError:
                atomic_json(ledger, ledger_value("local_launch_failed_consumed")); raise
            submit_json = None
            try: submit_json = json.loads(submitted["stdout"].decode("utf-8-sig"))
            except (UnicodeDecodeError, json.JSONDecodeError): pass
            if submit_json is not None: persist_generation_id(submit_json)
            if submitted["timeout"] or submitted["overflow"] or submitted["code"] != 0:
                raise RuntimeError("submission outcome is ambiguous")
            submit_json = require_good(submitted, "submission")
            generation_id = generation_id_from(submit_json)
            if generation_id is None: raise RuntimeError("generation_id missing")
            atomic_json(ledger, ledger_value("submitted", generationId=generation_id, submitOutputSha256=
                        hashlib.sha256(submitted["stdout"] + submitted["stderr"]).hexdigest()))
    except LaunchError as exc: raise SystemExit(f"CLI did not launch; attempt is consumed: {exc}") from exc
    except BaseException as exc:
        current = ledger_value("ambiguous", note="Never resubmit; reconcile query_tasks/account credits")
        if observed["generationId"] is not None: current["generationId"] = observed["generationId"]
        if "submitted" in locals():
            current.update({"submitOutputSha256": hashlib.sha256(submitted["stdout"] + submitted["stderr"]).hexdigest(),
                            "submitExitCode": submitted["code"], "submitTimedOut": submitted["timeout"],
                            "submitOverflow": submitted["overflow"]})
        atomic_json(ledger, current); raise SystemExit("post-launch outcome is ambiguous") from exc

deadline, failures, terminal = time.monotonic() + args.query_timeout, 0, None
while time.monotonic() < deadline:
    try:
        queried = run_bounded([cli["path"], "query_tasks", generation_id, "--quiet"], 60)
        query_json = require_good(queried, "query_tasks"); failures = 0
        state = task_state(query_json)
        if state in {"success", "failure"}: terminal = state; break
    except (LaunchError, RuntimeError):
        failures += 1
        if failures >= 3: break
    time.sleep(min(10, 2 ** failures))
if terminal != "success":
    atomic_json(ledger, ledger_value("terminal_failure" if terminal == "failure" else "submitted_unreconciled",
                generationId=generation_id,
                note="Do not create a duplicate; query this generation ID"))
    raise SystemExit("task did not reach a retrievable success")
try:
    url, work_id = result_url(query_json)
    artifact = acquire(url, attempt_dir, policy["allowedHosts"], policy["maxBytes"], scanner)
    manifest = {"generationId": generation_id, "workId": work_id, "artifact": artifact,
                "model": args.model, "settings": plan["settings"], "references": plan["references"],
                "estimatedCredits": args.estimated_credits, "approvalSha256": approval,
                "review": {"syntheticDisclosure": "required", "creativeQa": "pending", "rightsReview": "required"}}
    atomic_json(attempt_dir / "manifest.json", manifest)
    atomic_json(ledger, ledger_value("artifact_saved", generationId=generation_id,
                artifactSha256=artifact["sha256"]))
    print(json.dumps(manifest, indent=2))
except BaseException as exc:
    atomic_json(ledger, ledger_value("artifact_failed", generationId=generation_id,
                note="Recover this task; never regenerate solely for the artifact"))
    raise SystemExit("task succeeded but secure artifact acquisition failed") from exc

ラッパーは生のタスク JSON や署名付き URL を表示または永続化しません。承認は解決済みの実行ルートと試行ディレクトリに紐づき、すべての確保／台帳レコードにもそのルートが記録されます。そのため、同じ UUID と承認情報を別の場所にコピーしても実行は許可されません。CLI の出力は上限を設けてストリーミングし、完全な generation_id JSON が現れた場合は、プロセスが後からタイムアウトしたり、出力が上限を超えたり、ゼロ以外の終了コードを返したりしても、その時点ですぐに永続化します。--resume --execute は同じ承認／ルートのレコードを検証し、query_tasks のみを実行します。作成処理の分岐には進みません。ラッパーは TLS ホスト名検証を通じて検証済みのパブリック DNS アドレスを固定し、リダイレクトを拒否します。また、宣言されたサイズとストリーミングされたサイズ、MIME タイプ、シグネチャ、完全なデコード、ピクセル数を検証し、承認済みのスキャナーコマンドを正確に実行したうえで、成果物と許可リストに登録されたマニフェストをアトミックに昇格させます。失敗した場合も、消費済みの試行は消費済みのままです。既存の generation_id を復旧してください。別の URL を取得するために代替の生成を行ってはいけません。

オープンな Kolors チェックポイントを意図的に実行する

公式リポジトリでは、Kolors を256トークンのコンテキストを持つ、中国語と英語のバイリンガル対応の潜在拡散モデルとして説明しています。2024年の評価では、1,000件を超えるプロンプト、14カテゴリ、12の評価軸、50名の画像専門家が使用されました。報告された結果はチーム独自の2024年4月版のベンチマークとして扱い、現在の普遍的なランキングとは見なさないでください。

リポジトリでは、デフォルトの Euler スケジューラーをガイダンス5.0、ステップ数50で使用する方法、または EDM DPM-Solver Multistep をガイダンス5.0、ステップ数25で使用する方法を推奨しています。また、img2img、IP-Adapter-Plus、FaceID-Plus、Canny／depth／pose ControlNets、inpainting、DreamBooth-LoRA も個別に公開しています。複数の制御方式を組み合わせる前に、使用する方式を1つ選択してください。

完全な例：ローカルでのバイリンガル対応テキストから画像への生成

この例では、まずドライランを行い、明示されていないネットワークからのダウンロードを拒否します。また、不変の40桁の16進数チェックポイントコミット、依存関係ロックのハッシュ、およびレビュー済みの MODEL_LICENSE のハッシュを承認に紐づけます。プロバイダー推奨の Euler／50ステップのベースラインを使用し、実際のランタイムバージョンを記録します。これはあくまで一例であり、これらの設定がすべてのプロンプトに適しているという意味ではありません。

依存関係ロックは、kolors-runtime-lock/v1 スキーマに準拠した UTF-8 JSON です。schema、python、packages のみを含みます。python には正確な implementation と version の文字列を含め、各パッケージのエントリーには正確な version と recordSha256 の値を含めます。recordSha256 は、インストール済みディストリビューションの UTF-8 RECORD テキストの小文字 SHA-256 値です。ロックには少なくとも torch、diffusers、transformers、Pillow を含める必要があります。ML モジュールをインポートする前に、記載されたすべてのディストリビューションが検証されます。例：

{
  "schema": "kolors-runtime-lock/v1",
  "python": {"implementation": "CPython", "version": "3.11.9"},
  "packages": {
    "torch": {"version": "2.2.1", "recordSha256": "<64 lowercase hex>"},
    "diffusers": {"version": "0.25.0", "recordSha256": "<64 lowercase hex>"},
    "transformers": {"version": "4.37.2", "recordSha256": "<64 lowercase hex>"},
    "Pillow": {"version": "10.2.0", "recordSha256": "<64 lowercase hex>"}
  }
}

#!/usr/bin/env python3
import argparse, hashlib, importlib.metadata, json, os, platform, re, stat, tempfile
from pathlib import Path

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

def evidence(path, cap):
    if path.is_symlink(): raise ValueError("symlinks are rejected")
    resolved = path.resolve(strict=True)
    with resolved.open("rb") as f:
        info = os.fstat(f.fileno())
        if not stat.S_ISREG(info.st_mode) or not 0 < info.st_size <= cap: raise ValueError("evidence file rejected")
        data = f.read(cap + 1)
    if len(data) != info.st_size: raise ValueError("evidence file changed while reading")
    return resolved, data, hashlib.sha256(data).hexdigest()

def validate_dependency_lock(path, data, digest):
    try: lock = json.loads(data.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError("dependency lock must be UTF-8 JSON") from exc
    if not isinstance(lock, dict) or set(lock) != {"schema", "python", "packages"}:
        raise ValueError("dependency lock keys rejected")
    if lock["schema"] != "kolors-runtime-lock/v1": raise ValueError("dependency lock schema rejected")
    python_spec = lock["python"]
    if not isinstance(python_spec, dict) or set(python_spec) != {"implementation", "version"}:
        raise ValueError("dependency lock Python record rejected")
    actual_python = {"implementation": platform.python_implementation(), "version": platform.python_version()}
    if python_spec != actual_python: raise ValueError("Python runtime does not match dependency lock")
    packages = lock["packages"]
    if not isinstance(packages, dict) or not packages: raise ValueError("dependency packages rejected")
    validated = {}
    for requested_name, spec in packages.items():
        if not isinstance(requested_name, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", requested_name):
            raise ValueError("dependency distribution name rejected")
        normalized = re.sub(r"[-_.]+", "-", requested_name).lower()
        if normalized in validated: raise ValueError("duplicate normalized dependency name")
        if not isinstance(spec, dict) or set(spec) != {"version", "recordSha256"}:
            raise ValueError("dependency package record rejected")
        version, expected_record = spec["version"], spec["recordSha256"]
        if not isinstance(version, str) or not version or len(version) > 200:
            raise ValueError("dependency version rejected")
        if not isinstance(expected_record, str) or not re.fullmatch(r"[0-9a-f]{64}", expected_record):
            raise ValueError("dependency RECORD hash rejected")
        try: distribution = importlib.metadata.distribution(requested_name)
        except importlib.metadata.PackageNotFoundError as exc:
            raise ValueError(f"locked distribution is not installed: {requested_name}") from exc
        record = distribution.read_text("RECORD")
        if not isinstance(record, str) or not record: raise ValueError(f"installed distribution lacks RECORD: {requested_name}")
        actual_record = hashlib.sha256(record.encode("utf-8")).hexdigest()
        if distribution.version != version or actual_record != expected_record:
            raise ValueError(f"installed distribution does not match lock: {requested_name}")
        validated[normalized] = {"distribution": distribution.metadata["Name"], "version": distribution.version,
                                 "recordSha256": actual_record}
    required = {"torch", "diffusers", "transformers", "pillow"}
    if not required.issubset(validated): raise ValueError("dependency lock omits a required distribution")
    return {"path": str(path), "sha256": digest, "schema": lock["schema"],
            "python": actual_python, "validatedPackages": validated}

def atomic_json(path, value):
    tmp = None
    try:
        with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, suffix=".tmp", delete=False) as f:
            tmp = Path(f.name); os.chmod(tmp, 0o600)
            json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path); tmp = None
    finally:
        if tmp is not None and tmp.exists(): tmp.unlink()

parser = argparse.ArgumentParser()
parser.add_argument("--prompt", required=True)
parser.add_argument("--negative", default="")
parser.add_argument("--seed", type=int, default=66)
parser.add_argument("--output", type=Path, default=Path("kolors-output.png"))
parser.add_argument("--revision", required=True, help="immutable 40-hex Hugging Face commit")
parser.add_argument("--dependency-lock", type=Path, required=True)
parser.add_argument("--model-license", type=Path, required=True)
parser.add_argument("--allow-download", action="store_true")
parser.add_argument("--execute", action="store_true")
args = parser.parse_args()
if not re.fullmatch(r"[0-9a-fA-F]{40}", args.revision): raise SystemExit("revision must be an immutable 40-hex commit")
lock_path, lock_data, lock_hash = evidence(args.dependency_lock, 10 * 1024 * 1024)
license_path, _, license_hash = evidence(args.model_license, 1024 * 1024)
try: lock_info = validate_dependency_lock(lock_path, lock_data, lock_hash)
except ValueError as exc: raise SystemExit(f"dependency lock validation failed: {exc}") from exc
target = args.output.resolve()
plan = {"checkpoint": "Kwai-Kolors/Kolors-diffusers", "revision": args.revision.lower(),
        "scheduler": "EulerDiscreteScheduler",
        "guidanceScale": 5.0, "steps": 50, "seed": args.seed,
        "promptSha256": hashlib.sha256(args.prompt.encode()).hexdigest(),
        "negativeSha256": hashlib.sha256(args.negative.encode()).hexdigest(),
        "output": str(target), "allowDownload": args.allow_download,
        "dependencyLock": lock_info,
        "modelLicense": {"path": str(license_path), "sha256": license_hash}}
approval = hashlib.sha256(canonical(plan)).hexdigest()
print(json.dumps({"dryRun": not args.execute, "approvalSha256": approval, "plan": plan}, indent=2))
if not args.execute: raise SystemExit(0)
if os.environ.get("KOLORS_APPROVED_SHA256") != approval: raise SystemExit("plan approval mismatch")
if os.environ.get("KOLORS_MODEL_LICENSE_ACCEPTED_SHA256") != license_hash:
    raise SystemExit("review MODEL_LICENSE and accept this exact SHA-256")
if target.exists() or target.with_suffix(target.suffix + ".json").exists(): raise SystemExit("output already exists")

import PIL, diffusers, torch, transformers
from diffusers import EulerDiscreteScheduler, KolorsPipeline
from PIL import Image
module_versions = {"torch": str(torch.__version__), "diffusers": str(diffusers.__version__),
                   "transformers": str(transformers.__version__), "pillow": str(PIL.__version__)}
for name, version in module_versions.items():
    if version != lock_info["validatedPackages"][name]["version"]:
        raise SystemExit(f"imported module version does not match validated lock: {name}")
Image.MAX_IMAGE_PIXELS = 40_000_000
pipe = KolorsPipeline.from_pretrained(
    plan["checkpoint"], torch_dtype=torch.float16, variant="fp16",
    revision=plan["revision"], local_files_only=not args.allow_download,
).to("cuda")
pipe.scheduler = EulerDiscreteScheduler.from_config(pipe.scheduler.config)
generator = torch.Generator(device=pipe.device).manual_seed(args.seed)
image = pipe(prompt=args.prompt, negative_prompt=args.negative, guidance_scale=5.0,
             num_inference_steps=50, generator=generator).images[0]
target.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(dir=target.parent, suffix=".png", delete=False) as f:
    tmp = Path(f.name)
try:
    image.save(tmp, format="PNG")
    with Image.open(tmp) as check:
        dimensions = list(check.size)
        if check.size[0] * check.size[1] > Image.MAX_IMAGE_PIXELS: raise ValueError("pixel cap exceeded")
        check.verify()
    with Image.open(tmp) as check: check.load()
    digest = hashlib.sha256(tmp.read_bytes()).hexdigest()
    os.replace(tmp, target)
finally:
    if tmp.exists(): tmp.unlink()
manifest = dict(plan, artifact={"path": str(target), "sha256": digest,
                                 "dimensions": dimensions, "mime": "image/png"},
                runtime={"python": lock_info["python"], "modules": module_versions,
                         "dependencyLockSha256": lock_info["sha256"],
                         "validatedPackages": lock_info["validatedPackages"]},
                review={"syntheticDisclosure": "required", "creativeQa": "pending",
                        "rightsReview": "required"})
manifest_path = target.with_suffix(target.suffix + ".json")
try: atomic_json(manifest_path, manifest)
except BaseException:
    if target.exists(): target.unlink()
    raise
print(json.dumps(manifest, indent=2))

ローカルで実行すれば、プロンプトや参照画像を Kling に送信せずに済みますが、ライセンス、データセット、同意、安全性、開示、出力レビューに関する義務がなくなるわけではありません。再現性を確保するため、チェックポイントのリビジョンと依存関係のバージョンを固定してください。シードは、ハードウェアやバージョンをまたいだバイト単位の同一性を保証するものではありません。

ステージごとの失敗への対処

失敗	対応
who_am_i に機能／モデルがない	停止します。ユーザーの承認を得た場合に限り、利用権のある現行モデルを選択します
プロンプト／参照画像／比率が無効	ローカルで修正します。変更せずに再試行しません
認証失敗	サーバー側の認証情報をローテーションするか修正します。決して表示しません
コンテンツモデレーションによる失敗	フィルターを回避しません。ポリシーと権利に照らして修正するか、依頼を拒否します
クレジット／リソースの枯渇	停止します。自動購入やワークスペースの切り替えは行いません
429／同時実行コード 1303	現在のパッケージの上限を守り、ジッターを加えた回数制限付きのバックオフを行います
クエリ中の 5xx／読み取り失敗	回数制限付きのバックオフを行って、読み取りを再試行します
作成後のタイムアウト／5xx／不正な応答	結果が不確定なものとして扱います。再送信する前に、履歴、タスククエリ、控除内容を照合します
24 時間後に結果 URL が期限切れ	サポートされている場合は既存のタスクを再クエリします。URL を復旧するためだけに重複作成しません
ローカルの CUDA OOM	寸法／バッチサイズを小さくするか、文書化されたメモリ制御を有効にするか、承認を得てハードウェアを変更します。プロンプト／シード／設定は保持します

公開ドキュメントでは、現在提供されているすべての画像生成機能にネイティブの作成用冪等性キーがあるとは保証されていません。ゲートウェイやクライアントで生成したラベルから、そのようなキーの存在を推測しないでください。

権利、データ、開示を管理する

ホスト型サービスの利用規約とオープンモデルのライセンスを分けて考えてください。

有料のグローバル API について、2026-04-21 に発効した利用規約では、Kling はモデルのトレーニングや改善に顧客データを使用せず、プロンプトと応答のデータ処理者として扱い、プロンプトと応答を30日間記録すると定められています。API のプライバシーポリシーでは、データはシンガポールのサーバーに保存され、世界各地のサポート、エンジニアリング、モデレーションの各チームがアクセスする可能性があり、国際的に移転される場合があるとされています。また、目的や法的必要性に応じて保存されますが、削除には例外があります。有料 API の「トレーニングに使用しない」という約束を、コンシューマー向けウェブアプリ、中国向けサービス、フィードバックプログラム、公開共有、または無関係な契約にまで一般化しないでください。

2026-04-21 に発効したグローバルのコンシューマー向け利用規約では、適用法の下でユーザーがコンテンツを所有するか、正当な所有者に帰属するとされており、Kling は所有権を主張しません。ただし、所有権だけで許諾の全容が決まるわけではありません。同規約では、コンテンツに対する広範な運用上および開発上の権利も Kling に付与され、モデルの開発やトレーニングへの入力の使用が明示的に認められています。また、書面による許可なく出力を商用利用、複製、配布、改変したり、派生物を作成したりすることは禁止されており、Kling のブランド表示、または「Kling AI によって生成」と目立つ形で表示することが求められます。一方、有料 API の利用規約では、API の出力に対する商用利用は制限されていないと別途定められています。また、有料会員ポリシーでは、独自の制限に従うことを条件に、会員の出力が別途許可される場合があります。非公開の参照素材をアップロードしたり、商用配布を承認したりする前に、利用形態がコンシューマー向けウェブ、 有料会員、有料 API、またはエンタープライズ契約のいずれなのかを確認してください。これは運用上の指針であり、法的助言ではありません。

オープンな Kolors について：

• リポジトリのコードには Apache-2.0 が適用されますが、重みとその派生物には別途 MODEL_LICENSE が適用されます。
• 商用利用者は、自社と関連会社の製品・サービスの月間アクティブユーザー数の合計が3億人以下の場合、ライセンサーに登録する必要があります。この基準を超える場合、権利を行使する前に明示的なライセンスを取得する必要があります。
• ライセンスでは、配布時のソースおよびライセンスに関する表示を求め、利用制限を設けています。また、出力だけではモデルの派生物に当たらないと定める一方、Kolors およびモデルの派生物を除き、モデルの成果物や出力を別のモデルの改善に使用することを禁止しています。
• モデルと出力には、正確性、安全性、非侵害性、または特定目的適合性に関する保証はありません。

いずれの利用形態でも、入力素材に対する権利を取得し、人物を識別できる場合は本人の同意を得てください。本人になりすます、欺く、同意なく性的に扱う、またはセンシティブな特性を推測する目的で、顔の参照素材を使用しないでください。ロゴ、製品に関する表示、文化的シンボル、制服、地図、公人、未成年者、および規制対象の用途は、人によるレビューで確認してください。元の作品の権利をユーザーが所有し、削除が合法である場合を除き、ウォーターマークを削除しないでください。ソースと出力のハッシュを保持し、合成コンテンツであることを開示し、適用される AI 表示規則に従ってください。ここで確認した一次資料のいずれも、すべての出力に C2PA クレデンシャルが付与されることを保証していません。

納品前の確認

候補をそれぞれ原寸と納品用のトリミングで確認してください。

• プロンプトと編集内容への準拠、および対象領域外で意図しない変更がないか
• 被写体の数、空間的な位置関係、解剖学的形状、手、目、反射、影、遠近法
• 参照素材に照らした、人物や製品の形状、およびフレーム間の連続性
• 文字の一つひとつ、ロゴの比率、商標の使用、および禁止対象のアーティファクト
• エッジのハロー、テクスチャの繰り返し、模様の歪み、継ぎ目、圧縮、色の段差、およびアルファチャンネルへの期待
• 正確な寸法、アスペクト比、MIME タイプ、バイトサイズ、カラープロファイル、および後工程との互換性
• モデレーション結果、権利と同意、プライバシー、開示、および保存・削除計画

必須の事実に反する画像は、見た目が魅力的でも却下してください。生成を繰り返しても確実に収束できない場合は、決定論的なタイポグラフィ、正確なパッケージ形状、またはピクセル単位で正確な局所変更には、従来の画像編集を使用してください。

一次情報の根拠一覧

特に記載がない限り、2026-07-10 に確認済みです：

• Kling Open Platform の概要／認証、API エラー、中国国外向けエンドポイント：https://kling.ai/document-api/quickStart%2FproductIntroduction%2Foverview および https://kling.ai/document-api/apiReference/commonInfo
• 現行の画像 API のランディングページ（スキーマの確認にはサインインが必要です）：https://kling.ai/document-api/api/image/2-1、https://kling.ai/document-api/api/image/o1/image-generation、https://kling.ai/document-api/api/image/3-0-omni/image-generation
• 公式 CLI／MCP ガイド：https://kling.ai/app/mcp/guide
• IMAGE 3.0 および 3.0 Omni ガイド：https://kling.ai/quickstart/klingai-image-3-model-user-guide および https://kling.ai/quickstart/klingai-image-3-omni-user-guide
• 公式の画像プロンプト、アスペクト比、参照ガイド：https://kling.ai/quickstart/ai-image-prompt-formula、https://kling.ai/quickstart/ai-image-size-guide、および https://kling.ai/quickstart/ai-image-reference-guide
• Kuaishou による 3.0 リリース発表：https://ir.kuaishou.com/news-releases/news-release-details/kling-ai-launches-30-model-ushering-era-where-everyone-can-be
• 2026-04-21 発効／公開の有料 API 利用規約およびプライバシーポリシー：https://kling.ai/document-api/protocols/paidServiceProtocol および https://kling.ai/document-api/protocols/privacyPolicy
• グローバルユーザー規約：https://kling.ai/docs/user-policy
• 独立した Kolors サービスの終了告知：https://kolors.kuaishou.com/
• オープンソースの Kolors リポジトリ、技術レポート、モデルライセンス：https://github.com/Kwai-Kolors/Kolors、https://github.com/Kwai-Kolors/Kolors/blob/master/imgs/Kolors_paper.pdf、および https://github.com/Kwai-Kolors/Kolors/blob/master/MODEL_LICENSE