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} 配列です。より多くの参照を指定できることが契約で確認されるまでは、1つだけ使用してください。
character_ref	ドキュメントに記載されている identity0: {images: [...]} を持つオブジェクトです。ガイドでは、同じ人物の画像を最大4枚まで指定できます。
modify_image_ref	指示に基づいて変更する元画像を指定する {url, weight} オブジェクトです。
sync, sync_timeout	OpenAPI／リファレンスにはデフォルト値 false と 60 で記載されていますが、ポーリングのみがサポートされる取得方法であるという画像ガイドの説明とは矛盾しています。

Photon には、ドキュメントに記載されたシード、ネガティブプロンプト用フィールド、マスク、ガイダンススケール、ステップ数、base64／データ URI 入力、マルチパートアップロード、出力 n、任意のピクセルサイズ、解像度セレクターはありません。1回の作成呼び出しで得られる生成オブジェクトは1つで、その assets.image は単一です。バリエーションを作成する場合は、別途予算を確保したリクエストを送信してください。

日付のあるドキュメント間の矛盾

一次情報源の内容が一致しない場合は、保守的に判断してください。

• 画像ガイドでは、image_ref が最大4枚の画像をサポートすると記載されています。Luma の公式 luma-api-mcp README には8枚と記載されています。ライブリファレンスまたは Luma のサポートによって矛盾が解消されるまでは、本番リクエストを4枚までに制限してください。
• REST/OpenAPI リファレンスには sync と sync_timeout が記載されていますが、Python と JavaScript の画像ガイドでは、現在、画像を取得する方法としてポーリングのみがサポートされていると記載されています。sync: false を維持し、上限を設けたポーリングを使用してください。
• Photon の製品ページでは、Photon Flash の料金として2MP/1080pの画像1枚あたり**$0.004が公開されていますが、以前の v1.2 API changelog では$0.002と記載されています。どちらも Photon の料金として$0.015**を公開しています。Flash については、公開されている高い方の料金を予算に見込み、アカウントへの請求額と照合してください。いずれの料金も、現在の確定料金として提示しないでください。
• 公開されている Photon リファレンスでは、出力 URL の有効期間、API の入力／出力の固定保持期間、リージョン別エンドポイント、データレジデンシーの選択肢、冪等性キー、または生成ごとのコストフィールドについて、数値は公開されていません。これらを作り上げないでください。

参照画像と変更の方針

現在、Photon が参照入力として受け付けるのは、Luma のサーバーが取得できる URL のみです。認証付き取得ヘッダーについては文書化されていません。承認済みのアセットは、自社が管理する HTTPS オリジンまたは有効期間の短い署名付き CDN URL にホストしてください。アプリケーションが管理する正確な送信元ホストのみを許可し、ジョブが終端状態に達するまで各 URL を有効に保ってください。利便性だけを理由に、ワークステーションのパス、file: URL、プライベートネットワーク URL、ベアラー認証情報、恒久的なパブリックバケット、またはサードパーティのアップロードサービスを使用しないでください。

各ソースにつき、次の意味的な制御を1つ選択してください。

• image_ref：コンセプト、構図、ポーズ、オブジェクト、または幅広い視覚的特徴を新しい画像に引き継ぎます。プロンプトで各参照画像の役割を明示してください。
• style_ref：ピクセル単位の保持を保証せずに、描画表現、パレット、質感、またはライティングを転写します。
• character_ref：同一の許可を得た人物またはキャラクターの、鮮明な画像を1～4枚使って、文書化されている identity0 を構築します。異なる角度の画像でも、特徴を定義する要素が一致するようにしてください。
• modify_image_ref：1枚の画像をもとに、局所的な変更を1つ説明します。重みを高くすると、入力に近くなり、多様性が低くなると記載されています。変更が難しい色の場合は特に 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 は dreaming、completed、failed の更新を POST し、繰り返し呼び出すことがあり、タイムアウトは 5 秒で、200 以外のレスポンスに対しては 100 ms 間隔で最大 3 回再試行します。ファーストパーティの署名方式は文書化されていないため、クエリを含まないコールバックパスに推測困難な不透明トークンを設定し、エンドポイントにレート制限を設け、JSON のサイズに上限を設けて解析します。重複イベントや順序が前後したイベントを受け入れ、UUID と状態を検証したうえで、ベアラー認証付き GET により生成を再取得してから処理します。

完全な例：ドライランを優先する REST ジョブ

これは一例であり、必須の方式ではありません。まず Pillow をインストールしてください（python -m pip install Pillow）。テキスト、画像参照、スタイル参照、キャラクター参照、変更の各モードに対応しています。デフォルトでは情報を伏せたドライランを実行し、有料リクエストの送信には明示的なオプトインを必須とします。Luma のベアラートークンをソースホストや出力ホストに転送することはありません。また、期限を設けてポーリングし、出力のリダイレクトを経由する各ホップを検証し、バイト数とピクセル数に上限を設け、完全にデコードしてからアトミックに保存し、マニフェストを記録します。

環境変数による制御：

• 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：2 つの正確な 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. 1 回のターンで変更する意味要素はひとつにします。承認済みの元画像を基準として再設定し、変化が重なった派生画像を繰り返し編集しないようにします。
9. アイデンティティ、形状、タイポグラフィ、照明のずれが蓄積した場合は、最後に承認された成果物からやり直します。

本番運用のヒューリスティック：次のような簡潔で構造化されたプロンプトを使います。「成果物 → 被写体／動作 → 構図／カメラ → 光／配色 → 素材／スタイル → 正確なテキスト → 保護する不変要素」。これはプロバイダー固有の構文ではありません。

デコードしたフル解像度のピクセルを、元画像およびブリーフと照合してレビューします。

• ブリーフ： 成果物、アスペクト比、安全領域、被写体の数、トリミング、階層、ネガティブスペース。
• 参照画像／編集： アイデンティティ、製品の形状、ポーズ、ラベル／ロゴ、背景、照明、影、保護対象の領域すべて。
• テキスト： OCR の後に、人がスペル、句読点、大文字・小文字、改行、カーニング、読みやすさを確認します。
• ビジュアル： 顔／手、反復構造、反射、オクルージョン、遠近感、素材の一貫性、エッジのハロー、ノイズ、圧縮。
• 技術： MIME／マジックナンバー／デコーダーの一致、寸法、ピクセル面積、色／プロファイルのメタデータ、ファイルサイズ、ハッシュ、保存期間。
• 権利／安全性： ソースの使用許可、肖像の同意、商標の文脈、モデレーション、開示、来歴の保持。

両方のモデル、使用するすべてのアスペクト比、JPG／PNG、テキストのみ、1 枚／4 枚の参照画像、スタイル、1 方向／4 方向のキャラクタービュー、低／高の modify weight、難しい色編集、モデレーション失敗、429、作成タイムアウト、ポーリングタイムアウト、リダイレクト、不正な MIME、サイズ超過ファイル、期限切れの入力／出力 URL を対象とする回帰テストセットを維持します。

失敗、レート制御、コスト

エラーガイドには、送信前の検証失敗と、送信後の failure_reason の値として、モデレーション、アクセスできない入力、プロンプト処理、ディスパッチ、ジョブ、コールバックのエラーが記載されています。プロンプト／スキーマ／認証／モデレーション／ソースアクセスに関する失敗は、修正するまで再試行不可として扱います。GET のタイムアウト、429、一時的な 5xx は、上限付きの指数バックオフ、ジッター、存在する場合の Retry-After、全体の期限、同時実行の受け入れ制御を適用し、再試行可能なものとして扱います。

レート制限のページでは、Build ティアについて、Photon／Photon Flash の同時生成数 40 件、作成リクエスト毎分 80 件、月間利用上限 5,000 ドルが公開されています。Scale については、より高いとだけ記載されています。これらは上限であり、目標値ではありません。ユーザー、プロジェクト、日ごとに、アプリケーション側のより低い上限を適用し、ローンチ前にアカウントの最新の制限を再確認します。

クレジットエンドポイントは、USD セント単位の credit_balance を返します。生成スキーマにはジョブごとのコストは公開されていません。送信前に見積もり、単独のカナリアテストでは実行前後の残高を記録し、請求書／ダッシュボードと照合して、不一致に対してアラートを設定します。2026-07-09 時点で確認できる最新の Photon 固有の公開価格には前述のとおり矛盾があるため、不確実性を隠してはいけません。Luma アプリのクレジットと API のクレジットは分けて扱います。

アーティファクト、認証情報、ソースのセキュリティ

• LUMAAI_API_KEY はサーバー側で管理し、インテグレーションのみにスコープを限定してください。URL、プロンプト、コールバック、ログ、ブラウザー、出力マニフェストには含めないでください。API 利用規約では、認証情報の侵害を発見してから 24 時間以内に Luma へ通知することが求められています。
• ベアラー認証は api.lumalabs.ai にのみ送信してください。ソース画像や生成された CDN アーティファクトには送信しないでください。
• 所有する HTTPS ソースホスト、パブリックネットワークの DNS、できるだけ有効期間の短い署名付き URL を必須とし、ユーザー名やパスワードの埋め込み、フラグメントは許可しないでください。プロバイダーが安全に取得できない機密性の高い入力は送信しないでください。
• assets.image は永続ストレージではなく、転送先として扱ってください。公開ドキュメントには数値で示された有効期間は記載されていません。すぐにダウンロードし、リダイレクトを手動で追跡して各ホップを検証し、申告サイズと実際のサイズの上限を設定してください。MIME タイプ、マジックナンバー、デコーダーの判定が一致することを必須とし、完全なデコード前に画像の寸法を制限し、アプリケーションのポリシーに従ってスキャンしてください。元のバイト列を保持し、ハッシュを計算してから、アトミックに昇格させてください。
• 署名付きクエリをログに記録しないでください。終端状態に達したらソース URL を無効化または期限切れにし、生成 URL を管理できる場合は取得後に無効化してください。
• DELETE /generations/{id} は生成リソースを削除しますが、アセット URL が無効になることや、法的な消去要件を満たすことはドキュメントに記載されていません。いずれについても、そのように断言しないでください。
• プロバイダーによって追加されたウォーターマーク、コンテンツ認証情報、来歴メタデータは保持してください。利用規約では、Luma がこれらを埋め込む権利を留保し、削除を禁止しています。Photon API では、特定の出力に C2PA や可視マークが含まれることは保証されていません。

データ、権利、安全性のゲート

アカウントに実際に適用される契約を確認してください。確認日時点では、次のとおりです。

• API 規約では、Luma は API Input または API Output を使用して、AI/ML モデルのトレーニング、ファインチューニング、開発を行わないと定めています。この API 固有の条項は、一般的な規約のうち矛盾する条項に優先します。ただし、保持が一切ないこと、人または自動によるモデレーションが一切ないこと、Usage/Aggregated Data が一切ないこと、または処理が一切ないことを保証するものではありません。
• 利用規約では、当事者間において、かつ法律で認められる最大限の範囲で、顧客が Output を所有し、Luma は自身の権利を譲渡すると定めています。これは、第三者の権利を付与するものでも、独自性、著作権保護の可否、非侵害、正確性を保証するものでも、それだけで商用利用を許可するものでもありません。商用利用には、それを許可する有効な有料サブスクリプションまたは Order が必要です。
• 公開文書には、API Input/Output の固定された保持期間や、地域またはデータレジデンシーに関するエンドポイントの記載はありません。プライバシーポリシーでは、目的および法律に基づく保持基準が定められており、エンタープライズ向けのデータ処理には、DPAおよび Order が適用される場合があります。必要に応じて、保持、削除、サブプロセッサー、データレジデンシー、セキュリティ、インシデント対応について、書面による確約を取得してください。
• 規約では、禁止データに GDPR 第 9 条の特別カテゴリー、PHI、PCI データ、COPPA/GLBA の規制対象データ、政府発行の ID、および同様の規制対象データが含まれると定めています。Luma は、サービスは HIPAA 準拠を目的として設計されておらず、Luma は Business Associate ではないと明記しています。このようなデータを送信しないでください。
• 生成された Output が AI 生成であることを、すべての下流 API ユーザーに伝えてください。また、すべての素材、顔、肖像、キャラクター、ロゴ、商標、著作物、場所について、文書化された権利および同意を取得してください。Output が人物を特定する、または人物に似ている場合、規約により AI 生成であることを公に開示する必要があります。規約では、Output を人間が生成したものと偽ることも禁止されています。
• 規約には、アップロードされた写真に描かれた人物に影響する特定のライセンス条項があります。通常の素材の所有権だけですべての被写体の権利が解決すると想定せず、実在人物を参照するプログラムは弁護士に相談してください。
• コンテンツモデレーションポリシーおよび規約では、性的/NSFW コンテンツ、有害または違法なコンテンツ、ヘイト/差別、権利侵害、欺瞞的なディープフェイク/なりすまし、政治的操作/誤情報、詐欺、児童の性的対象化、その他の虐待的な利用などのカテゴリーが禁止されています。アプリケーションレベルでの事前チェックと人によるレビューを実施してください。プロバイダーのモデレーションは、権利関係の確認に代わるものではありません。
• API 顧客は、下流ユーザーに対して、Luma の AUP と同等以上に保護的な書面による制限を課し、苦情処理プロセスを運用し、Luma による一時停止/終了の要求に従う必要があります。また、API 規約では、API または Output を使用して、AI のトレーニング、ファインチューニング、評価、その他のモデル開発用データセットを構築することも禁止されています。

監査可能なマニフェストに、ブリーフ ID、リクエスト/プロンプトのハッシュ、モデル、参照素材のハッシュと権利/同意の記録、生成 ID/状態/失敗、タイムスタンプ、Output の 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、サブプロセッサー