在不必管理多個不同 API 的情況下,以最簡單方式在規模化自動生成影像,就是將工作流程與模型供應商解耦。把每個影像請求都放入同一個佇列,將每個作業路由到當前模型 ID,並透過一把 CometAPI 金鑰與 OpenAI 相容的基底 URL https://api.cometapi.com/v1. 將相容請求發送出去。
本教學以 Python 搭建該管線。它從一個 JSON Lines 佇列接收產品、廣告與內容作業;選擇模型;限制並發;重試暫時性失敗;以 URL 或 base64 形式儲存結果;並記錄每個作業的使用量與預估成本。範例將這些生產必需項目濃縮在一個精簡區塊裡,讓你可以在不把文章變成程式碼參考的前提下測試工作流程。
統一影像 API 如何簡化批次生成
最終,工作流程將會是這樣:
jobs.jsonl → bounded worker pool → CometAPI /v1/images/generations → object storage → manifest.jsonl
佇列與儲存層由你自行掌控。切換影像模型只需更改 model 的值,而非變動驗證系統或主要請求路徑。這就是統一影像 API 的實務優勢:模型選擇變成同一條管線內的路由決策,而不是另外整合一個供應商。
自動化影像生成需要什麼?
你需要 Python 3.10 或以上版本、requests 套件、一把 CometAPI 金鑰、一個可寫入的輸出位置,以及至少一個當前有效的影像模型 ID。
安裝唯一依賴:
pip install requests
在伺服器上設定你的金鑰,切勿放在瀏覽器端程式碼或版本庫中:
export COMETAPI_KEY="your-key"
基底 URL 是 https://api.cometapi.com/v1,相容的文本轉影像工作使用 POST /images/generations。在部署前,請於即時模型目錄驗證每個模型;該目錄會返回當前 ID、支援的端點、功能與定價中繼資料,且無需授權標頭。
截至 2026 年 8 月 20 日,即時目錄列出了下列兩個實用路由:
| 工作負載 | 模型 ID | 適用原因 |
|---|---|---|
| 具可控輸出設定的產品圖片 | gpt-image-2 | 在已文件化的 OpenAI 相容路由上返回使用量資料與 base64 影像內容 |
| 大量廣告與內容概念 | doubao-seedream-4-5-251128 | 使用相同的生成路由,並以每次請求定價列出 |
此表僅作為起點,並非聲稱這些模型具有相同能力。尺寸、品質、格式、參考影像支援與回應行為仍是模型特定的。請在傳遞可選參數前檢查模型記錄及其連結文件。
如何用 Python 建立批次影像生成工作流程
1. 為每個作業給一個可持久的 ID
使用每行一個 JSON 物件,讓佇列、資料庫匯出或試算表作業都能餵給同一個工作器:
{"id":"sku-1001","kind":"product","prompt":"Studio product photo of a ceramic coffee dripper on a warm neutral background"}
{"id":"campaign-204","kind":"ad","prompt":"Editorial summer travel image, vivid natural light, wide composition, no text"}
{"id":"blog-088","kind":"content","prompt":"Minimal illustration of a developer automating a creative workflow, no text"}
該 ID 會成為輸出檔名與清單鍵。於生產環境,將它作為冪等鍵,並在重新處理佇列前略過已標記為成功的 ID。
2. 先按作業類型路由,然後對照即時目錄驗證
此範例將產品工作映射到 gpt-image-2;將廣告或內容工作映射到 doubao-seedream-4-5-251128。作業可用自身的 model 欄位覆寫該選擇。啟動時,工作器會下載公開目錄,並拒絕已不再列出的 ID。
這比在整個應用中硬編碼供應商專屬 SDK 更安全。你可以在評估你自己提示的品質、延遲與價格後,在一個映射中更改路由。
3. 限制並發,而不是一次啟動整個批次
工作器預設以四個並行請求啟動。這個數字是保守的應用設定,而非通用的服務限制。測量你的帳戶延遲與 429 回應,然後有意識地調高或調低 MAX_WORKERS。
僅對 408、429 與 5xx 回應進行帶抖動的指數退避重試。驗證錯誤、無效的模型 ID 與不支援的參數會立即失敗,因為重試相同的不良請求只會增加延遲。
4. 在儲存前正規化結果
影像模型不一定返回相同的容器。已文件化的 GPT Image 回應包含 data[0].b64_json;其他相容模型可能返回 data[0].url。工作器同時處理兩者,將影像寫入暫存檔,且僅在下載或解碼成功後才重新命名。
在生產環境,將本機 output/ 目錄替換為 S3、R2、GCS 或其他物件儲存。除非保留政策明確說明,否則不要將供應商託管的 URL 視為永久儲存。
5. 記錄使用量、嘗試次數與預估成本
每個結果都會成為精簡的清單列,包含作業 ID、模型、儲存路徑、狀態,以及在即時目錄提供充分定價資料時的預估美元成本。失敗的作業保留錯誤,而不是從批次中消失。
批次影像生成的完整 Python 腳本
將以下內容存為 batch_image_pipeline.py,將佇列放在同目錄的 jobs.jsonl,然後執行 python3 batch_image_pipeline.py。
import base64, json, os, random, time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import requests
BASE_URL = "https://api.cometapi.com/v1"
KEY = os.environ["COMETAPI_KEY"]
WORKERS = int(os.getenv("MAX_WORKERS", "4"))
OUT = Path("output")
ROUTES = {
"product": "gpt-image-2",
"ad": "doubao-seedream-4-5-251128",
"content": "doubao-seedream-4-5-251128",
}
catalog = requests.get("https://api.cometapi.com/api/models", timeout=30)
catalog.raise_for_status()
CATALOG = {model["id"]: model for model in catalog.json()["data"]}
def generate(job):
model = job.get("model", ROUTES[job["kind"]])
if model not in CATALOG:
raise ValueError(f"Unknown model: {model}")
payload = {"model": model, "prompt": job["prompt"], "n": 1}
if model == "gpt-image-2":
payload.update(quality="low", size="1024x1024", output_format="jpeg")
for attempt in range(4):
response = requests.post(
f"{BASE_URL}/images/generations",
headers={"Authorization": f"Bearer {KEY}"},
json=payload,
timeout=180,
)
if response.status_code not in {408, 429} and response.status_code < 500:
break
time.sleep(2**attempt + random.random())
response.raise_for_status()
body = response.json()
item = body["data"][0]
if item.get("b64_json"):
data = base64.b64decode(item["b64_json"])
extension = body.get("output_format", "png")
else:
download = requests.get(item["url"], timeout=120)
download.raise_for_status()
data = download.content
extension = {"image/png": "png", "image/webp": "webp"}.get(
download.headers.get("content-type"), "jpg"
)
path = OUT / f"{job['id']}.{extension}"
path.write_bytes(data)
price, usage = CATALOG[model].get("pricing") or {}, body.get("usage", {})
cost = price.get("per_request")
if cost is None and price.get("input") is not None:
cost = (usage.get("input_tokens", 0) * price["input"] +
usage.get("output_tokens", 0) * price["output"]) / 1_000_000
return {"id": job["id"], "model": model, "path": str(path),
"estimated_usd": cost * price.get("ratio", 1) if cost is not None else None}
def safe_generate(job):
try:
return {"status": "success", **generate(job)}
except Exception as error:
return {"id": job["id"], "status": "failed", "error": str(error)}
OUT.mkdir(exist_ok=True)
jobs = [json.loads(line) for line in Path("jobs.jsonl").read_text().splitlines() if line]
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
results = list(pool.map(safe_generate, jobs))
with (OUT / "manifest.jsonl").open("w") as manifest:
manifest.writelines(json.dumps(result) + "\n" for result in results)
此腳本會在執行時使用當前目錄,而兩個後備映射是於 2026 年 8 月 20 日驗證過的範例。在發布或於其他日期部署此程式碼前,請重新檢查。
如何測試批次影像生成工作流程
從一個作業與一個工作器開始:
MAX_WORKERS=1 python3 batch_image_pipeline.py
成功的 GPT Image 回應結構如下:
{
"created": 1776841943,
"output_format": "jpeg",
"quality": "low",
"size": "1024x1024",
"usage": {
"input_tokens": 16,
"output_tokens": 208,
"total_tokens": 224
},
"data": [{"b64_json": "<base64-image-data>"}]
}
工作器會解碼影像,寫入 output/<job-id>.jpeg,並在 output/manifest.jsonl 新增一筆成功記錄。若某個模型改為返回 URL,工作器會下載該檔並以相同清單格式儲存本機路徑。
此程式已在本地進行語法檢查。實際生成呼叫仍需要你的 CometAPI 金鑰,因此在提高並發前請先執行單作業的冒煙測試。
批次影像生成要花多少錢?
定價必須有時間戳記,因為模型費率會變動。截至 2026 年 8 月 20 日,CometAPI 即時模型目錄 返回以下基本價格欄位以及 0.8 的計費比率:
gpt-image-2:每 100 萬輸入權杖 $5、每 100 萬輸出權杖 $30;套用所列比率後,分別為每 100 萬 $4 與 $24。doubao-seedream-4-5-251128:每次請求 $0.04;套用所列比率後為每次請求 $0.032。
CometAPI 定價指南 說明對具官方定價的模型採用權杖計費、以及對每次請求計價的模型採用呼叫制計費。腳本在執行時讀取目錄,並使用相同規則:
token cost = ratio × (input tokens × input rate + output tokens × output rate) / 1,000,000
request cost = ratio × per-request price
例如,上述已文件化的 GPT Image 回應報告了 16 個輸入權杖與 208 個輸出權杖。使用 8 月 20 日的目錄數值,該示例結果的預估成本約為 $0.005056。實際總額會因模型、品質、尺寸、提示、重試次數與回應使用量而變動。請將 API 回應與帳戶使用儀表板視為計費紀錄,而非固定的每張影像假設。
也要為未成功的工作預算。未確認逾時後的重試可能產生第二次可計費結果,而技術上成功但審核未通過的影像同樣會消耗預算。追蹤 API 成本與接受率:
effective cost per accepted image = total batch spend / approved images
常見影像生成 API 錯誤與修正方法
| 症狀 | 可能原因 | 修正 |
|---|---|---|
| 401 | 缺少或無效的金鑰 | 檢查伺服端的 COMETAPI_KEY |
| 400 | 無效模型或不支援的選項 | 重新檢查即時目錄並移除模型特定欄位 |
| 429 | 並發過高 | 降低 MAX_WORKERS 並保留指數退避 |
| 重複 5xx | 上游暫時性失敗 | 在上限內重試,然後將作業移至死信佇列 |
| 沒有儲存影像 | 回應使用了不同的容器 | 檢查 data[0],同時支援 b64_json 與 url |
| 重複花費 | 在部分失敗後重播了作業 | 使用可持久 ID,並僅在儲存成功後才確認 |
不要重試所有錯誤。永久性的 400 請求會持續無效,而不受限的 429 重試迴圈會把流量尖峰變成積壓。
規模化生產影像生成的最佳實務
在涉及多個工作器時,從 JSON Lines 轉換到可持久的佇列。設定一個比最大生成時間更長的可見逾時,僅在影像與清單都儲存後才確認作業,並將耗盡重試的作業送入死信佇列審查。
將可選控制放在模型特定的設定中。共享負載應僅包含 model、prompt 與 n: 1 等共通欄位;僅在所選模型文件確認後才加入 quality、size 或 output_format。若加入後備路由,請選擇支援相同任務的模型,並為該模型重建負載,而不是盲目重播供應商專屬選項。
將 API 金鑰存放於秘密管理器、限制提示輸入、依你的政策掃描生成資產,並避免將供應商 URL 放入長期產品記錄。記錄作業 ID、模型 ID、延遲、嘗試次數、使用量、儲存路徑、審核結果與目錄快照日期。這些欄位可讓你以「每張被接受影像的成本」比較模型,而不僅是標稱價格。
最後設定預算護欄:最大批次大小、每作業重試上限、每日花費警示,以及當通過率下降時的停止條件。把糟糕的提示放大規模並非優化。
自動化影像生成常見問答
以最簡單方式在規模化自動生成影像、且不必管理多個 API 的方法是什麼?
使用一個佇列與儲存工作流程,然後透過一把 CometAPI 金鑰與 https://api.cometapi.com/v1/images/generations. 發送相容的影像請求。在你的路由層修改模型 ID,而不是維護獨立的驗證與供應商 SDK。
我可以發送一個請求並讓多個影像模型同時生成嗎?
範例是每個作業使用一個模型。扇出屬於應用工作流程:複製一個作業,使用不同的 ID 與模型值,然後比較儲存的輸出。這能讓成本與審核狀態歸屬到各自模型。
應該使用多少並發?
沒有適用於每個帳戶與模型的通用數字。從小型受限的工作池開始,例如四個工作器,監控延遲與 429 回應,並依據證據調整。
我應該儲存返回的 URL 還是影像本身?
將影像儲存在你自己的物件儲存中。返回的 URL 可能是暫時性的,而 GPT Image 模型可能返回 base64 內容而非 URL。
我如何選擇最便宜的模型?
計算每張被接受影像的成本,而不只是每次呼叫的價格。納入權杖或請求費用、重試、失敗下載、被拒絕的資產、後處理與人工審核。於發布或部署當天重新檢查即時模型目錄。