fal genmedia CLI：批量队列、重试与成本控制

fal genmedia CLI 已经覆盖了模型发现、Schema 查看、异步请求 ID、文件下载和 JSON 回执等环节。真正需要自己补上的，是一个批处理编排层：控制并发、识别可恢复的错误、断点续跑，并为异步任务保存仍在处理中的请求 ID。

先完成安装配置，再让每次运行都可复现

macOS 和 Linux 可按项目 README 中的方式安装：

curl https://genmedia.sh/install -fsS | bash
genmedia setup --non-interactive --api-key "$FAL_KEY" --no-auto-update

Windows 则使用文档提供的 PowerShell 安装命令：

irm https://genmedia.sh/install.ps1 | iex
genmedia setup --non-interactive --api-key "$env:FAL_KEY"

genmedia CLI README 说明，genmedia 支持 FAL_KEY、JSON 输出、非交互式配置，以及可选的后台更新检查。CI 环境中，应通过运行器的密钥存储注入 API Key，不要把它直接写进脚本或命令日志。

启动批任务前，先固定目标端点，并查看它当前的实际契约：

genmedia models "image to video" --json
genmedia schema bytedance/seedance-2.0/image-to-video --json
genmedia pricing bytedance/seedance-2.0/image-to-video --json

fal 指南建议的顺序很实用：models → schema → run → status → download。不要假定某个视频端点可用的参数，换到另一个端点后也同样可用；workflow skill 特别建议，在出现校验错误后重新检查 Schema。

批处理要按队列设计，不要只写一个 shell 循环

官方 genmedia 文档提供了异步执行能力，但并没有承诺原生支持“读取 CSV 并运行 500 行”这样的命令，也没有提供自动重试策略。更合理的分工是：CLI 负责调用面向提供商的能力，外层封装负责充当队列控制器。

一条可持久化的任务记录，至少应包含以下字段：

字段	用途
`id`	用于恢复和去重的稳定输入标识
`endpoint`	实际调用的精确模型路由
`prompt`	支持复现与审计
`status`	`pending`、`submitted`、`complete`、`failed` 或 `skipped`
`request_id`	`genmedia status` 所需的请求标识
`attempts`	避免无限重试
`output`	确定性的本地输出路径
`error`	让失败任务可以被定位和处理

耗时较长的图生视频或视频任务应使用 --async。提交后立刻保存返回的 request_id，随后同时携带端点 ID 和请求 ID 轮询：

genmedia run bytedance/seedance-2.0/image-to-video \
  --image_url "$IMAGE_URL" \
  --prompt "Slow product turn on a studio table; no text or logo" \
  --duration 4 --resolution 720p --aspect_ratio 16:9 \
  --async --json > logs/shot-001-submit.json

REQUEST_ID=$(jq -r '.request_id' logs/shot-001-submit.json)
genmedia status bytedance/seedance-2.0/image-to-video "$REQUEST_ID" \
  --download "outputs/{request_id}_{index}.{ext}" --json > logs/shot-001-result.json

{request_id}_{index}.{ext} 这种命名模式可降低不同任务悄悄覆盖彼此文件的风险。JSON 回执应与生成的媒体文件放在一起，而不是丢进单独的临时目录。

只重试有机会恢复的失败

可以从下面这套策略开始：

重试网络超时、连接重置、HTTP 429 以及短暂的 5xx 响应。
采用指数退避，例如等待 2、4、8、16、32 秒，并加入少量随机抖动。
为每一行设置最大尝试次数，例如最多提交四次，或在一个时间窗口内最多轮询五次。
不要重试 401/403 认证错误、422 Schema 校验错误、安全拒绝或格式错误的 JSON。
如果提交超时，但请求可能已经到达提供商，创建第二个付费请求前应先检查已保存的请求记录。

这套策略应由你的封装层实现；公开的 genmedia README 介绍的是命令和生命周期操作，并未承诺自动重试行为。遇到 422 时，先读取 validation_errors，重新执行 genmedia schema，再修正被点名的字段，而不是盲目重新提交。

一份可直接复制的 Python 批处理封装

下面的封装脚本是同步图片批处理的起点：它通过参数列表调用 subprocess.run，跳过已有完整输出的路径，限制并发工作量，重试临时性的进程失败，并以原子方式写入清单。对于耗时的视频任务，应在轮询前持久化异步提交结果：submit -> write endpoint/request_id -> restart -> poll saved request_id -> download；只有不存在请求 ID 时才重新提交。

#!/usr/bin/env python3
import concurrent.futures as pool
import json, os, random, subprocess, tempfile, threading, time
from pathlib import Path

ENDPOINT = "fal-ai/flux/dev"
OUT = Path("outputs/images")
LOG = Path("outputs/logs")
MAX_WORKERS = 3
MAX_ATTEMPTS = 4
MANIFEST_LOCK = threading.Lock()

JOBS = [
    {"id": "shoe-001", "prompt": "Black running shoe, clean studio product photo", "file": "shoe-001.png"},
    {"id": "shoe-002", "prompt": "Black running shoe on wet pavement at dawn", "file": "shoe-002.png"},
]

OUT.mkdir(parents=True, exist_ok=True)
LOG.mkdir(parents=True, exist_ok=True)
MANIFEST = Path("outputs/manifest.json")
PREVIOUS = json.loads(MANIFEST.read_text()) if MANIFEST.exists() else {"results": []}
STATE = {r["id"]: r for r in PREVIOUS.get("results", [])}
DONE = {k: r for k, r in STATE.items() if r.get("status") == "complete"}
MAX_REQUESTS = len(JOBS) * MAX_ATTEMPTS
if MAX_REQUESTS > 100:
    raise SystemExit(f"request ceiling exceeded: {MAX_REQUESTS}")

def save_result(result):
    with MANIFEST_LOCK:
        STATE[result["id"]] = result
        payload = {"endpoint": ENDPOINT, "max_requests": MAX_REQUESTS,
                   "results": list(STATE.values())}
        fd, tmp = tempfile.mkstemp(dir=MANIFEST.parent, prefix="manifest.", text=True)
        with os.fdopen(fd, "w") as f:
            json.dump(payload, f, indent=2)
        os.replace(tmp, MANIFEST)

TRANSIENT_WORDS = ("429", "500", "502", "503", "504", "timeout", "temporarily", "connection")

def run_one(job):
    target = OUT / job["file"]
    receipt = LOG / f"{job['id']}.json"
    if job["id"] in DONE and target.exists() and target.stat().st_size > 0 and receipt.exists():
        return DONE[job["id"]]

    cmd = ["genmedia", "run", ENDPOINT, "--prompt", job["prompt"],
           "--num_images", "1", "--download", str(target), "--json"]
    last_error = ""
    for attempt in range(1, MAX_ATTEMPTS + 1):
        try:
            p = subprocess.run(cmd, text=True, capture_output=True, timeout=900)
            raw = p.stdout.strip()
            if p.returncode != 0:
                last_error = p.stderr[-1000:] or raw[-1000:]
                if not any(w in last_error.lower() for w in TRANSIENT_WORDS):
                    break
                if attempt < MAX_ATTEMPTS:
                    time.sleep((2 ** attempt) + random.random())
                continue
            try:
                data = json.loads(raw) if raw else {}
            except json.JSONDecodeError as exc:
                return {**job, "status": "failed", "attempts": attempt, "error": f"invalid JSON: {exc}"}
            if target.exists():
                (LOG / f"{job['id']}.json").write_text(json.dumps(data, indent=2))
                return {**job, "status": "complete", "attempts": attempt, "output": str(target)}
            last_error = p.stderr[-1000:] or raw[-1000:]
            if not any(w in last_error.lower() for w in TRANSIENT_WORDS):
                break
        except (subprocess.TimeoutExpired, OSError) as exc:
            last_error = str(exc)
        if attempt < MAX_ATTEMPTS:
            time.sleep((2 ** attempt) + random.random())
    return {**job, "status": "failed", "attempts": MAX_ATTEMPTS, "error": last_error}

results = []
with pool.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
    futures = [executor.submit(run_one, job) for job in JOBS]
    for future in pool.as_completed(futures):
        result = future.result()
        results.append(result)
        save_result(result)

print(json.dumps(results, indent=2))

处理视频时，替换 ENDPOINT，并补充该端点 Schema 所要求的参数。图生视频链路中，可先用 genmedia upload ./frame.png --json 上传本地图片一次，将返回的 URL 传给视频任务，并把两份记录都保存在清单中。这个封装并不表示 genmedia 自身能够估算或强制执行最高支出；它只负责防止本地重复工作，并限制重试次数。

付费生成前，先设置成本闸门

genmedia pricing <endpoint_id> --json 是价格查询，不是费用预留，也不是预算上限。应在开始批处理前查询价格，再根据行数、每行输出数、分辨率和时长设置、最大重试次数计算一个保守的上限。

一套实用的成本闸门可以包括：

控制项	实现方式
硬性行数上限	如果清单超过批准数量，则拒绝启动
模型档位	先用更便宜或更快的端点打样；通过 QA 后再渲染最终版本
输出数量上限	显式设置 `num_images`，不要依赖默认值
重试预算	将重试次数与首次尝试分开统计
断点续跑	跳过已经验证存在本地输出的行
取消任务	适当时使用 `genmedia status ... --cancel` 取消排队中的任务

应将价格查询响应连同清单一并保存，因为模型价格和端点可用性都可能变化。若提供商没有给出可比较的计价单位，就应将结果标记为请求次数上限，而不是账单金额。

fal genmedia CLI、Replicate CLI 与自定义脚本怎么选

这三类工具解决的是不同层次的问题。Replicate 官方 CLI 提供运行和流式获取预测、查看模型 Schema、上传、训练与模型管理等命令。如果你的工作从发现 fal 端点开始，并需要走完 fal 的队列与下载生命周期，genmedia 会更顺手。

选择	最适合的场景	主要取舍
fal genmedia CLI	fal 原生模型搜索、Schema 查询、异步任务、下载，以及 Agent Shell 使用	仅适用于特定提供商；批处理策略仍需在 CLI 外实现
Replicate CLI	Replicate 预测、流式输出、模型与 Schema 操作，以及训练命令	使用不同的提供商目录和生命周期；不要期待 fal 的端点 ID 或参数可以直接迁移
自定义 Python/HTTP 脚本	多提供商路由、审批闸门、数据库状态、队列和计费策略	认证、Schema 变化、轮询、下载和错误处理都由你负责

我的建议很直接：探索阶段直接使用 genmedia；fal 单一提供商的生产批处理，则在外面加一层小型 Python 封装。只有在切换提供商确实是需求时，才引入自定义的提供商抽象层，而不是因为封装脚本看起来不够“企业级”。

把输出当作记录管理，再进行 QA

公开 workflow skill 建议使用一份精简清单，记录目标、节点 ID、端点 ID、请求 ID、输入 URL、输出 URL、已下载文件和缺陷备注。相比堆满自动生成文件名的文件夹，这种方式更有用。

验收整批任务前，逐项确认：

每一条 complete 记录都有本地文件和 JSON 回执。
没有任何 failed 记录被悄悄遗漏。
图片尺寸符合预期，且文件大小不为零。
视频能够正常打开，并符合预期的时长、分辨率和帧率；可使用 ffprobe 检查。
保留的资产仍能追溯到对应的 Prompt 和端点 ID。
重新运行同一份清单时，应跳过已有任务，而不是产生重复文件。

fal 指南强调，生成的媒体文件应与其 JSON 元数据放在相近位置。这也会让日后的提供商迁移成为可能：你可以对比输出、请求和成本记录，而不必从文件名中反推一次任务的来龙去脉。

FAQ

fal genmedia CLI 原生支持数百条 Prompt 的批处理吗？

文档中的命令涵盖模型执行、异步状态处理、JSON 输出、上传和下载。但要实现可恢复的数百条 Prompt 批处理，仍需要额外的清单读取器和并发控制器。

genmedia 比 Replicate CLI 更便宜吗？

CLI 不决定提供商的模型价格。应针对你的实际工作负载，对比具体端点定价、输出设置、重试次数和传输行为；面对不同模型目录，无法给出工具层面的“更便宜”结论。

如果追求最短路径，就用 genmedia 完成发现与执行，再加入由清单驱动的封装层来处理批量任务；只有在需要多提供商路由或集中式任务状态时，才选择自定义脚本。

fal genmedia CLI：批量队列、重试与成本控制

先完成安装配置，再让每次运行都可复现

批处理要按队列设计，不要只写一个 shell 循环

只重试有机会恢复的失败

一份可直接复制的 Python 批处理封装

付费生成前，先设置成本闸门

fal genmedia CLI、Replicate CLI 与自定义脚本怎么选

把输出当作记录管理，再进行 QA

FAQ

fal genmedia CLI 原生支持数百条 Prompt 的批处理吗？

genmedia 比 Replicate CLI 更便宜吗？

Claude Opus 5.5

Claude Fable 5

Claude Fable 5.1

Claude Opus 4.8

Claude Opus 5