3.1 API 调用与参数

3.1 API 调用与参数

本节摘要：本节把生成段接到真实服务上。先看调用形态：2026 年主流的视频生成 API 几乎都是异步三段式——提交任务拿到任务号、轮询状态直到成功或失败、下载产物 URL，同步直返的只有极短的轻量档；本章代码一律标写法示意，参数名以各官方文档为准。再过参数三件事：分辨率、时长、帧率——分辨率受模型档位约束（720p 与 1080p 档并行），时长普遍在几秒到十几秒档，帧率 24 与 30 常见；铁律是全片统一，混档的片段在剪辑段全部要二次转码，白白多花一遍钱。然后实现带指数退避的轮询骨架，覆盖排队与限流两态。最后给失败分类表：限流退避重试、超时查状态再定、内容审核拒绝改提示词——三类失败三种处置，不要一律无脑重试。

学习目标

• 画出提交、轮询、下载三段式调用形态，说出它与同步调用的成本差异。
• 列出分辨率、时长、帧率的常见档位（示意口径）与全片统一铁律的理由。
• 实现带指数退避、封顶等待的轮询函数。
• 按失败分类表处置限流、超时、审核拒绝三类故障。

一、调用形态：三段式

视频生成是重计算任务，API 形态几乎收敛为异步三段式：

 你的脚本                        生成服务
    │  1. POST 提交任务（prompt、参数）   │
    │ ─────────────────────────────▶   │
    │  2. 返回 task_id                  │
    │ ◀─────────────────────────────   │   （排队中……计算中……）
    │  3. GET 查询状态（每 5 秒一次）     │
    │ ─────────────────────────────▶   │
    │  4. succeeded + video_url         │
    │ ◀─────────────────────────────   │
    │  5. GET 下载 mp4 到 04_clips/     │
    │ ─────────────────────────────▶   │
​

# gen_client.py —— 三段式客户端骨架（写法示意，以各官方文档为准）
import time
from pathlib import Path

def submit_and_wait(client, prompt, params, max_wait=600):
    task_id = client.submit(prompt, **params)
    waited, interval = 0, 5
    while waited < max_wait:
        status = client.task_status(task_id)
        if status["state"] == "succeeded":
            return status["video_url"]
        if status["state"] == "failed":
            raise RuntimeError("生成失败：%s" % status.get("error", "未知原因"))
        time.sleep(interval)
        waited += interval
        interval = min(interval * 2, 60)      # 指数退避，封顶 60 秒
    raise TimeoutError("等待超过 %d 秒" % max_wait)

def download(url, shot_id):
    out = Path("build/04_clips") / ("shot_%03d.mp4" % shot_id)
    out.parent.mkdir(parents=True, exist_ok=True)
    client_fetch(url, out)                     # 以你的 HTTP 库实现
    return out
​

三段式的工程含义：一次生成从「一次调用」变成「一个状态机」，失败点多了排队超时与轮询中断两处；好处是长任务不占连接，任务号可以落盘——生成段的断点续跑天然成立（重启后先查任务状态，而不是重新提交重新花钱）。

二、参数三件事：分辨率、时长、帧率

「全片统一」是本节唯一要划重点的参数纪律。分辨率混档的代价在剪辑段显形：ffmpeg 统一分辨率要一遍转码，画质与时间双重损失；帧率混档更隐蔽，24 与 30 混拼会导致运动不流畅，修复同样要重采样。把这三个参数钉进 project.json 的 pipeline 段（1.3 节），生成任务的构造函数只从配置读——这是「配置驱动」在生成段的落点。

三、排队与轮询

轮询的节奏有讲究：起步 5 秒一查，每次翻倍，封顶 60 秒（上面的骨架已含）。节奏背后的账：查得太密，浪费请求配额还可能撞限流；查得太疏，任务早完成了还在傻等，批量生产时整批的总等待被拉长。批量场景再加一层并发控制：同时挂起的任务数封顶（如 4 个，示意口径），一个完成补一个提交——像食堂打饭窗口，窗口数固定，队伍自流转。

把并发控制与断点续跑落成代码，核心是把「任务号落盘」做实：

# batch_render.py —— 并发受控的批量生成与断点续跑（纯标准库骨架，client 以你的 SDK 为准）
import json
import threading
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

STATE = Path("build/03_assets/tasks_state.json")
LOCK = threading.Lock()
MAX_RUNNING = 4                    # 同时挂起的任务数上限（示意值）

def load_state():
    if STATE.exists():
        return json.loads(STATE.read_text(encoding="utf-8"))
    return {}

def save_state(state):
    with LOCK:
        STATE.write_text(json.dumps(state, ensure_ascii=False, indent=2),
                         encoding="utf-8")

def wait_task(client, task_id, max_wait=600):
    waited, interval = 0, 5
    while waited < max_wait:
        status = client.task_status(task_id)
        if status["state"] == "succeeded":
            return status["video_url"]
        if status["state"] == "failed":
            raise RuntimeError("任务 %s 失败：%s" % (task_id, status.get("error")))
        waited += interval
        interval = min(interval * 2, 60)
    raise TimeoutError("任务 %s 超过 %d 秒" % (task_id, max_wait))

def run_one(client, task, state):
    key = str(task["shot_id"])
    tid = state.get(key)
    if tid is None:                        # 断点续跑：没有旧任务号才提交新的
        tid = client.submit(task["prompt"], duration=task["duration"])
        state[key] = tid
        save_state(state)
    url = wait_task(client, tid)
    target = Path("build/04_clips") / ("shot_%03d.mp4" % task["shot_id"])
    client.fetch(url, target)
    state.pop(key, None)                   # 完成即出队，状态文件里只剩在途任务
    save_state(state)

def main(client, tasks):
    state = load_state()
    with ThreadPoolExecutor(max_workers=MAX_RUNNING) as pool:
        futures = [pool.submit(run_one, client, t, state) for t in tasks]
        for f in futures:
            f.result()                     # 取结果，让工作线程的异常在主线程显形
    print("批量完成：%d 个任务" % len(tasks))
​

四个要点。状态文件是断点续跑的锚：任务号先落盘再干活，进程被杀后重启，load_state 读到的旧任务号会先走查询而不是重新提交——钱不花第二遍。完成即出队：任务号从状态文件移除，文件里躺着的永远是在途任务，语义干净。锁保护读写：多线程写同一个 JSON 文件，不加锁会互相覆盖。f.result() 不可省：不取结果，工作线程的异常会被吞掉，主线程带着假绿退出——与第 5.1 节 ffmpeg 封装「错误不能吞」是同一条纪律。

四、失败分类与处置

两条纪律。同类同参不无脑重试：审核拒绝的根因在提示词，重试一百次也是拒绝。失败要留痕：每个失败任务的 task_id、参数、错误信息写进 build/04_clips/failures.json，第 8 章（后半本）的编排段会把它变成重跑清单，第 10 章（后半本）的成核算账也会读它。

拿一次六镜头批处理做故障走查（镜头与故障均为示意）。任务提交后：镜头 102 第一次轮询返回限流，退避 10 秒重试，第三次通过；镜头 203 直接返回内容审核拒绝，改写 prompt（把「手部特写撕裂纸张」具体化为「手部特写缓慢分开一张纸」）后重提通过；镜头 301 下载成功但用 ffprobe 查出时长为零，换种子重跑一次正常；其余三个一次通过。收工时 failures.json 长这样：

# failures.json —— 批处理收工时的失败台账（示意）
[
  {"shot_id": 102, "type": "限流", "retries": 3, "resolved": true},
  {"shot_id": 203, "type": "审核拒绝", "action": "改写 prompt", "resolved": true},
  {"shot_id": 301, "type": "产物异常", "action": "换种子", "resolved": true}
]
​

台账的价值在 resolved 这个字段：已解决的失败留给第 10 章（后半本）算通过率，未解决的失败变成第 8 章（后半本）编排段的重跑清单——同一份数据，两个消费方，这就是「失败留痕」的下游回报。

五、常见问题与排查

排查的总原则只有一条：先查状态，再做决定——三段式把「现在发生了什么」存在服务端，任何处置动作之前都值得多花一次查询。

本节要点回顾

• 主流 API 是提交、轮询、下载三段式；任务号落盘让生成段天然支持断点续跑。
• 分辨率、时长、帧率全片统一，参数只从 project.json 读。
• 轮询指数退避封顶 60 秒，批量加并发上限。
• 失败四分类四处置：限流退避、超时查态、拒绝改词、异常换种；失败留痕喂给下游。

能调通了，下一节解决「调出来的画面不听指挥」——首尾帧怎么锚定画面，运镜怎么用提示词驱动，多镜头的主角怎么才能长得一样。