キューを崩壊させない画像・動画のバッチ生成

イラスト：画像用と動画用のキューを分けると、それぞれの負荷が自分のペースで進みます。

バッチ生成が壊れるのは、たいてい地味な場面です。1 枚の画像を生成しているときではなく、顧客が 20,000 行をインポートし、アプリケーションがそのすべてを一度にライブのリクエストに変えたときです。

API は正常で、すべてのペイロードが有効でも、システムは崩壊します。データベース接続が埋まり、コールバックがコミットの速度を上回って届き、大きなテナントが他のテナントを締め出し、再試行が高価な動画を重複させ、トラフィックの急増がそのまま請求額になります。

待機中の作業は永続キューに置き、実行中のタスク数に上限を設け、リモートタスクがすでに存在するかどうかが分かっている場合にだけ再試行します。Promise.all を大きくしても、そのどれも実現できません。

以下のすべてを左右する事実が 1 つあります。SpicyAPI には一括・バッチ用のエンドポイントがありません。各行は、独自のタスク、ホールド、精算を持つ 1 回の createTask 呼び出しであり、受け付けられたタスクはキャンセルできません。そのため、キュー、同時実行数の上限、停止ボタンはすべてアプリケーション側に置くことになります。

まず Playground で試す

z-image Playground を開きます。代表的な Prompt を書き、Aspect ratio を選び、バッチで使う予定のとおりに Prompt expansion を設定して、Run model を押し、Output を確認します。その 1 件の結果が望みどおりになってから規模を広げてください。スケジューラーはリクエストを、その誤りごと何倍にも増やします。

結論

• バックログは自前の永続キューに置きます。HTTP リクエストやメモリ上の promise には置きません。
• 実行中タスク数の上限は自分で決めます。 プラットフォームのレート制限（読み取りと書き込みそれぞれ 10 秒あたり 1,000 リクエスト）は、受け付け済みタスクの同時実行数を制限しません。制限するのは残高と支出上限です。メディア種別ごとに数を決め、タスクが終了状態になるまでその枠を保持します。
• 受け入れ判定はキューの長さではなくキューの滞留時間で行い、最初のディスパッチの前にバッチ全体のコストを確定します。
• 論理的な出力 1 つにつき、保存した冪等性キーを 1 つ使います。常に同じ API キーで送信し、24 時間を過ぎたら信頼しません。
• エラーコードで再試行を判断します。容量や通信のエラーは再試行し、無効な入力、残高ゼロ、支出上限は再試行しません。
• テナント間はラウンドロビンにし、完了処理は動かしたままディスパッチだけを止められるキルスイッチを用意します。

3 つの上限と、それぞれを決める主体

送信、ディスパッチ、完了を別々の段階として扱います。

クライアントのバッチ

│

▼

受け入れ + コスト確認 ──► 永続キュー ──► 上限付きワーカー ──► createTask

│

ポーリング / Webhook による完了

│

出力をコピー ──► 結果ストア + 顧客向けイベント

レート制限が管理するのは送信であって同時実行数ではありません。受け付け済みのタスクは、生成中にリクエストバケットを消費し続けません。

1 秒あたり 5 件の完了なら、作成するタスクも約 5 件/秒で、100 件/秒のバケットのごく一部です。ほとんどのバッチでは、送信の上限はスロットルではなく安全ヒューズです。先に効くのは、実行中タスクが保持する資金と、自分で設定した上限です。ポーリングは別の読み取りバケットを使うため、暴走したポーリングループが createTask を妨げることはありませんが、429 が recordInfo で返り始めて状況が見えなくなることはあります。デフォルト値を決め打ちせず、ヘッダーを読んでください。

他の API では、同時実行数と待機キューの上限が別々に課される場合があります。Leonardo の同時実行ガイドはそれらを区別しています。SpicyAPI には実行中タスク数の数値上限が文書化されていないため、リクエストレート制限に頼らず、自前の上限を設定してください。

実行中タスクの上限を時間と費用から見積もる

安定した 1 つの負荷クラスについて、リトルの法則は、システム内の平均タスク数を、完了レートと、各タスクがそこで過ごす平均時間に結び付けます。

平均の実行中タスク数 ≈ 1 秒あたりの完了数 × 受け付けから終了状態までの平均秒数

画像タスクが平均 8 秒で、1 秒あたり 5 件必要な場合：

5 件/秒 × 8 秒 = 同時 40 件の画像タスク

40 は平均的な占有数であり、保証ではありません。最初の上限を 50 にすると 25 % の余裕が加わります。引き上げるのは、キューの滞留時間、エラー、支出が健全な範囲にある間だけにしてください。分布の裾やキュー滞留時間のアラームには p95 と p99 を使います。タスクの所要時間は、createdAt と completedAt（どちらも recordInfo にあります）から、モデルと設定ごとに測定します。

動画は専用のプールに分けます。ワークロードで Seedance 2.0 のクリップが平均 70 秒かかり、1 秒あたり 0.2 件にしたい場合：

0.2 件/秒 × 70 秒 = 同時 14 件の動画タスク

8 秒と 70 秒という平均は規模見積もり用の仮定であり、約束された生成時間ではありません。

50 の枠を 1 つのプールで共有すると、長い動画が集中したときにすべての枠を占有し、短い画像が待たされます。画像用 50 枠と動画用 14 枠に分ければ、どちらのクラスも進みます。Z-Image と Seedream 5.0 Pro も、所要時間や優先度が違うなら別々のプールにできます。

次に、上限を費用と照らし合わせます。受け付けられた各タスクは、精算されるまで estimatedCost をホールドし、そのホールドはすべてのキーとチームの上限に算入されます。公開カタログの価格で計算すると次のとおりです。

これらのホールドを賄えない残高では、スケジューラーがどれほど優秀でも、新しいタスクは 40201 で失敗します。それより重要なのは総支出です。1K の Seedream 5.0 Pro 画像 20,000 枚は $720、Prompt expansion をオフにした 1 枚 $0.01 の Z-Image 画像 20,000 枚は $200 です。価格は変わるため、最新のモデルページか見積もりで確認してください。悪いバッチを止める上限は、開始前に設定した上限です。ディスパッチの前に、本番キーに 1 日の支出上限を設定します。額は、最悪の 1 日に失っても耐えられる最大額に合わせます。上限が空欄なら無制限で、チームの制限でも新規タスクが 40202 で止まることがあります。

もう半分は、処理しきるまでの時間の見積もりです。

処理しきるまでの時間 ≈ キュー内のジョブ数 ÷ 持続可能な 1 秒あたりの完了数

キューにある 18,000 枚の画像を毎秒 5 枚で処理すると、1 時間分の作業です。開始までの待ち時間を 15 分と約束するなら、受け入れは 18,000 件よりはるか手前で絞る必要があります。増え続けるバックログは定常状態ではないので、方程式より、観測したキュー滞留時間を信頼してください。

バックプレッシャーはディスパッチの前から始まる

バックプレッシャーとは、下流のシステムが上流の生産者を減速させられることです。これがないと、キューは際限なく作業を吸い込み、顧客は過負荷に何時間も後で気づきます。

キュー滞留時間（最も古い実行可能アイテムの経過時間）に対して、受け入れの閾値を 2 つ設けます。アイテムのコストがまちまちなとき、長さは実態を隠します。滞留時間なら、約束した待ち時間が守られているかが直接分かります。

• ソフト閾値： バッチは受け付け、遅延見込みを返し、生産者を減速させます。
• ハード閾値： リモートタスクが作られる前に、再試行可能なプロダクトのレスポンスで新規作業を保留または拒否します。

金額は 2 進浮動小数点ではなく、正確な 10 進数の単位で比較します。キューのアダプターで、予約はバッチの受け入れとアトミックにコミットし、同時の送信が同じ残りの予算を使わないようにします。

内部の過負荷をすべて、自社の顧客向けの API 429 に変えてはいけません。作業が永続的に保存されているなら、遅延状態で受け付け済みのバッチ ID を返します。責任を持って保持できない作業だけを拒否し、その際は再試行の時刻を示し、バッチの送信そのものを冪等にします。

この API の 2 つの仕様が、アイテムをキューで安全に待たせられる時間を決めます。

• 参照ファイルは 1 日で失効します。 アップロードで得た spicy:// URI は、コミットのレスポンスにある expiresAt（既定では 1 日）まで有効です。再試行は元の入力を再利用するため、それ以降の再試行は 400 を返します。キューには公開 HTTPS URL か自前のソース ID を保持し、アップロードはディスパッチ時に行います。
• 見積もりの有効期間は 5 分です。 価格をロックしたいなら、ディスパッチ時に見積もりを取り、quoteId と expectedCost を createTask に送ります。価格が変わっていれば、新しい金額を請求せず 40901 で失敗します。見積もりは無料ですが読み取りバケットを消費するため、受け入れ時に 20,000 行ぶんの見積もりを取るのは適切ではありません。

冪等性は 2 層ある

防ぐべき重複は 2 種類あります。顧客が同じバッチを 2 回送信すること、そしてディスパッチャーが作成の成否を把握できていないことです。バッチには安定した識別子を与え、新しい論理アイテムごとに UUID を生成して保存します。復旧時は、そのアイテムの保存済みキーを再利用します。

そのキーにはデータベースで一意制約を付けます。API を呼ぶ前にアイテムを「intent」として書き込み、キーを Idempotency-Key として送り、返ってきた taskId を同じ行に保存します。クラッシュ後、受け付けの成否がまだ不明なら、次のワーカーが保存済みキーを再利用します。送信後にようやく intent を書くと、再起動によって同じ出力を二重に購入してしまう隙間が残ります。

冪等性のドキュメントが追加で定めており、キューが守るべき点は次のとおりです。

• キーは 最初に受け付けられたリクエストから 24 時間有効で、リプレイしても延長されません。それ以降は、同じキーで新しい課金対象のタスクが作られます。キューまたはデッドレターに残った古いアイテムを再実行する前に、キーを信用せず、保存済みの taskId を照会してください。
• キーは 同じアカウントと同じ API キー、および厳密に同じモデル、input、callBackUrl に紐づきます。どれかを変えると 409 が返ります。同じアカウントの別のキーも 409 になり、タスクを読み取れません。タスクを作成したキーだけが、それを読み取り、再試行できます。ワーカー群では、作成、ポーリング、再試行に同じキーを使ってください。
• リプレイは元のタスクを現在の状態で返し、リプレイであることを示す印は付きません。
• 同じキーでまれに競合が起きると、409「Idempotency key conflict, please retry shortly」が返ります。数百ミリ秒後に 1 回だけ再試行してください。他の 2 種類の 409 メッセージは、再試行しても成功しません。

タスクが終わるまで枠を保持するディスパッチャー

batch.jsonl に 1 行 1 オブジェクトで保存します。たとえば {"model": "alibaba/z-image-turbo/text-to-image", "input": {"prompt": "<one row's prompt>"}} のように、選んだモデルが受け付ける入力を使います。SPICY_API_KEY、BATCH_BUDGET_USD、必要なら BATCH_WORKERS（既定は 2）を設定します。この Python の例を実行し、バッチの上限を承認します。各ワーカーは、タスクが終了状態になり、精算が済み、出力の準備中でなくなるまで占有されたままになります。

復旧用に batch-run フォルダーを保存し、元の行の順序は変えないでください。同じフォルダーが、再起動をまたいで精算済みの支出と予約を保持します。1 つのフォルダーにつきコーディネーターは 1 つだけ実行します。

import json
import os
import time
import uuid
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from decimal import Decimal
from pathlib import Path
from threading import Event, Lock

BASE = "https://api.spicyapi.ai/api/v1"
API_KEY = os.environ["SPICY_API_KEY"]
CAP = Decimal(os.environ["BATCH_BUDGET_USD"])
WORKERS = int(os.environ.get("BATCH_WORKERS", "2"))
ROOT = Path(os.environ.get("BATCH_DIR", "batch-run"))
ROOT.mkdir(exist_ok=True, mode=0o700)
lock, stop = Lock(), Event()
reserved = Decimal("0")


class APIError(Exception):
    def __init__(self, status, body):
        self.status, self.code = status, body.get("code")
        super().__init__(f"{status}/{self.code}: {body.get('msg')}")


def call(path, payload=None, key=None):
    headers = {"Authorization": "Bearer " + API_KEY, "Content-Type": "application/json"}
    if key:
        headers["Idempotency-Key"] = key
    for attempt in range(5):
        request = urllib.request.Request(BASE + path,
            data=None if payload is None else json.dumps(payload).encode(), headers=headers)
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                body = json.load(response)
                if body.get("code") != 200:
                    raise APIError(response.status, body)
                return body["data"]
        except urllib.error.HTTPError as error:
            try:
                body = json.load(error)
            except ValueError:
                body = {}
            if error.code != 429 and error.code < 500:
                raise APIError(error.code, body)
            delay = float(error.headers.get("Retry-After") or min(15, 2 ** attempt))
        except (urllib.error.URLError, TimeoutError, ValueError):
            delay = min(15, 2 ** attempt)
        time.sleep(delay)
    raise RuntimeError("Unresolved API call; preserve this row for recovery")


def save(path, row):
    temporary = path.with_suffix(".tmp")
    temporary.write_text(json.dumps(row))
    temporary.chmod(0o600)
    temporary.replace(path)


def worker(pair):
    global reserved
    index, line = pair
    if stop.is_set():
        return
    try:
        request = json.loads(line)
        if not isinstance(request.get("model"), str) or not isinstance(request.get("input"), dict):
            raise ValueError("model and input are required")
    except (ValueError, TypeError, AttributeError) as error:
        print(index, "invalid local row:", error)
        return
    path = ROOT / f"{index}.json"
    row = json.loads(path.read_text()) if path.exists() else {
        "request": request, "key": str(uuid.uuid4())}
    if row["request"] != request:
        stop.set()
        raise ValueError("Batch changed; preserve the old folder and use a new one")
    if row.get("settled") or row.get("rejected"):
        return
    accounted = "sentAt" in row
    try:
        if not row.get("taskId"):
            if time.time() - row.get("sentAt", time.time()) >= 23 * 3600:
                raise RuntimeError("Reconcile old acceptance before submitting again")
            if "submission" not in row:
                quote = call("/jobs/quote", request)
                row["hold"] = quote["maxCharge"]
                row["submission"] = {**request, "quoteId": quote["quoteId"],
                                     "expectedCost": quote["estimatedCost"]}
            with lock:
                if stop.is_set():
                    return
                hold = Decimal(row["hold"])
                if not accounted and reserved + hold > CAP:
                    stop.set()
                    print(index, "batch budget reached")
                    return
                if not accounted:
                    reserved += hold
                    accounted = True
            row.setdefault("sentAt", time.time())
            save(path, row)
            accepted = call("/jobs/createTask", row["submission"], row["key"])
            row["taskId"] = accepted["taskId"]
            save(path, row)
        task_id = row["taskId"]
        for _ in range(3000):
            task = call("/jobs/recordInfo?taskId=" + urllib.parse.quote(task_id))
            if task["state"] not in {"queued", "running", "succeeded", "failed", "expired", "canceled"}:
                raise RuntimeError("Unknown state: keep the reservation")
            if task["state"] in {"succeeded", "failed", "expired", "canceled"} and task.get("settled") is True:
                assets = task.get("output", {}).get("assets", [])
                if task["state"] == "succeeded" and any(a.get("pending") for a in assets):
                    time.sleep(5)
                    continue
                row["result"], row["settled"] = task, True
                save(path, row)
                with lock:
                    reserved += Decimal(task["cost"]) - Decimal(row["hold"])
                print(index, task_id, task["state"], task["cost"])
                return
            time.sleep(5)
        raise RuntimeError("Wait interrupted; resume the saved task, not a new generation")
    except APIError as error:
        if error.status == 400 and error.code in {400, 40003, 40004} and not row.get("taskId"):
            row["rejected"] = str(error)
            save(path, row)
            with lock:
                if accounted:
                    reserved -= Decimal(row.get("hold", "0"))
            print(index, "rejected input:", error)
            return
        stop.set()
        print(index, "paused:", error)
    except Exception as error:
        stop.set()
        print(index, "paused:", error)


if CAP <= 0 or not 1 <= WORKERS <= 20:
    raise ValueError("Use a positive budget and 1–20 workers")
# Count settled spend from this batch again on restart.
for path in ROOT.glob("*.json"):
    row = json.loads(path.read_text())
    if row.get("settled"):
        reserved += Decimal(row["result"]["cost"])
    elif row.get("sentAt") and not row.get("rejected"):
        reserved += Decimal(row["hold"])
if input(f"Authorize this batch up to ${CAP}? Type yes: ") != "yes":
    raise SystemExit("No tasks submitted")
rows = Path("batch.jsonl").read_text().splitlines()
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
    list(pool.map(worker, enumerate(rows)))

デプロイについては本番連携のドキュメントを参照してください。この例は、無効な行を切り離し、残高や不確かなエラーが出たら新規の受け入れを一時停止し、一時的な再試行に上限を設けます。受け入れを止めた後も、すでに受け付け済みのタスクは回収され続けます。有料の自動再試行は意図的に行いません。失敗した行は、精算後に明示的にやり直すために記録されたまま残ります。このようなプロセスをメディア種別ごとに 1 つ実行し、ファイルの前段にテナントのラウンドロビンを置いてください。

公平なスケジューリング：小さなテナントを守る

先入れ先出しはジョブに対して公平であって、顧客に対して公平ではありません。テナント B が 10 件をキューに入れる前に、テナント A が 1 分間に 100,000 件を入れると、B は A のすべての後ろで待たされます。テナントごとにキューを持ち、1 巡ごとにテナントあたり 1 件ずつ取り出します。

1 巡目：A1、B1、C1

2 巡目：A2、B2、C2

重みで購入した容量を反映してもかまいませんが、アクティブな各テナントに最低限の取り分を与えてください。テナントごとの実行中タスク数にも上限を設けます。全体の上限はアカウントを守り、テナントの上限はうるさい隣人を封じ込めます。テナント内では、優先度レーン（インタラクティブなプレビュー、通常のバッチ、バックフィル）を追加する価値があるのは、プレビューのレイテンシーが現実の不満になってからです。追加するなら、優先度をキュー内の時間とともにエージング（加齢）させてください。さもないと、プレビューが絶えず流れ込み、バックフィルが永久に飢餓状態になります。1 つのアカウント上のすべてが同じ 2 つのリクエストバケットと同じ残高を共有するため、テナント間の公平性を担うのは API ではなくあなたです。

再試行する前に失敗を分類する

「3 回再試行」は方針ではありません。正しい対応は、何が分かっているか、そしてタスクがすでに存在し得るかで決まります。この表は文書化されたコードを使っています。code の値はレスポンスボディにあり、タスクの失敗は state: "failed" と errorCode として、成功した recordInfo 呼び出し（code: 200）の中で届きます。

ワーカーが一斉に再試行しないよう、ランダム化したバックオフを使います。

確定した失敗と拒否された入力は、分類と安全な再実行アクションを付けてデッドレターキューに移します。作成の成否が曖昧なものや、受け付け済みで未解決のタスクは、保存したキーとタスク ID を付けて照合キューに残します。ローカルの再試行期限が過ぎたことは、失敗の証明にはなりません。失敗したタスクは通常、ホールドを全額解放しますが、課金対象の content_rejected の一部は見積もり額で精算されます。返金を前提とせず、settled と cost を読んでください。

コスト保護はスケジューラーの役割

同時実行数の制限が守るのはマシンであって、お金ではありません。完全に安定したワーカーでも、払えないバッチをディスパッチしてしまうことはあります。ディスパッチの前に、見積もりコストをすべての枠と照合します。

許可額 = テナントの 1 日の残り

∩ テナントの月間の残り

∩ プロジェクトのバッチの残り

∩ 全体の 1 時間あたりの残り

実際の精算と照合します。createTask は保持された estimatedCost を返し、成功したタスクはそれ以下の額で精算され、settled: true と cost が最終的な請求額です。API に拒否されたアイテムは自前の予約を解放し、受け付けられたアイテムは settled に切り替わるまで予約を保持します。HTTP のタイムアウトから推測してはいけません。

有効なガードは次のとおりです。

• 大きなバッチが始まる前に担当者へ提示する、バッチあたりの見積もりコストの上限
• テナントごとの 1 時間あたりのディスパッチ予算と、全体の支出速度のアラーム
• モデルごと、またはテナントごとのサーキットブレーカー
• アカウント残高より低く設定した API キーの上限と、残高低下のアラート
• 完了処理は動かしたまま、新規のディスパッチを止めるキルスイッチ

受け付け済みのタスクはキャンセルできないため、作成を止めることが唯一のブレーキです。コストのインシデントで被害を抑えるのは、キャンセルボタンではなく、実行中タスク数の上限です。Webhook の受信、古いタスクのポーリング、予約の精算、出力のコピーは続けてください。

バッチ規模での完了処理

タスクごとの callBackUrl への配信は 15 秒でタイムアウトします。登録済みのアカウントエンドポイントにも独自の配信が届き得るため、受信側は生のボディに対して署名を検証し、イベントを inbox テーブルに書き込み、2xx を返して、残りはワーカーに任せるべきです。v2 の配信 request_id で重複排除し、到着順ではなく data.state で判断します。取りこぼしたイベントへの備えとして、低頻度のポーリングも残します（Webhook のドキュメントを参照）。

タスクが終わるたびに、出力を自前のストレージにコピーします。結果のリンクは約 20 分間有効で、生成されたメディアの保持は最長でおよそ 14 日です（保持期間）。1 秒あたり 5 件の完了では、コピーの工程もスループットの一部になり、遅いアップローダーが「完了済み」のタスクの中に潜むボトルネックになります。保持期間を短くすると、すでに実行したタスクの結果が消えることもあります。

エラーだけでなくフローを観測する

役に立つダッシュボードは、4 つの問いに答えます。作業を受け付けているか、開始しているか、完了しているか、コストは想定どおりに精算されているか。モデル、メディア種別、テナントの階層、結果ごとに追跡します。

相関の連鎖（batchId、アイテム ID、冪等性キー、taskId、request_id）を 1 本、ログとイベントを通して引き継ぎます。プロンプト、API キー、非公開アセットの URL は、メトリクスのラベルや通常のログに含めません。アラートは症状に対して設定します。「キューに 50,000 件」は正常かもしれません。「最も古いインタラクティブなアイテムが 12 分経過」は対処が必要な状態です。

上限付きのバッチから始め、そのあと容量を増やす

容量を増やすのは、最も古いアイテムの経過時間、タスクの失敗、支出がしきい値内にある間だけにしてください。ワーカーを増やしても、過負荷の完了経路は直りません。

最初の有料バッチの前に、モックを使い、送信の直前と直後にワーカーを中断し、429、失敗したタスク、支出上限エラーを返させます。復旧では、保存済みの intent を再利用し、不明なタスクは未解決のまま残し、完了処理は止めずに新規のディスパッチを止めることが求められます。そのうえで、大きなインポートに受け入れを開放する前に、承認済みの小さなバッチを実行します。

完了処理については「Polling vs webhooks for async generation jobs」を、バッチの予算については Per-unit pricing vs subscriptions を参照してください。本番チェックリストでは、キーの設定とローンチ前のテストを扱っています。

FAQ

SpicyAPI にバッチ用エンドポイントや同時実行数の上限はありますか？

バッチ用エンドポイントはなく、各行が 1 回の createTask です。実行中タスク数の文書化された上限もありません（上限は上の表にあり、現在のレートは X-RateLimit-Limit で確認できます）。ピーク時にレート制限が低すぎる場合は、コンソールのサポートに、ピーク時の 1 秒あたりのリクエスト数とバッチのサイズを伝えて、より高い上限を依頼してください。

誤って開始したバッチをキャンセルできますか？

できません。完了処理は動かしたまま新規のタスクを止めるには、ディスパッチのキルスイッチを使います。

429 は再試行の回数を消費すべきですか？

Retry-After にジッターを加えて待ち、同じ冪等性キーを保ちます。スロットリングが繰り返されるなら、ディスパッチレートを下げてください。再試行は回数と時間の両方で上限を設けます。その枠を使い切っても受け付けの成否が不明なら、代わりのタスクを作る前に、保存済みの intent を照合します。

タイムアウト後の二重課金を防ぐには？

24 時間以内に、同じ Idempotency-Key と同じ API キーで再送し、taskId は取得したらすぐに保存します。24 時間を過ぎたら、代わりに保存済みの taskId を照会します。

キューだけで過負荷を防げますか？

いいえ。キューは過負荷を蓄えるだけで、制御するのはバックプレッシャーです。キュー滞留時間に対する受け入れの閾値、実行中タスク数の上限、そして約束した待ち時間を守れないときのプロダクトとしての回答が、それでも必要です。

同時実行数は最初いくつがよいですか？

メディア種別ごとに、1 秒あたりの完了数に平均タスク時間を掛け、約 25 % の余裕を足します。1 秒あたり 5 枚の画像を 1 枚 8 秒で処理するなら 40 です。引き上げるのは、キューの滞留時間、エラー、支出が健全な間だけにしてください。

優先度の高いジョブは制限を迂回できますか？

できません。優先度は、受け入れ可能な作業の中から選ぶだけで、全体の上限や予算は決して迂回しません。インタラクティブな作業には予約済みの容量を与え、優先度の低い作業はエージングさせて飢餓状態にならないようにします。

Agent に呼び出しを任せる

入力の準備と結果の返却を Agent に任せ、価格とクリエイティブな選択はあなたが承認したいときは、Agent を使います。

• CLI： 1 行ぶんの input オブジェクトを input.json として保存し（Z-Image の場合は prompt、aspect_ratio、enable_prompt_expansion、output_format、seed）、npx @spicyapi/cli tasks create --model alibaba/z-image-turbo/text-to-image --input-file input.json --wait を使います。CLI は価格を確認します。復旧用に、返されたタスク ID を保存してください。
• MCP： アシスタントに、spicyapi_task_create を使って alibaba/z-image-turbo/text-to-image のタスクを作成し、続けて spicyapi_task_wait を使うよう依頼します。MCP サーバーは、作成の前に価格の確認を求めます。
• Skill： npx skills add spicyapi-ai/spicy-skill でインストールし、$spicyapi に同じワークフローを実行して承認のために止まるよう依頼します。Skill は、設定済みの CLI または MCP の接続を使います。