luma-photon

Luma Photon 生产指南

此技能适用于由 photon-1 和 photon-flash-1 支持的旧版路径 Luma API 图像接口。请将其与 Ray 视频生成、较新的 UNI 图像工作流、Luma Agents、消费者应用、ComfyUI 封装和托管的第三方网关区分开来。尽管 Luma 当前的品牌指南称 dream-machine 这一术语已弃用，但文档中的 Photon REST 路径仍保留该字符串。

以下所有易变信息均已于 2026-07-09 根据第一方来源核实。在作出生产承诺前，请立即重新检查 API 参考、计费控制面板、速率限制页面和适用协议。

审慎选择 Photon

文档中列出的图像模型 ID 如下：

• photon-1 — 默认 Photon 模型。Luma 将其定位为更高保真度的生成模型。
• photon-flash-1 — 速度更快、成本更低的迭代模型。

请将这些描述视为提供方的定位，而非独立的质量结论。使用预期交付物进行固定的提示词和参考图对比测试，然后比较提示词遵循度、来源保真度、文本、解剖结构、材质连续性、延迟、拒绝率和实际账户扣费。如果 Flash 达到要求，可用于广泛探索；只有在实测质量足以证明成本合理时，才将入围方案升级为 Photon。

请勿将 Photon 任务发送到 /generations/video、Ray 模型、Luma Agents、消费者会话端点或 OpenAI 兼容网关。第一方 Photon 创建路由为：

POST https://api.lumalabs.ai/dream-machine/v1/generations/image
Authorization: Bearer $LUMAAI_API_KEY
Content-Type: application/json

通用管理路由为：

GET    /dream-machine/v1/generations/{id}
GET    /dream-machine/v1/generations?limit=100&offset=0
DELETE /dream-machine/v1/generations/{id}
GET    /dream-machine/v1/credits

官方 Python 包为 lumaai；其图像创建接口为 client.generations.image.create(...)，轮询则使用 client.generations.get(id=...)。完整示例采用直接 REST 调用，以便明确处理超时、重定向、字节、解码和持久化行为。

请求约定

图像指南、端点参考和第一方 OpenAPI记录了以下请求格式：

字段	生产环境中的说明
model	架构要求的字段；只能使用 photon-1 或 photon-flash-1。
prompt	自然语言图像描述。通用错误指南记录的长度为 3–5000 个字符；请明确写出关键文本和约束。
aspect_ratio	1:1、16:9、9:16、4:3、3:4、21:9 或 9:21；默认值为 16:9。
format	jpg 或 png；默认值为 jpg。请检查返回的字节，而不要仅凭此请求值判断格式。
callback_url	可选的 HTTPS 端点，用于接收重复发送的生成对象。文档未说明回调签名或自定义身份验证标头。
image_ref	{url, weight} 引用数组，用于提供概念／构图／变化指导。
style_ref	{url, weight} 风格引用数组。在契约确认支持更多数量之前，请使用一个。
character_ref	包含文档所述 identity0: {images: [...]} 的对象；指南支持最多四张同一身份的图像。
modify_image_ref	一个 {url, weight} 源图像，用于根据指令进行修改。
sync, sync_timeout	在 OpenAPI／参考文档中列出，默认值分别为 false 和 60；但图像指南称仅支持轮询来获取结果，与此相矛盾。

文档未说明 Photon 支持 seed、负面提示词字段、蒙版、引导比例、步数、base64／数据 URI 输入、多部分上传、输出 n、任意像素尺寸或分辨率选择器。一次创建调用会生成一个包含单个 assets.image 的生成对象。生成多个变体时，请分别预算并发起请求。

有日期的文档冲突

当第一方来源存在分歧时，应采取保守做法：

• 图像指南称 image_ref 最多支持四张图像。Luma 官方的 luma-api-mcp README 称支持八张。在实时参考文档或 Luma 支持团队解决这一分歧之前，生产请求上限设为四张。
• REST/OpenAPI 参考文档列出了 sync 和 sync_timeout；Python 和 JavaScript 图像指南称目前仅支持通过轮询获取图像。请保持 sync: false，并使用有界轮询。
• Photon 产品页面公布的 Photon Flash 价格为每张 2MP/1080p 图像 $0.004，而较早的 v1.2 API 更新日志公布的价格为 $0.002。两处公布的 Photon 价格均为 $0.015。请按较高的公开价格为 Flash 做预算，并核对账户扣费；不要将任一价格称为有保证的当前费率。
• Photon 公开参考文档未公布输出 URL 的具体有效期、固定的 API 输入/输出保留期限、区域端点、数据驻留选项、幂等键或每次生成的成本字段。不要臆造这些信息。

参考与修改策略

Photon 目前仅接受 Luma 服务器能够抓取的 URL 形式的参考输入。文档未说明支持带身份验证的抓取请求头。请将获准的素材托管在自有 HTTPS 来源或短期有效的签名 CDN URL 上，仅允许使用应用所控制的确切来源主机，并确保每个 URL 在任务进入终止状态前始终有效。切勿仅为图方便而使用工作站路径、file: URL、专用网络 URL、承载令牌凭据、永久公开的存储桶或第三方上传服务。

为每个来源选择一种语义控制：

• image_ref：将概念、构图、姿势、物体或宽泛的视觉线索带入新图像。在提示词中标明每张参考图的作用。
• style_ref：迁移渲染风格、调色板、纹理或光照，不要声称能逐像素保留。
• character_ref：使用同一位获授权的人物或角色的 1 至 4 张清晰图像，构建文档所述的 identity0。不同角度的图像应在关键特征上保持一致。
• modify_image_ref：以一张图像为基础，并描述一项局部更改。文档说明，较高的权重会使结果更接近输入图像，多样性也更低。文档仅针对难以实现的颜色更改建议使用 0.0–0.1；并未公布通用数值范围。

不要承诺精确保留身份、徽标、字体排版、布局或未更改区域。参考图生成属于条件控制，而非确定性的合成器。编辑时，先说明要更改的元素，再列出少量需要保留的不变量。请将每个输出都与源图像进行全分辨率对比。

请求正文示例，并非必须遵循的公式：

{
  "model": "photon-flash-1",
  "prompt": "Use the reference only for the cobalt ceramic shape and three-quarter camera angle. Place that object on warm white paper under soft overhead light; editorial product photograph, open negative space on the left.",
  "aspect_ratio": "16:9",
  "format": "png",
  "image_ref": [
    {"url": "https://assets.example.com/signed/product-reference.png", "weight": 0.8}
  ],
  "sync": false
}

{
  "model": "photon-1",
  "prompt": "Replace only the white daisies with yellow sunflowers. Preserve the vase geometry, tabletop, camera angle, crop, lighting direction, shadows, and background.",
  "aspect_ratio": "4:3",
  "format": "png",
  "modify_image_ref": {
    "url": "https://assets.example.com/signed/approved-vase.png",
    "weight": 1.0
  },
  "sync": false
}

上传前请计算参考文件字节的哈希值，并记录权利/同意情况、用途、所有者、源 MIME 类型/尺寸、校验和、签名 URL 过期时间，以及在提示词中指定的作用。日志中仅存储签名 URL 的哈希值或查询参数已删除的形式。

异步生命周期

默认采用文档说明的创建并轮询流程：

1. 验证需求简述、政策、来源权利、模型、请求数量、保守的单价、月度上限和当前余额。
2. 在创建请求前持久化客户端任务键和请求哈希。这是应用级账本，不是服务提供商的幂等机制。
3. POST /generations/image。201 响应会返回一个带有 UUID id 的生成对象。
4. 在设定的截止时间内轮询 GET /generations/{id}，采用有上限的抖动退避策略。OpenAPI 列出了 queued、dreaming、completed 和 failed 状态。
5. 状态为 failed 时，记录 failure_reason；不要假设会退款。状态为 completed 时，必须检查 assets.image 是否存在，并立即将其保存到受控存储中。
6. 记录终态对象、未受并发账户使用影响时的余额变化，以及最终制品清单。

如果在获取有效生成 ID 之前创建请求超时或连接中断，应视为状态不明：付费任务可能已经存在。公开契约未提供幂等键，列表分页也不支持按客户端引用筛选。不要盲目重放 POST 请求。核对账户使用情况，列出最近的生成记录，谨慎地匹配请求、模型和时间；如果唯一性至关重要，则要求操作员作出决定。获取 ID 后，只重试安全的 GET 轮询和制品下载。

回调是可选提示，不是事实来源。指南说明 Luma 会 POST dreaming、completed 和 failed 更新，可能重复调用，超时时间为 5 秒；如果响应不是 200，则会以 100 毫秒的间隔最多重试 3 次。由于没有记录一方签名方案，应在不含查询参数的回调路径中放置高熵、不透明的令牌，对端点进行速率限制，限制并解析 JSON，接受重复或乱序事件，验证 UUID 和状态，然后通过带 bearer 认证的 GET 重新获取生成对象，再采取操作。

完整示例：优先试运行的 REST 任务

这是一个示例，并非强制公式。请先安装 Pillow（python -m pip install Pillow）。此示例支持文本、图像引用、风格引用、角色引用和修改模式；默认执行脱敏试运行；只有明确选择付费发送后才会发送；绝不会将 Luma bearer 令牌转发到来源或输出主机；轮询时限明确；验证每一跳输出重定向；限制字节数和像素数；完整解码；以原子方式持久化；并记录清单。

环境控制项：

• LUMA_MODE：text（默认）、image-reference、style-reference、character-reference 或 modify。
• LUMA_REFERENCE_URLS_JSON：非文本模式下，自有 HTTPS URL 的 JSON 数组。
• LUMA_INPUT_HOSTS：这些输入使用的逗号分隔精确主机名允许列表。
• LUMA_ASSET_HOSTS：精确的输出/重定向主机名。默认使用 Luma 公开示例中的主机 storage.cdn-luma.com；如果生产环境返回其他主机，则应拒绝，并在添加前进行验证。
• LUMA_REFERENCE_WEIGHT：示例权重，默认为 0.8。
• LUMA_MODEL：两个精确 Photon ID 中的一个。
• SEND_LUMA_REQUEST=1：启用 API 和资源网络调用的必需设置。

import hashlib
import ipaddress
import json
import math
import os
import pathlib
import random
import socket
import tempfile
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
import urllib.error
import urllib.parse
import urllib.request
import uuid

from PIL import Image, UnidentifiedImageError

API_ROOT = "https://api.lumalabs.ai/dream-machine/v1"
CREATE_URL = f"{API_ROOT}/generations/image"
OUT = pathlib.Path("luma-photon-output")
MAX_JSON_BYTES = 2 * 1024 * 1024
MAX_IMAGE_BYTES = 30 * 1024 * 1024
MAX_IMAGE_PIXELS = 20_000_000
MAX_POLL_SECONDS = 600
Image.MAX_IMAGE_PIXELS = MAX_IMAGE_PIXELS


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


api_opener = urllib.request.build_opener(NoRedirect())


class LumaHTTPError(RuntimeError):
    def __init__(self, code, detail, retry_after=None):
        super().__init__(f"Luma HTTP {code}: {detail}")
        self.code = code
        self.retry_after = retry_after


def host_set(name, defaults=()):
    values = {value.strip().lower() for value in os.getenv(name, "").split(",") if value.strip()}
    return values | set(defaults)


def redact_url(url):
    parsed = urllib.parse.urlsplit(url)
    return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))


def validate_url_shape(url, allowed_hosts):
    parsed = urllib.parse.urlsplit(url)
    if (
        parsed.scheme != "https"
        or parsed.hostname is None
        or parsed.hostname.lower() not in allowed_hosts
        or parsed.username
        or parsed.password
        or parsed.fragment
        or parsed.port not in (None, 443)
    ):
        raise ValueError("URL must be HTTPS on an approved exact host without credentials, fragment, or non-443 port")


def validate_public_dns(url, allowed_hosts):
    validate_url_shape(url, allowed_hosts)
    parsed = urllib.parse.urlsplit(url)
    addresses = {info[4][0] for info in socket.getaddrinfo(parsed.hostname, 443)}
    if not addresses or any(not ipaddress.ip_address(value).is_global for value in addresses):
        raise ValueError("URL host did not resolve exclusively to public addresses")


def bounded_json(request, expected_status, timeout):
    try:
        # Never follow an API redirect with the bearer token attached. The fixed
        # api.lumalabs.ai origin must answer directly or the request fails closed.
        with api_opener.open(request, timeout=timeout) as response:
            if response.status != expected_status:
                raise RuntimeError(f"Unexpected HTTP status {response.status}")
            raw = response.read(MAX_JSON_BYTES + 1)
    except urllib.error.HTTPError as exc:
        try:
            retry_after = exc.headers.get("Retry-After")
            detail = exc.read(64 * 1024).decode("utf-8", "replace")
        finally:
            exc.close()
        raise LumaHTTPError(exc.code, detail, retry_after) from exc
    if len(raw) > MAX_JSON_BYTES:
        raise RuntimeError("Luma JSON response exceeded configured cap")
    try:
        value = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise RuntimeError("Luma response was not valid bounded JSON") from exc
    if not isinstance(value, dict):
        raise RuntimeError("Luma response was not a JSON object")
    return value


def api_request(method, path, key, payload=None, expected_status=200, timeout=90):
    body = None if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8")
    headers = {"Authorization": f"Bearer {key}", "Accept": "application/json"}
    if body is not None:
        headers["Content-Type"] = "application/json"
    request = urllib.request.Request(
        f"{API_ROOT}{path}", data=body, method=method, headers=headers
    )
    return bounded_json(request, expected_status, timeout)


def safe_credit_balance(key):
    try:
        value = api_request("GET", "/credits", key)
        return float(value["credit_balance"])
    except (KeyError, TypeError, ValueError, RuntimeError, urllib.error.URLError) as exc:
        print({"credit_balance_warning": str(exc)})
        return None


def atomic_write(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=".luma-manifest-")
    try:
        with os.fdopen(fd, "wb") as output:
            fd = None
            output.write(data)
            output.flush()
            os.fsync(output.fileno())
        os.replace(temporary, path)
    except Exception:
        if fd is not None:
            os.close(fd)
        try:
            os.unlink(temporary)
        except FileNotFoundError:
            pass
        raise


def read_local_json(path):
    if path.stat().st_size > MAX_JSON_BYTES:
        raise ValueError("Local ledger exceeded JSON cap")
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError("Local ledger was not a JSON object")
    return value


def retry_after_seconds(value):
    if not value:
        return 0.0
    try:
        seconds = float(value)
    except ValueError:
        try:
            moment = parsedate_to_datetime(value)
            if moment.tzinfo is None:
                moment = moment.replace(tzinfo=timezone.utc)
            seconds = (moment - datetime.now(timezone.utc)).total_seconds()
        except (TypeError, ValueError, OverflowError):
            return 0.0
    return seconds if math.isfinite(seconds) and seconds > 0 else 0.0


def get_generation_with_retry(generation_id, key, deadline, initial_wait=0.0):
    wait = max(0.0, initial_wait)
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError("generation GET deadline exceeded")
        if wait:
            sleep_for = wait + random.random()
            if sleep_for >= remaining:
                raise TimeoutError("required generation GET delay exceeds remaining deadline")
            time.sleep(sleep_for)
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise TimeoutError("generation GET deadline exceeded after wait")
        try:
            return api_request("GET", f"/generations/{generation_id}", key,
                               timeout=max(0.001, min(90.0, remaining)))
        except LumaHTTPError as exc:
            if exc.code not in {429, 500, 502, 503, 504}:
                raise
            provider_wait = retry_after_seconds(exc.retry_after)
            wait = max(min(max(wait, 2.0) * 1.6, 30.0), provider_wait, 2.0)
        except (urllib.error.URLError, TimeoutError):
            wait = min(max(wait, 2.0) * 1.6, 30.0)


def download_image(url, allowed_hosts, directory, requested_format):
    opener = urllib.request.build_opener(NoRedirect())
    current = url
    response = None
    for _ in range(6):
        validate_public_dns(current, allowed_hosts)
        # Deliberately no LUMAAI_API_KEY header on any asset hop.
        request = urllib.request.Request(
            current, headers={"Accept": "image/jpeg,image/png"}
        )
        try:
            response = opener.open(request, timeout=90)
            break
        except urllib.error.HTTPError as exc:
            if exc.code in {301, 302, 303, 307, 308}:
                location = exc.headers.get("Location")
                exc.close()
                if not location:
                    raise ValueError("Artifact redirect omitted Location")
                current = urllib.parse.urljoin(current, location)
                continue
            try:
                detail = exc.read(64 * 1024).decode("utf-8", "replace")
            finally:
                exc.close()
            raise RuntimeError(f"Artifact HTTP {exc.code}: {detail}") from exc
    else:
        raise ValueError("Too many artifact redirects")

    directory.mkdir(parents=True, exist_ok=True)
    fd = None
    temporary = None
    total = 0
    prefix = b""
    hasher = hashlib.sha256()
    try:
        with response:
            mime = response.headers.get_content_type()
            if mime not in {"image/jpeg", "image/png"}:
                raise ValueError(f"Unexpected artifact MIME: {mime}")
            declared = response.headers.get("Content-Length")
            if declared:
                try:
                    if int(declared) > MAX_IMAGE_BYTES:
                        raise ValueError("Artifact exceeds declared-size cap")
                except ValueError as exc:
                    raise ValueError("Invalid or excessive Content-Length") from exc
            fd, temporary = tempfile.mkstemp(dir=directory, prefix=".luma-image-")
            with os.fdopen(fd, "wb") as output:
                fd = None
                while True:
                    chunk = response.read(64 * 1024)
                    if not chunk:
                        break
                    total += len(chunk)
                    if total > MAX_IMAGE_BYTES:
                        raise ValueError("Artifact exceeded streaming byte cap")
                    prefix += chunk[: max(0, 12 - len(prefix))]
                    hasher.update(chunk)
                    output.write(chunk)
                output.flush()
                os.fsync(output.fileno())

        if prefix.startswith(b"\xff\xd8\xff"):
            expected_mime, expected_format, suffix = "image/jpeg", "JPEG", ".jpg"
        elif prefix.startswith(b"\x89PNG\r\n\x1a\n"):
            expected_mime, expected_format, suffix = "image/png", "PNG", ".png"
        else:
            raise ValueError("Artifact signature is neither JPEG nor PNG")
        if mime != expected_mime:
            raise ValueError("Artifact MIME/signature mismatch")
        requested_mime = {"jpg": "image/jpeg", "png": "image/png"}[requested_format]
        if mime != requested_mime:
            raise ValueError("Artifact MIME did not match the requested output format")

        try:
            with Image.open(temporary) as probe:
                dimensions = probe.size
                if dimensions[0] * dimensions[1] > MAX_IMAGE_PIXELS:
                    raise ValueError(f"Artifact exceeds pixel cap: {dimensions}")
                decoded_format = probe.format
                probe.verify()
            with Image.open(temporary) as probe:
                probe.load()
        except (UnidentifiedImageError, OSError, Image.DecompressionBombError) as exc:
            raise ValueError("Artifact failed bounded full decode") from exc
        if decoded_format != expected_format:
            raise ValueError("Artifact decoder/signature mismatch")

        destination = directory / f"image{suffix}"
        os.replace(temporary, destination)
        temporary = None
        return {
            "path": str(destination),
            "bytes": total,
            "mime_type": mime,
            "dimensions": list(dimensions),
            "sha256": hasher.hexdigest(),
        }
    except Exception:
        if fd is not None:
            os.close(fd)
        try:
            if temporary is not None:
                os.unlink(temporary)
        except FileNotFoundError:
            pass
        raise


model = os.getenv("LUMA_MODEL", "photon-flash-1")
mode = os.getenv("LUMA_MODE", "text")
if model not in {"photon-1", "photon-flash-1"}:
    raise SystemExit("LUMA_MODEL must be photon-1 or photon-flash-1")
if mode not in {"text", "image-reference", "style-reference", "character-reference", "modify"}:
    raise SystemExit("Unsupported LUMA_MODE")

prompt = (
    "16:9 editorial product photograph of a cobalt ceramic tea set on warm white "
    "paper at dawn, tabletop camera height, soft overhead light, restrained cool "
    "shadows, clean negative space on the left, realistic glaze and contact shadows. "
    "No added lettering or logos."
)
payload = {
    "model": model,
    "prompt": prompt,
    "aspect_ratio": "16:9",
    "format": "png",
    "sync": False,
}

reference_urls = []
if mode != "text":
    try:
        reference_urls = json.loads(os.getenv("LUMA_REFERENCE_URLS_JSON", "[]"))
    except json.JSONDecodeError as exc:
        raise SystemExit("LUMA_REFERENCE_URLS_JSON must be a JSON array") from exc
    if not isinstance(reference_urls, list) or not all(isinstance(url, str) for url in reference_urls):
        raise SystemExit("LUMA_REFERENCE_URLS_JSON must contain only URL strings")
    expected = (1, 4) if mode in {"image-reference", "character-reference"} else (1, 1)
    if not expected[0] <= len(reference_urls) <= expected[1]:
        raise SystemExit(f"{mode} requires {expected[0]}..{expected[1]} reference URLs")
    input_hosts = host_set("LUMA_INPUT_HOSTS")
    if not input_hosts:
        raise SystemExit("Set LUMA_INPUT_HOSTS to the exact owned source hosts")
    for url in reference_urls:
        validate_url_shape(url, input_hosts)
    weight = float(os.getenv("LUMA_REFERENCE_WEIGHT", "0.8"))
    if not math.isfinite(weight):
        raise SystemExit("LUMA_REFERENCE_WEIGHT must be finite")
    if mode == "image-reference":
        payload["image_ref"] = [{"url": url, "weight": weight} for url in reference_urls]
    elif mode == "style-reference":
        payload["style_ref"] = [{"url": reference_urls[0], "weight": weight}]
    elif mode == "character-reference":
        payload["character_ref"] = {"identity0": {"images": reference_urls}}
    else:
        payload["modify_image_ref"] = {"url": reference_urls[0], "weight": weight}

body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
request_sha256 = hashlib.sha256(body).hexdigest()
client_job_key = os.getenv("LUMA_CLIENT_JOB_KEY", f"local-{request_sha256[:20]}")
display_payload = json.loads(json.dumps(payload))
for key in ("image_ref", "style_ref"):
    for value in display_payload.get(key, []):
        value["url"] = redact_url(value["url"])
if display_payload.get("character_ref"):
    display_payload["character_ref"]["identity0"]["images"] = [
        redact_url(url) for url in reference_urls
    ]
if display_payload.get("modify_image_ref"):
    display_payload["modify_image_ref"]["url"] = redact_url(reference_urls[0])

public_price_snapshot = {"photon-1": 0.015, "photon-flash-1": 0.004}[model]
print(json.dumps({
    "dry_run": os.getenv("SEND_LUMA_REQUEST") != "1",
    "client_job_key": client_job_key,
    "request": display_payload,
    "request_sha256": request_sha256,
    "conservative_public_unit_price_usd_snapshot_2026_07_09": public_price_snapshot,
    "price_requires_dashboard_verification": True,
}, indent=2))

if os.getenv("SEND_LUMA_REQUEST") != "1":
    raise SystemExit("Dry run only; set SEND_LUMA_REQUEST=1 after approving the paid call")
key = os.getenv("LUMAAI_API_KEY")
if not key:
    raise SystemExit("Set LUMAAI_API_KEY before the paid call")

job_token = hashlib.sha256(client_job_key.encode("utf-8")).hexdigest()[:20]
job_dir = OUT / job_token
ledger_path = job_dir / "job.json"
manifest_path = job_dir / "manifest.json"


def save_job(state, generation_id=None, failure_reason=None, note=None):
    value = {
        "client_job_key": client_job_key,
        "request_sha256": request_sha256,
        "model": model,
        "state": state,
        "generation_id": generation_id,
        "failure_reason": failure_reason,
        "note": note,
        "updated_unix": time.time(),
        "reference_url_sha256": [
            hashlib.sha256(url.encode("utf-8")).hexdigest() for url in reference_urls
        ],
    }
    atomic_write(ledger_path, json.dumps(value, indent=2).encode("utf-8"))


def claim_new_job():
    value = {
        "client_job_key": client_job_key,
        "request_sha256": request_sha256,
        "model": model,
        "state": "submitting",
        "generation_id": None,
        "failure_reason": None,
        "note": "Exclusive local claim created before any paid call",
        "updated_unix": time.time(),
        "reference_url_sha256": [
            hashlib.sha256(url.encode("utf-8")).hexdigest() for url in reference_urls
        ],
    }
    job_dir.mkdir(parents=True, exist_ok=True)
    fd = os.open(ledger_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as output:
            fd = None
            json.dump(value, output, indent=2)
            output.flush(); os.fsync(output.fileno())
    finally:
        if fd is not None: os.close(fd)

if reference_urls:
    for url in reference_urls:
        validate_public_dns(url, input_hosts)

if manifest_path.exists():
    print(json.dumps(read_local_json(manifest_path), indent=2))
    raise SystemExit("This client job is already complete; choose a new LUMA_CLIENT_JOB_KEY for a new paid job")

credit_before = None
deadline = time.monotonic() + MAX_POLL_SECONDS
if ledger_path.exists():
    existing = read_local_json(ledger_path)
    if existing.get("client_job_key") != client_job_key or existing.get("request_sha256") != request_sha256:
        raise SystemExit("Existing ledger does not match this client job/request")
    generation_id = existing.get("generation_id")
    if not generation_id:
        raise SystemExit(
            "Existing create has no confirmed generation ID; treat it as ambiguous and reconcile. "
            "Use a new explicit LUMA_CLIENT_JOB_KEY only after an operator approves a new paid job."
        )
    try: uuid.UUID(str(generation_id))
    except (ValueError, AttributeError) as exc:
        raise SystemExit("Existing ledger contains an invalid generation UUID") from exc
    try:
        generation = get_generation_with_retry(generation_id, key, deadline)
    except TimeoutError as exc:
        raise SystemExit("Resume GET deadline exceeded; keep the ID and reconcile later") from exc
else:
    try:
        claim_new_job()
    except FileExistsError as exc:
        raise SystemExit(
            "Another worker claimed this client job key; do not submit. Retry later to resume its ledger."
        ) from exc
    credit_before = safe_credit_balance(key)
    try:
        generation = api_request(
            "POST", "/generations/image", key, payload, expected_status=201, timeout=180
        )
    except (urllib.error.URLError, TimeoutError) as exc:
        save_job("ambiguous_create", note="Do not replay; reconcile recent generations and credits")
        raise SystemExit(
            "Ambiguous create outcome: do not replay; reconcile recent generations and credits"
        ) from exc
    generation_id = generation.get("id")
    try:
        uuid.UUID(str(generation_id))
    except (ValueError, AttributeError) as exc:
        save_job("invalid_create_response", note="Create returned no valid generation UUID")
        raise SystemExit("Create response lacked a valid generation UUID") from exc
    save_job("accepted", generation_id)

delay = 2.0
while True:
    state = generation.get("state")
    save_job(state, generation_id, generation.get("failure_reason"))
    if state == "completed":
        break
    if state == "failed":
        raise SystemExit(f"Luma generation failed: {generation.get('failure_reason')}")
    if state not in {"queued", "dreaming"}:
        raise SystemExit(f"Unexpected generation state: {state!r}")
    remaining = deadline - time.monotonic()
    if remaining <= 0:
        save_job(state, generation_id, note="Polling deadline exceeded; reconcile later")
        raise SystemExit("Polling deadline exceeded; keep the ID and reconcile later")
    try:
        generation = get_generation_with_retry(generation_id, key, deadline, initial_wait=delay)
        delay = min(delay * 1.6, 10.0)
    except TimeoutError as exc:
        save_job(state, generation_id, note="Polling/retry deadline exceeded; reconcile later")
        raise SystemExit("Polling deadline exceeded; keep the ID and reconcile later") from exc

asset_url = (generation.get("assets") or {}).get("image")
if not asset_url:
    raise SystemExit("Completed generation has no assets.image URL")
asset_hosts = host_set("LUMA_ASSET_HOSTS", {"storage.cdn-luma.com"})
artifact = download_image(asset_url, asset_hosts, job_dir, payload["format"])
credit_after = safe_credit_balance(key)
delta_cents = None
if credit_before is not None and credit_after is not None:
    delta_cents = credit_before - credit_after

manifest = {
    "generation_id": generation_id,
    "state": generation.get("state"),
    "model_requested": model,
    "model_reported": generation.get("model"),
    "client_job_key": client_job_key,
    "request_sha256": request_sha256,
    "prompt_sha256": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
    "reference_url_sha256": [hashlib.sha256(url.encode("utf-8")).hexdigest() for url in reference_urls],
    "public_unit_price_snapshot_usd": public_price_snapshot,
    "credit_balance_before_cents": credit_before,
    "credit_balance_after_cents": credit_after,
    "observed_balance_delta_cents": delta_cents,
    "artifact": artifact,
    "review_status": "pending",
    "downstream_ai_notice": "required for every API user",
    "public_ai_disclosure_review": "required when output identifies or resembles a person",
}
atomic_write(manifest_path, json.dumps(manifest, indent=2).encode("utf-8"))
print(json.dumps(manifest, indent=2))

只有在测量时间段内没有其他任务使用该账户时，余额变化才可归因于此任务。将其视为对账依据，而不是通用费率。如果出现新的制品主机，不要将验证放宽为“任何 HTTPS URL”；先验证其所有权和用途，再将精确主机名添加到配置中。

提示、迭代与审核

Photon 的官方资料建议使用自然、详尽的语言，并称该系列能够遵循提示词、进行迭代编辑、处理文本和参考图。这些属于服务提供商的说法。将其转化为受控的生产流程：

1. 说明交付物和受众：主视觉图、包装概念、编辑版面、海报、分镜或角色关键美术图。
2. 描述主体和动作，然后说明空间布局和相机／视角。
3. 指定光线、调色板、材质、环境和一个统一的风格锚点。
4. 准确引用所需的可见文本，并说明其位置、层级和大小写。仍须进行 OCR 和人工校对。
5. 只列出有意义的排除项。Photon 没有文档说明的负面提示词字段；排除项仍须用普通提示词语言表达。
6. 对于参考图，将每个输入映射到一个用途，并重复说明不可变的身份／产品／品牌特征。
7. 生成数量有限的探索集。根据需求简报进行选择，而非只看新颖性。
8. 每轮只做一项语义变更。以已批准的源图为锚点，而不是反复编辑逐渐偏离的衍生图。
9. 当身份、几何结构、排版或光线逐渐偏移时，从最近一份已批准的成品重新开始。

生产经验法则：使用简洁的结构化提示词，例如“交付物 → 主体／动作 → 构图／相机 → 光线／调色板 → 材质／风格 → 准确文本 → 受保护的不变量”。这不是服务提供商规定的语法。

根据源图和需求简报，审核解码后的全分辨率像素：

• **需求简报：**交付物、宽高比、安全区域、主体数量、裁切、层级、留白。
• **参考图／编辑：**身份、产品几何结构、姿势、标签／徽标、背景、光线、阴影以及每个受保护区域。
• **文本：**先进行 OCR，再由人工检查拼写、标点、大小写、换行、字距和易读性。
• **视觉：**面部／手部、重复结构、反射、遮挡、透视、材质连贯性、边缘光晕、噪点和压缩。
• **技术：**MIME／魔数／解码器是否一致、尺寸、像素面积、色彩／配置文件元数据、文件大小、哈希值和存储时限。
• **权利／安全：**源素材授权、肖像同意、商标使用场景、审核、披露和来源信息保留。

为这两个模型维护回归测试集，涵盖每种使用中的宽高比、JPG／PNG、纯文本、一张／四张参考图、风格、一张／四张角色视图、低／高修改权重、困难的颜色编辑、审核失败、429、创建超时、轮询超时、重定向、损坏的 MIME、超大文件以及过期的输入／输出 URL。

故障、速率控制与成本

错误指南列出了提交前的验证失败，以及提交后的 failure_reason 值，包括审核、输入无法访问、提示词处理、调度、任务和回调错误。在修正之前，应将提示词／架构／身份验证／审核／源素材访问问题视为不可重试。对于 GET 超时、429 和暂时性 5xx，应视为可重试，并采用有上限的指数退避、抖动、存在时使用 Retry-After、设置总截止时间以及并发准入控制。

速率限制页面公布，Build 层级可并发生成 40 个 Photon／Photon Flash 任务，每分钟最多发起 80 个创建请求，月度使用限额为 $5,000；页面仅称 Scale 层级的限制更高。这些是上限，不是目标。按用户、项目和日期设置更低的应用级限额，并在上线前重新读取账户的实时限额。

额度端点以美分为单位返回 credit_balance。生成架构未公开单个任务的成本。提交前进行估算，为隔离的金丝雀测试记录前后余额，对账发票／仪表板，并在出现差异时发出警报。截至 2026-07-09 查到的最新 Photon 专属公开价格存在上述冲突；绝不要掩盖这种不确定性。将 Luma 应用额度与 API 额度分开。

资产、凭据和来源安全

• 将 LUMAAI_API_KEY 保留在服务器端，仅限此集成使用，不得放入 URL、提示词、回调、日志、浏览器或输出清单中。API 条款要求在发现凭据遭泄露后 24 小时内通知 Luma。
• 仅向 api.lumalabs.ai 发送 Bearer 身份验证信息。来源图像和生成的 CDN 资产不会收到该信息。
• 要求来源主机归你所有且使用 HTTPS、DNS 解析到公共网络、签名 URL 的有效期尽可能短，且 URL 中不得嵌入用户名、密码或片段。避免提交无法安全地交由服务提供商获取的敏感输入。
• 将 assets.image 视为传输位置，而非持久存储。公开文档未说明具体有效期。应立即下载、手动跟随重定向并验证每一跳、限制声明字节数和实际传输字节数、要求 MIME 类型、文件魔数与解码器相符、在完整解码前检查尺寸、按应用策略扫描，并保留原始字节、计算哈希后原子式地提升文件。
• 不要记录签名查询参数。进入终态后撤销或使来源 URL 过期；若生成的 URL 由你控制，则在获取资源后撤销或使其过期。
• DELETE /generations/{id} 会删除生成资源，但文档没有说明此操作会撤销资产 URL 或满足法律上的删除要求。不要声称它具有这两种效果。
• 保留服务提供商添加的水印、内容凭据或来源信息元数据。条款保留了 Luma 嵌入这些信息的权利，并禁止移除；Photon API 不保证任何特定输出都包含 C2PA 或可见标记。

数据、权利与安全门槛

请阅读实际适用于该账户的协议。截至核验日期：

• API Terms规定，Luma 不会使用 API Input 或 API Output 来训练、微调或开发其 AI/ML 模型。此 API 专属条款优先于与之冲突的一般 Terms 条款。但这并不承诺零保留、零人工／自动审核、零 Usage/Aggregated Data 或零处理。
• Terms of Service规定，在双方之间且在法律允许的最大范围内，客户拥有 Output，Luma 则转让其相关权益。这不代表客户获得第三方权利，不保证 Output 具有独特性、可受版权保护、不侵权或准确，也不因此自动授权商业用途。商业用途须有允许此类用途的有效付费订阅／Order。
• 公开文件未规定 API Input/Output 的固定保留期限，也未提供区域性／数据驻留端点。Privacy Policy采用基于目的和法律的保留标准；企业数据处理可能受 DPA 和 Order 管辖。如有需要，请获取关于保留、删除、子处理方、数据驻留、安全和事件响应的书面承诺。
• Terms 将 Prohibited Data 定义为包括 GDPR 第 9 条特别类别数据、PHI、PCI 数据、受 COPPA/GLBA 监管的数据、政府身份证件以及类似受监管数据。Luma 表示其 Services 并非为符合 HIPAA 而设计，且其并非 Business Associate。请勿提交此类数据。
• 告知每位下游 API 用户，生成的 Output 为 AI 生成。另请为每个来源、面孔、肖像、角色、标识、商标、受版权保护作品和地点取得有记录的权利与同意；如果 Output 可识别或类似某个人，Terms 要求公开披露其为 AI 生成。Terms 也禁止将 Output 虚假描述为人工生成。
• Terms 包含一项影响上传照片中人物的特定许可条款。涉及真实人物的参考素材项目应交由法律顾问审查，不要假设拥有普通素材的所有权就能解决所有肖像权问题。
• Content Moderation Policy 和 Terms 禁止的类别包括色情／NSFW 内容、有害或非法内容、仇恨／歧视、侵权、欺骗性深度伪造／冒充、政治操纵／虚假信息、欺诈、儿童色情化及其他滥用行为。请在应用层进行预检并安排人工审核；服务提供方的审核并不等于权利许可。
• API 客户必须要求下游用户遵守书面限制，且其保护力度至少与 Luma 的 AUP 相当；必须运行投诉处理流程，并遵守 Luma 的暂停／终止请求。API Terms 还禁止使用 API 或 Output 构建用于 AI 训练、微调、评估或其他模型开发的数据集。

请记录一份可审计清单，其中包含简介 ID、请求／提示词哈希、模型、参考素材哈希及权利／同意记录、生成 ID／状态／失败信息、时间戳、输出 MIME／尺寸／哈希、价格来源与估算、观察到的积分变化、审核／审查人员结果、AI 披露决定、来源状态、受控存储路径、保留／删除计划及发布审批。

一方来源

已于 2026-07-09 核实：

• Photon REST 图像指南
• 图像生成 API 参考
• 官方 OpenAPI 代码库
• 官方 Python SDK 图像指南
• 官方 JavaScript SDK 图像指南
• 生成、列表、删除和积分参考
• 错误和速率限制
• Photon 产品页面和Photon API 更新日志
• Luma 应用与 API 积分的区分
• 官方 Luma API MCP 代码库
• 当前 Luma 品牌/API 说明
• API 条款、服务条款、隐私政策、内容审核政策、企业 DPA和子处理方