kling-kolors-image

使用 Kling 和 Kolors 生成静态图像

首先要厘清名称。“Kling image”、“Kolors”和“可图”现在指的是相关但不可互换的产品。

产品	说明	适用场景
Kling AI IMAGE 3.0	当前托管式静态图像生成和自然语言编辑功能，支持多参考图	单张静态图、产品／角色连续性、风格重绘、局部编辑
Kling AI IMAGE 3.0 Omni	当前托管式叙事图像模型，支持图像系列工作流和直接生成 2K／4K 图像	分镜、连贯的图像序列、高分辨率母版
Kling IMAGE O1／2.1 和较早的 Open Platform 页面	较早的托管式工作流，可能取决于账号／API 权限	只有在实时确认功能后，才用于复现现有的已批准工作流
Open Kolors 检查点	快手于 2024 年发布的潜空间扩散模型，以权重和代码形式分发	本地中英双语文生图，以及单独发布的图生图、IP-Adapter、ControlNet、局部重绘、FaceID 和 DreamBooth-LoRA 工作流

已记录事实，核实日期：2026-07-10： Kling 的概述称，其托管式图像功能基于快手的 Kolors 技术，但没有公布证据表明 IMAGE 3.0、IMAGE 3.0 Omni 或 O1 与 Kwai-Kolors/Kolors-diffusers 使用相同的检查点或 API 合约。未经直接支持，切勿在这些产品之间迁移参数、许可证或基准测试结论。

明确制作决策

打开生成界面前，先记录一份简短需求说明：

• 交付内容、投放位置、裁剪／宽高比、所需像素、最终图像数量和截止时间；
• 主体、动作、环境、构图、光线、材质、色彩方案，以及任何必须准确呈现的可见文字；
• 参考图用于控制身份、产品几何形状、姿势／构图、环境还是风格；
• 哪些细节必须保持不变，哪些可以更改；
• 所有参考图的使用权、商标／标志限制、肖像权同意情况和隐私分类；
• 托管式还是本地执行、区域／账号、保留要求和成本上限；
• 全分辨率下的通过／失败标准。

根据需求选择方案，而不是根据版本号：

1. 单次创作、多参考图构图、风格迁移或自然语言编辑，使用 IMAGE 3.0。
2. 交付内容依赖系列模式或直接生成 2K／4K 图像时，使用 IMAGE 3.0 Omni。在检查返回的尺寸和文件之前，应将“原生”和质量相关说法视为服务提供方的描述。
3. 数据必须留在受控基础设施上、需要确定性的种子／调度器，或需要开源适配器时，使用本地 Open Kolors。下载权重前，先确认自定义模型的许可证。
4. 只有在 who_am_i、实时控制台或当前官方 API 页面能证明你拥有权限且参数可用时，才使用较早的托管式模型。不要凭记忆选择旧版模型。

默认只生成一张满足最低要求分辨率的图像。未经重新估算成本并获得批准，不要将样例扩展为九张图的批次、启用 4K、添加参考图或更换模型。

了解实际记录的内容

以下托管信息截至 2026-07-10 为最新：

• Kling 于 2026 年 2 月在全球推出 IMAGE 3.0 和 IMAGE 3.0 Omni。官方指南称，当前 3.0 创作工作流最多支持十张参考图。
• IMAGE 3.0 支持文本创作、参考图引导创作、风格／角色／肖像引导、多图融合，以及通过自然语言进行局部编辑。
• IMAGE 3.0 Omni 增加了文生图、图生图、多图系列工作流，以及直接输出 2K/4K 的功能。
• 2026 年 6 月的第一方营销页面对于直接输出 2K/4K 是否为 Omni 独有，还是适用于整个 IMAGE 3.0 系列，表述不一致。模型比较指南将这项差异归于 Omni，而其他帖子则将其描述为更广泛的功能。请确认实时模型选择器和返回的像素尺寸，不要仅凭系列名称承诺此功能。
• 通用网页版生图指南支持中文和英文提示词、大小不超过 10 MB 且两边尺寸均至少为 300 px 的 JPG/PNG 参考图、列出的七种 1K 比例，以及最多九张输出图。这些网页版限制并不自动适用于 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、文件上传，以及积分／会员信息查询功能。其常见问题解答称，结果 URL 的有效期为 24 小时，已提交的任务无法取消，面向助手的接口上限为 5 QPS，目前仅支持使用付费个人工作区积分，不支持奖励积分和团队工作区积分。

目前公开的多代图像 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].

制作启发：

• 优先描述必须首先呈现的主体及其关系。将相机、光线和成片效果放在场景逻辑之后。
• 明确可观察的几何关系：“三分之四视角，右侧可见把手”，而不是“动感的商业角度”。
• 对于文字，引用准确文本，指出其表面和层级，生成少量变体，然后检查每个字形。提供商声称文字渲染效果有所改进，并不代表拼写一定正确。
• 避免堆砌“8K、杰作、超精细”等彰显品质的词语，除非它们能说明具体的视觉属性。
• 一次只更改一项约束。保存提示词、所选界面／模型标签、从账户中发现的参数、参考图及其哈希值、输出 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 仅提供冷调窗光氛围。”不要假设“参考图”意味着锁定身份。在通用网页工作流中，提高参考强度会让结果更接近参考图、远离提示词；调整时应固定提示词和种子／其他控制项。

制作系列图像前，先编写连续性说明：

• 不可更改的角色／产品事实；
• 环境地图、一天中的时间、调色板和镜头语言；
• 每帧的一个动作和构图；
• 相邻帧之间允许发生的连续性变化；
• 最终序列顺序和适用于裁切的安全区域。

长系列生成前，先生成少量图像验证连续性。将相邻帧放在一起审核；单张图像即使很出色，也可能无法满足序列要求。

使用托管服务时，不要凭空编造架构

终端工作流优先使用官方 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 key 或 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 和各命令的帮助信息。此防护脚本会自行构造文档所述的静态图像 argv：位置参数提示词、显式的 --model、--aspectRatio、--imgResolution，以及固定的 --imageCount 1；图生图会添加一个或多个本地 --image 快照。操作员需提供针对该确切 argv 的当前控制台估算值——脚本不会推算服务商定价。默认执行试运行。

试运行前，捕获 kling who_am_i --quiet 的输出，并通过 --who-am-i-json 传入原样的 JSON。使用全新的 UUIDv4 尝试 ID、经操作员审核的结果主机允许列表，以及恰好包含一个 {path} 占位符的扫描器 argv。对打印出的哈希批准一次。执行时会重新检查 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 个 token 的潜在扩散模型。其 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／深度／姿态 ControlNet、图像修复和 DreamBooth-LoRA。组合使用前，一次只选择一种控制机制。

完整示例：本地中英双语文生图

此示例优先进行试运行、拒绝隐式网络下载，并将不可变的 40 位十六进制检查点提交、依赖锁哈希和经审核的 MODEL_LICENSE 哈希绑定到审批。它采用服务商推荐的 Euler／50 步基线，并记录实际运行时版本。这只是一个示例，并不表示这些设置适用于所有提示词。

依赖锁是采用 UTF-8 编码的 JSON，架构为 kolors-runtime-lock/v1。它恰好包含 schema、python 和 packages；python 包含确切的 implementation 和 version 字符串，每个软件包条目都包含确切的 version 和 recordSha256 值。recordSha256 是已安装发行版的 UTF-8 RECORD 文本的小写 SHA-256 哈希。在导入任何 ML 模块之前，必须检查锁文件至少涵盖 torch、diffusers、transformers 和 Pillow；并检查列出的每个发行版。例如：

{
  "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 显存不足	减小尺寸/批次，启用有文档说明的内存控制，或经批准后更换硬件；保留提示词/种子/配置

公开文档并未承诺当前每种图像操作都支持原生创建幂等键。不要从网关或客户端生成的标签推断存在此类幂等键。

管理权利、数据与披露

将托管服务条款与开放模型许可证分开看待。

对于付费全球 API，2026-04-21 生效的条款规定，Kling 不会使用客户数据训练或改进模型；Kling 将提示词和响应视为数据处理者，并会将提示词和响应记录保留 30 天。API 隐私政策称，数据存储在新加坡的服务器上，全球支持、工程和审核团队可能会访问数据，数据也可能被跨境传输；除此之外，数据会根据目的和法律需求进行保留，但存在删除例外。不要将付费 API 的不用于训练承诺泛化到消费级 Web 应用、中国服务、反馈计划、公开分享或无关合同。

2026-04-21 生效的全球消费者条款规定，用户依据适用法律拥有内容，或内容归其合法所有者所有，而 Kling 不主张拥有权。拥有权并不能涵盖全部许可问题：这些条款还授予 Kling 对内容广泛的运营和开发权利，明确允许使用输入内容开发和训练模型，并禁止未经书面许可对输出进行商业使用、复制、分发、修改和衍生创作，还要求标注 Kling 品牌，或醒目注明“由 Kling AI 生成”。付费 API 条款另行规定，API 输出的商业用途不受限制；付费会员政策也可能另行授权会员使用输出，但须遵守其自身限制。在上传私人参考资料或批准商业分发之前，应先确定所使用的是消费级 Web 服务、付费会员、付费 API 还是企业订单。这是操作指南，不构成法律建议。

对于开源 Kolors：

• 仓库代码采用 Apache-2.0 许可证，但权重及其衍生作品采用单独的 MODEL_LICENSE；
• 若商业用户及其关联方的产品/服务月活跃用户总数不超过 3 亿，则必须向许可方注册；超过该门槛时，行使相关权利前必须取得明确许可；
• 许可证要求分发时提供源代码/许可证声明，对用途施加限制，规定单独的输出不属于模型衍生作品，并禁止使用模型作品或输出改进其他模型，Kolors/模型衍生作品除外；
• 模型及其输出均按“现状”提供，不保证准确性、安全性、不侵权或适用于特定用途。

无论采用哪种方式，都应取得输入内容的权利，并征得可识别人物的同意。不得使用人脸参考进行冒充、欺骗、未经同意的性化处理或敏感特征推断。应通过人工审核核实徽标、产品声明、文化符号、制服、地图、公众人物、未成年人及受监管用途。除非用户拥有底层作品且删除水印符合法律规定，否则不得移除水印。应保留来源和输出的哈希值，披露合成内容的来源，并遵守适用的 AI 标注规则；此处审阅的第一方来源均未保证每个输出都带有 C2PA 凭证。

交付前审核

以原始尺寸和交付裁切范围检查每个候选结果：

• 是否遵循提示词和编辑要求；目标区域以外是否发生意外变化；
• 主体数量、空间关系、解剖结构、手部、眼睛、反射、阴影和透视；
• 对照参考图检查身份/产品几何形状及跨帧连续性；
• 每个文字字符、徽标比例、商标使用情况及违禁伪影；
• 边缘光晕、重复纹理、图案扭曲、接缝、压缩、色带及 alpha 预期；
• 精确尺寸、纵横比、MIME、字节大小、色彩配置文件及下游兼容性；
• 审核结果、权利/同意、隐私、披露及保留/删除计划。

若图像未通过必要事实核验，即使画面吸引人也应拒绝。当生成式迭代无法可靠收敛时，应使用常规图像编辑来处理确定性排版、精确包装几何形状或像素级局部修改。

第一方证据登记表

截至 2026-07-10 已核实，除非注明了来源日期：

• Kling 开放平台概览／身份验证、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
• 快手 3.0 发布公告：https://ir.kuaishou.com/news-releases/news-release-details/kling-ai-launches-30-model-ushering-era-where-everyone-can-be
• 付费 API 条款和隐私政策，生效／发布于 2026-04-21：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