在 Python 中集成 AI 视频生成 API：轮询、Webhook 与成本控制

选择结果获取模式

与大多数 API 调用相比，视频生成速度很慢。一个生成几秒视频的请求可能需要几十秒到几分钟不等，这意味着创建任务的 HTTP 响应几乎从来不是包含视频的响应。在编写一行 Python 代码之前，先决定你将如何获取结果。有三种实用模式。

模式 1：提交并轮询

你发送一个生成请求，立即收到一个任务句柄，然后按固定间隔调用状态端点，直到任务达到终态。这是实现和调试最简单的模式：它就是普通的请求/响应代码，可以在任何防火墙后面工作，且没有入站网络要求。它的缺点是有延迟下限——结果的新鲜度只取决于你的轮询间隔——而且当轮询速度快于任务推进速度时会浪费调用。

轮询适合交互式工作负载：用户在 UI 中等待，任务预计在有界时间内完成，而你希望有一条可以推理的单一代码路径。

模式 2：Webhook 回调

你提交任务时附带一个回调 URL，将控制权交还给调用方，并让平台在任务完成时把结果 POST 到你的端点。这完全消除了轮询，并给你近乎即时的通知。它还能在提交任务的进程重启后继续工作，因为投递由外部驱动。

代价是运维上的：你需要一个可公开访问的 HTTPS 端点，需要验证回调确实来自提供方，还需要处理回调永远不来的情况——因为网络会故障、端点会配置错误、载荷会丢失。

Webhook 适合长时间运行或批量工作负载：大型任务队列、异步流水线，以及任何为几分钟而保持连接或空转循环都是浪费的场景。

模式 3：带兜底的混合模式

大多数生产系统最终采用的模式：以 Webhook 作为主要信号，但保留一个缓慢的周期性对账，在超过陈旧阈值后仍标记为待处理的任务上进行轮询。如果回调被错过，轮询器会捕获它。如果轮询器延迟，回调通常先到达并将任务标记为完成。

按工作负载选择，而不是按代码库选择。交互式端点可以轮询；后台队列可以使用回调；共享的任务存储和终态逻辑对两者应完全相同，这样获取策略可以改变而不触及系统其余部分。

设置客户端并提交任务

提交骨架应保持精简。以下所有内容假设只有一个地方存放你的 API 密钥和基础 URL，以及一个将提示词加参数转换为任务句柄的函数。

配置与密钥处理

永远不要硬编码密钥。从环境变量读取，并在启动时如果缺失就大声失败，这样你就不会在第一个客户请求时才发现问题。

import os

BASE_URL = os.environ.get("HIAPI_BASE_URL", "https://api.hiapi.ai")
API_KEY = os.environ["HIAPI_API_KEY"]  # 启动时的 KeyError 是故意的

让密钥远离日志。如果你为调试记录请求，记录提示词和参数，但要对授权头进行脱敏。

单次提交调用

提交调用应只做一件事：验证你控制的参数，发送请求，并返回任务标识符。不要在其中启动轮询，也不要下载任何东西。

import requests

def submit_video_job(prompt: str, *, idempotency_key: str, **params) -> dict:
    payload = {"prompt": prompt, **params}
    resp = requests.post(
        f"{BASE_URL}/v1/video/generations",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
            "Idempotency-Key": idempotency_key,
        },
        json=payload,
        timeout=(5, 30),  # 连接、读取
    )
    resp.raise_for_status()
    return resp.json()

幂等键很重要。如果你的提交调用超时，你实际上不知道任务是否已创建。使用相同的键重试可以安全地重发，而不是生成——并为——重复任务付费。

应立即持久化的内容

提交返回的那一刻，在做任何其他事情之前，先向你自己的数据库写入一行：

• 任务 ID 和幂等键
• 你自己的内部请求 ID 以及它所属的用户或批次
• 确切的请求参数，以便你能重现或审计它
• 状态 pending 和一个时间戳

如果你的进程在提交后、持久化之前崩溃，任务在上游存在但没有本地记录。先持久化，再继续，可以让你的数据库成为你必须对账的内容的事实来源。

以正确的方式轮询结果

轮询循环很容易写得很糟。朴素版本——一个没有休眠的 while True——会消耗配额、猛打 API，并让你的日志难以阅读。正确的循环有四个属性：间隔、退避、硬上限和终态检查。

循环

import time, random

def poll_job(job_id: str, *, max_attempts: int = 60, base: float = 2.0, cap: float = 20.0):
    for attempt in range(1, max_attempts + 1):
        resp = requests.get(
            f"{BASE_URL}/v1/video/generations/{job_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=(5, 15),
        )
        resp.raise_for_status()
        body = resp.json()
        status = body["status"]

        if status == "succeeded":
            return body
        if status in {"failed", "cancelled"}:
            raise RuntimeError(f"任务 {job_id} 以 {status} 结束：{body.get('error')}")

        delay = min(cap, base * (2 ** (attempt - 1)))
        time.sleep(delay + random.uniform(0, 1.0))

    raise TimeoutError(f"任务 {job_id} 在 {max_attempts} 次尝试后仍为待处理")

为什么每一部分都重要

退避。生成任务分阶段推进：排队、运行、渲染。在早期排队阶段每秒轮询会二十次返回相同答案。从几秒开始并翻倍直到上限，可以使调用量与任务持续时间成比例，而不是与你的耐心成比例。

抖动。添加随机的零点几秒可以防止你运行的每个客户端在部署或重试风暴后同步轮询。

最大尝试次数和挂钟时间预算。没有尝试次数上限的指数退避可能运行数小时。根据预期的最坏情况任务持续时间选择一个上限，并将超过它视为你要暴露的失败，而不是一个继续静默运行的循环。还要跟踪经过时间——一旦间隔上限生效，仅凭尝试次数会产生误导。

终态检测。明确区分成功、失败和取消。失败的任务不应由轮询器重试；它应带着提供方的错误报告，以便你决定是否用不同参数重新提交。

还有一条规则：不要在 Web 请求处理器内部轮询。将循环移到后台工作者，这样慢任务就不会占用你的请求线程池。

通过 Webhook 回调接收结果

Webhook 端点是一个公开的、默认未认证的 HTTP 路由，任何人都可以 POST 到它。请相应地对待它。

验证发送方

如果平台对回调签名，在解析 JSON 之前对原始请求体验证签名——签名检查针对字节运行，重新序列化载荷可能会改变它们。使用常量时间函数进行比较，并以非 2xx 状态拒绝不匹配。如果签名不可用，至少使用秘密路径段加上允许列表或共享头令牌，并且在将载荷与你创建的任务匹配之前，永远不要信任其内容。

快速返回，稍后工作

你的端点唯一的工作是确认收到。解析、验证、入队、响应。任何慢的事情——下载视频、在锁下调用数据库、调用下游服务——都在工作者中发生，而不是在处理器中。提供方会重试没有获得快速成功响应的回调，而慢处理器会把一个事件变成多个。

from flask import Flask, request, abort

app = Flask(__name__)

@app.post("/callbacks/video")
def video_callback():
    raw = request.get_data()
    if not verify_signature(raw, request.headers):
        abort(401)
    event = request.get_json(force=True)
    job_id = event.get("job_id")
    if not job_id:
        abort(400)
    queue.enqueue("process_video_event", event)  # 立即返回
    return {"ok": True}, 200

对照你的记录进行对账

在根据回调采取行动之前，在你自己的表中查找任务 ID。该查找会得出三种情况：

• 未知任务 ID：这可能是已删除任务的陈旧事件或伪造请求。记录并丢弃它。不要从入站载荷创建记录。
• 任务已标记完成：这是重复投递。确认它，不做其他事情。对同一任务重新下载和重新计费是经典的 Webhook 缺陷。
• 任务待处理：将其标记为完成（或失败），存储资产引用，并交给下载步骤。

使状态转换以当前状态为条件——即仅当任务仍为待处理时才成功的更新——这样两个并发投递不能同时获胜。将此与前面描述的混合轮询器结合，即使回调永远不来，也不会有任务丢失。

超时、重试与失败处理

重试是集成悄然出问题的地方。以下规则适用于提交、轮询和回调，但安全边界各不相同。

先对错误分类

瞬时：连接重置、5xx 响应、限流回复、读取超时。这些可能在第二次尝试时成功。

终止：4xx 验证错误、不支持的参数、认证失败，以及任务级失败（如提示词被拒绝）。重试这些只会重复相同的失败，对于已提交的任务，还会重复成本。

如有疑问，检查状态码和错误体，而不是默认重试一切。

带抖动的退避与硬停止

对瞬时错误使用带随机抖动的指数退避，并对尝试次数和总耗时都设上限。没有上限的重试策略是故障生成器：当依赖退化时，你的重试量会在它最承受不起的时候成倍增加负载。

def with_retries(fn, *, attempts: int = 4, base: float = 0.5, cap: float = 8.0):
    for i in range(attempts):
        try:
            return fn()
        except TransientError:
            if i == attempts - 1:
                raise
            delay = min(cap, base * (2 ** i))
            time.sleep(delay + random.uniform(0, delay / 2))

跨边界的幂等性

• 提交：重试提交时始终发送相同的幂等键。这是从可能已创建任务的请求超时中恢复的唯一安全方式。生成一次键，将其与预期请求一起存储，并复用。
• 轮询：天然幂等——只读。在你的时间预算内可以自由重试。
• 回调：投递本质上是至少一次。按任务 ID 加事件类型去重，并使处理器的状态转换有条件，这样重复投递就是无操作。

暴露终态

每个任务路径都应最终恰好存储一个结果：成功并带资产引用，或失败并带原因和尝试历史。永远不要让任务无限期停留在 pending——你的混合轮询器正是为了在陈旧阈值后强制做出决定而存在。将该状态暴露给消费任务的一方，以便上游代码可以决定重新提交、告警或退还关注，而不是永远等待。

下载并存储完成的视频

完成载荷中的 URL 是临时的。它可能被签名，可能过期，其主机不是你的应用在读取时应依赖的东西。集成的最后一步是将产物移入你控制的存储。

获取资产

使用流式请求下载，这样大文件不会作为单个 bytes 对象落入内存，并设置反映文件大小而非典型 API 调用的超时。

def download_asset(url: str, dest_path: str) -> None:
    with requests.get(url, stream=True, timeout=(5, 300)) as r:
        r.raise_for_status()
        with open(dest_path, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 20):
                f.write(chunk)

如果你的存储后端支持流式写入，直接将响应迭代器交给它，而不是暂存到本地文件。

信任之前先验证

至少检查响应状态成功、内容长度与任何声明的大小匹配、文件大小非平凡。如果完成载荷包含时长或格式元数据，抽查下载的产物是否可能是一个视频，而不是错误页面或被截断的部分文件。到达用户的损坏资产比一个你标记为失败并重新提交的任务更糟。

移入你自己的存储

将对象写入你的存储桶或 CDN 源站，键由你的内部任务 ID 派生，并在任务行上记录该键。从你自己的 URL 服务用户。好处会累积：

• 临时提供方 URL 永远不会泄露到你的前端、邮件或日志中。
• 过期不再重要；你可以按自己的时间表重新生成衍生品或重新编码。
• 保留、访问控制和删除成为你的策略，而不是提供方的。

保持下载步骤幂等：如果目标键已存在且有效，跳过获取。在重复回调时，这个检查正是防止第二次下载和第二次存储写入的原因。

控制生成成本

生成按请求的工作量计价——通常按输出时长或令牌等效单位计费，确切费率显示在平台的定价面板上。由于每次调用比典型文本请求贵得多，成本控制应放在调用点，而不是月度审查中。

限制并发

最有效的单一护栏是限制并发生成任务数。没有它，重试风暴、失控循环或突发用户请求可能以任何仪表盘都来不及显示的速度成倍增加支出。在提交前放置信号量或有界工作者池，并有意设置其大小。

from threading import Semaphore

_submit_gate = Semaphore(5)  # 根据你的预算和提供方限制调整

def guarded_submit(prompt, *, idempotency_key, **params):
    with _submit_gate:
        return submit_video_job(prompt, idempotency_key=idempotency_key, **params)

提交前验证

大多数意外超支来自本不应发送的请求：时长参数远高于功能所需、重试循环重新提交仍在运行的任务、空提示词触发客户端缺陷。在请求离开你的进程之前验证参数——时长、分辨率、提示词长度、输入媒体引用——并尽早以清晰错误拒绝。

记录任务量，而不仅仅是错误

记录每次提交及其参数、请求时长、用户和幂等键。按功能和按用户按周期统计任务数。聚合任务量是真正跟踪支出的指标，它让你能在异常模式——某个调用方以正常速率十倍生成——还小的时候捕获它们。

不要为同一任务付两次费

重复生成的失败模式有三个常见来源，每个都有防御措施：

• 提交超时重试时没有幂等键。复用该键。
• Webhook 重投递触发新的下载或新的任务。按任务 ID 去重并使转换有条件。
• 轮询器放弃并在原任务仍在运行时重新提交。在创建替代任务之前验证现有任务的状态，并在任务表中跟踪取代关系。

将每次重新提交视为需要记录原因的决定，而不是默认恢复路径。

上线前检查清单

在集成上线之前，端到端验证这些决策和失败模式。

获取与生命周期

• 按工作负载选择获取模式，并带有混合兜底，这样错过的回调不会丢失任务。
• 提交后在任何其他副作用之前写入任务行，记录参数、幂等键和所有者。
• 每个任务恰好达到一个终态；陈旧阈值强制对任何仍待处理的事项做出决定。
• 状态转换以当前状态为条件，因此重复投递和并发轮询是无害的。

网络行为

• 每个请求都设置连接和读取超时，包括下载。
• 对瞬时和终止错误分类，只重试瞬时错误。
• 重试使用带抖动的指数退避，并对尝试次数和经过时间设硬上限。
• 轮询退避既有间隔上限也有挂钟时间预算，轮询循环在工作器中运行而不是请求处理器中。

回调与存储

• 回调端点针对原始体验证发送方，快速返回，并将工作入队。
• 每个回调在采取任何行动之前都对照你自己的任务记录进行对账。
• 完成的资产流式传入你自己的存储，经过验证，并且仅从你的 URL 提供；下载步骤是幂等的。

成本与运维

• 提交受并发限制以及对时长、分辨率和输入的参数验证门控。
• 按功能和按用户记录任务量，因此支出作为趋势可见，而不是意外。
• 重复生成路径——提交重试、Webhook 重投递、轮询器重新提交——都得到明确处理。
• 对超过阈值仍卡在待处理的任务以及持续触及并发限制的情况存在告警。

通过每条路径运行一个真实任务——成功、失败、错过的回调、重复回调——并确认终态、存储写入和计数器看起来正确。如果上述每一项在该演练中都成立，集成就可以针对 HiAPI.ai 视频生成 API 上线了。