如何通过 HiAPI API（curl + Python）调用 veo-3.1-fast/image-to-video

本教程构建什么，以及需要先准备什么

本教程将完整演示一次可运行的 HiAPI 集成，使用 veo-3.1-fast/image-to-video：你发送一张首帧图片和一段提示词，收到一个异步任务，轮询直到它完成，然后下载生成的视频及其原生音频。

到教程结束时，你将拥有两个等效实现——一个可在终端运行的 curl 命令，以及一个可放入脚本或服务的 Python 客户端——两者都访问同一个 HiAPI 端点。

前提条件

在开始之前，请准备以下内容：

• 一个 HiAPI API 密钥。 不要将其提交到源代码管理中。在下面的示例中，它从环境变量读取，而不是直接内联粘贴。
• HiAPI 基础 URL。 使用你的 HiAPI 仪表板或 API 参考中显示的地址；将其存储为变量，以便只需配置一次。
• 一张首帧图片。 公开可访问的 URL 是最简单的输入。使用你有权使用的真实图片，采用 JPEG 或 PNG 等常见格式。
• 一个 HTTP 客户端。 对于本演示，curl 就足够了。对于 Python 版本，你只需要 requests——无需供应商 SDK。
• 一个本地目录，用于存储下载的视频。

设置你的环境

导出密钥和基础 URL，以便 shell 和 Python 示例都能复用它们：

export HIAPI_API_KEY="your-api-key"
export HIAPI_BASE_URL="https://api.hiapi.ai"
export FIRST_FRAME_URL="https://example.com/your-first-frame.jpg"

veo-3.1-fast/image-to-video 的确切端点路径和参数名称记录在 HiAPI API 参考中模型条目旁边——请在那里查看规范字段列表，而不是依赖从其他地方复制的片段。本教程侧重于请求结构和工作流程，即使字段细节需要对照文档确认，这些内容也保持不变。

现在值得牢记一点：这是一个基于任务的模型，而不是同步模型。你不会在提交响应中直接获得视频。你会获得一个任务标识符，视频稍后到达。

图像生视频请求的结构

在编写任何代码之前，理解你实际发送什么以及返回什么会有所帮助。图像生视频生成比文本生成更重，因此 API 将其建模为后台任务。

输入：提示词和首帧

一个 veo-3.1-fast/image-to-video 请求有两个概念上的部分：

• 一个文本提示词，描述运动、镜头行为和场景意图——即首帧之后应该发生什么。
• 一张参考图片，用于锚定输出的首帧。这就是图像生视频与文本生视频的区别：模型从你的图像继续，而不是凭空创造一个开场帧。

模型名称本身告诉了你所做的权衡：fast 变体针对图像生视频任务的更快周转进行了调优，这使其非常适合迭代，以及延迟比最高保真度更重要的流水线。

生成的输出包含原生音频，因此最终文件是一个带声音的完整视频，而不是一个需要单独配乐的无声片段。

任务生命周期

由于生成需要时间，交互分为三个阶段：

1. 提交。 你将提示词和图像 POST 到模型端点。响应是一个轻量确认，包含任务标识符——而不是视频。
2. 轮询。 你使用该标识符反复 GET 任务状态，直到它报告完成（或失败）。
3. 获取。 一旦完成，任务响应会指向生成的视频，你下载并存储它。

在代码中保持这些阶段分离是最重要的结构性决策。将它们混在一起，重试、超时和错误处理都会变得模糊不清。

一个格式正确的提交是什么样

无论确切的序列化方式如何，请规划一个包含以下内容的请求：

• 模型标识符 veo-3.1-fast/image-to-video。
• 你的提示词文本。
• 首帧图片引用。
• 通过请求头中的 API 密钥进行身份验证。

其他所有内容——可选参数、默认值、输出分辨率选项——都属于该模型的 HiAPI API 参考。阅读一次，将确认的字段名放入常量中，不要将它们分散到代码库中。

将提交响应视为收据。记录它，存储任务 ID，然后继续进行轮询。

使用 curl 发起调用

让我们提交一个任务。下面的命令将包含提示词和首帧图片的 JSON 正文发送到 HiAPI 上的模型端点，并使用环境中的密钥进行身份验证。

curl -sS -X POST "$HIAPI_BASE_URL/v1/videos/generations" \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "veo-3.1-fast/image-to-video",
    "prompt": "Slow dolly forward through a sunlit forest path, leaves drifting, soft birdsong ambience",
    "image": "'"$FIRST_FRAME_URL"'"
  }'

关于此命令的几点说明：

• 端点路径和 image 字段名应对照 veo-3.1-fast/image-to-video 的 HiAPI API 参考进行检查；如果不同，请替换为规范值。这里的结构——模型、提示词、图像——才是关键。
• 提示词确实在发挥作用。明确描述运动和镜头移动（“缓慢向前推轨”“树叶飘动”），并提及你想要的音频氛围，因为模型会同时生成画面和原生音频。
• -sS 保持输出安静，但仍会显示错误，这正是你在终端演示中想要的。
• 图像以 URL 字符串形式传递。如果你更喜欢上传字节，请查看参考中支持的输入形式，而不是猜测。

读取提交响应

成功的提交返回一个紧凑的 JSON 对象，包含任务标识符和状态——大致如下：

{
  "id": "job_abc123",
  "status": "queued"
}

字段名仅为示例；请对照实际响应确认。重要的是你现在有了一个可用于轮询的 ID，以及一个不是已完成视频的状态。

在 shell 中捕获该 ID，以便后续步骤更简单：

JOB_ID="job_abc123"

如果你收到身份验证错误，说明你的密钥缺失或格式错误。验证错误通常意味着请求正文与文档化的模式不匹配——首先检查字段名和类型。速率限制响应意味着你发送得太频繁；应退避，而不是在紧密循环中重试。

Python 中的同一请求

以下是 Python 中的等效提交。它仅使用 requests，从环境变量读取配置，并将提交步骤与轮询分离，以便你可以复用每一部分。

import os
import time
import requests

API_KEY = os.environ["HIAPI_API_KEY"]
BASE_URL = os.environ["HIAPI_BASE_URL"].rstrip("/")
FIRST_FRAME_URL = os.environ["FIRST_FRAME_URL"]

HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

MODEL = "veo-3.1-fast/image-to-video"

def submit_job(prompt: str, image_url: str) -> str:
    payload = {
        "model": MODEL,
        "prompt": prompt,
        "image": image_url,
    }
    resp = requests.post(
        f"{BASE_URL}/v1/videos/generations",
        headers=HEADERS,
        json=payload,
        timeout=60,
    )
    resp.raise_for_status()
    data = resp.json()
    job_id = data["id"]
    print(f"submitted job: {job_id} (status={data.get('status')})")
    return job_id

if __name__ == "__main__":
    job_id = submit_job(
        prompt="Slow dolly forward through a sunlit forest path, "
               "leaves drifting, soft birdsong ambience",
        image_url=FIRST_FRAME_URL,
    )

为什么采用这种结构

• 配置放在常量中。 基础 URL、请求头、端点路径和模型名称只定义一次。当 API 参考告诉你字段名不同时，你只需改一行。
• timeout 是显式的。 永远挂起的提交比快速失败更糟。对于小的 JSON 正文，六十秒很充裕。
• raise_for_status() 尽早暴露失败。 在 4xx 和 5xx 时你会得到异常，而不是静默地将错误正文解析为任务。
• 函数只返回任务 ID。 调用者不需要知道提交负载是什么样；他们只需要一个用于轮询的东西。

如果你的应用程序是 Web 服务，请在请求处理器中调用 submit_job，并将返回的 ID 交给后台工作进程。不要在视频渲染时阻塞 HTTP 请求。

轮询异步任务直到视频就绪

现在任务已存在并排队。轮询是你了解它何时完成的方式。

基本的轮询循环

def poll_job(job_id: str, timeout_s: int = 900, initial_delay: float = 5.0) -> dict:
    url = f"{BASE_URL}/v1/videos/generations/{job_id}"
    delay = initial_delay
    deadline = time.time() + timeout_s

    while time.time() < deadline:
        resp = requests.get(url, headers=HEADERS, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        status = data.get("status")
        print(f"status={status}")

        if status == "succeeded":
            return data
        if status in ("failed", "cancelled"):
            raise RuntimeError(f"job {job_id} ended with status={status}: {data}")

        time.sleep(delay)
        delay = min(delay * 1.5, 30.0)  # back off, cap at 30s

    raise TimeoutError(f"job {job_id} did not finish within {timeout_s}s")

为什么每一部分都存在

• 轮询，而非 webhook。 最简单可靠的集成是按计划询问状态。Webhook 在大规模下更好，但需要公共端点；轮询在任何地方都有效。
• 带上限的指数退避。 早期轮询成本低且频繁；后期轮询间隔拉大。上限防止长任务上间隔失控。从大约五秒开始并逐渐增长。
• 显式的终止状态。 成功是一种状态；失败和取消是其他状态。将其他任何情况视为“继续等待”，但永远不要无限等待。
• 实际时间截止。 与文本相比，视频生成很慢。选择一个与你的产品耐心相匹配的宽松总超时，并在超过时干净地失败，而不是永远循环。
• 状态日志。 打印或记录每次状态转换。当出现问题时，这是你首先想要的东西。

从 shell 轮询

在调试时进行快速手动检查，一次 curl 调用就足够了：

curl -sS "$HIAPI_BASE_URL/v1/videos/generations/$JOB_ID" \
  -H "Authorization: Bearer $HIAPI_API_KEY"

每隔几秒重新运行它，或将其包裹在一个小的 while 循环中，直到状态翻转为终止状态。对照 HiAPI API 参考确认确切的字段和路径名——生命周期是稳定的，但线上细节才是事实来源。

一旦 poll_job 返回，响应应包含对已完成视频的引用。这就是你接下来要下载的内容。

下载并验证输出

任务已完成，响应指向生成的文件。下载它是一个普通的 HTTP GET，但要做好两件事：将其流式写入磁盘并验证你得到的内容。

获取视频

import os
from urllib.parse import urlparse

def download_video(job: dict, out_dir: str = "outputs") -> str:
    video_url = job["output"]["url"]  # confirm the field path in the API reference
    os.makedirs(out_dir, exist_ok=True)

    parsed = urlparse(video_url)
    ext = os.path.splitext(parsed.path)[1] or ".mp4"
    dest = os.path.join(out_dir, f"{job['id']}{ext}")

    with requests.get(video_url, stream=True, timeout=120) as r:
        r.raise_for_status()
        with open(dest, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 16):
                f.write(chunk)

    size = os.path.getsize(dest)
    if size == 0:
        raise IOError(f"downloaded file is empty: {dest}")
    print(f"saved {dest} ({size} bytes)")
    return dest

说明：

• 分块流式传输。 视频文件很大；使用 r.content 将其加载到内存中是浪费的，并可能在资源受限的主机上失败。
• 保留扩展名，以便播放器和下游工具识别容器格式。
• 按任务 ID 命名。 它唯一、可追踪，并在运行批次时避免冲突。
• 拒绝空文件。 零字节下载就是失败的下载，即使状态码显示 200。

如果视频 URL 是签名的或有时间限制，请在任务完成后立即下载，而不是存储 URL 以备后用。

验证它是带有音频的真实视频

文件存在并不等于文件可用。检查它：

• 在本地播放，使用媒体播放器并确认画面和声音。 veo-3.1-fast/image-to-video 生成原生音频，因此无声结果是一个危险信号，值得在到达用户之前调查。
• 检查容器，如果你安装了 ffmpeg，使用 ffprobe：

ffprobe -hide_banner outputs/job_abc123.mp4

在输出中查找视频流和音频流，以及合理的持续时间。这也是捕获截断文件的最快方法。

• 对首帧进行基本检查。 它应该类似于你的输入图像。如果不是，在怀疑模型之前，先怀疑图像 URL 或你用于传递它的字段。

一旦下载验证通过，将文件持久化到持久存储，并同时记录任务 ID。这种映射使后续调试、重新运行和计费对账成为可能。

扩展前的陷阱和检查

异步视频 API 的大多数集成问题不是模型问题。它们是管道问题。以下是你应该首先检查的内容。

图像输入

• 图像真的可访问吗？ 在你的浏览器中有效的 URL 可能需要身份验证、阻止热链接或会过期。优先使用稳定、公开可读的 URL。
• 它是支持的格式吗？ 除非 API 参考另有说明，否则坚持使用 JPEG 和 PNG 等常见格式。
• 内容正确吗？ 首帧锚定整个片段。低质量或偏离主题的图像会产生令人失望的视频，再多的提示词调整也无法修复。

请求结构

• 字段名和类型。 对照 veo-3.1-fast/image-to-video 的 HiAPI API 参考进行验证。从其他平台复制的片段会失败——端点是 HiAPI 的，字段名区分大小写。
• 内容类型。 发送 Content-Type: application/json 和实际的 JSON 正文，而不是表单数据。
• 模型标识符。 使用完整的 veo-3.1-fast/image-to-video 字符串；截断的名称会返回看似无关的错误。

超时和重试

• 每个阶段有独立的超时。 提交、轮询请求和文件下载各自值得拥有。单一的全局超时要么对下载太紧，要么对状态检查太松。
• 安全地重试。 提交可以在瞬态网络错误和 5xx 响应上重试，但前提是你能判断任务是否实际创建——否则你可能会双重计费和双重生成。轮询和下载是幂等的，可以自由重试。
• 在速率限制时退避。 遵守响应并放慢速度，而不是猛击端点。

轮询行为

• 永远不要在紧密循环中轮询。 它浪费配额，看起来像滥用。
• 始终有一个终止条件。 一个会抛出异常的截止时间比一个卡在内存中的任务更好。
• 记录每次状态转换的时间戳。你需要它来推断延迟。

运维卫生

• 将 API 密钥保留在服务器端。 永远不要将它们发送到浏览器或嵌入移动应用中。
• 将任务 ID 与每个工件一起存储。 它是你跨日志、文件和使用记录的连接键。
• 在发布前验证下载。 检查大小，理想情况下检查流。到达用户的空文件或无声文件，其修复成本远高于你重试的失败下载。
• 不要假设字段路径。 完成任务响应中的 output 位置正是那种会变化的细节——从 API 参考中读取它，而不是从教程中，包括本教程。

在针对模型提交工单之前，按顺序检查此列表。大多数情况下，修复方法是 URL、字段名或缺少退避。

这就是完整的循环：使用首帧图像和提示词提交，轮询任务直到完成，然后下载并验证带有原生音频的视频。从 HiAPI 上的 veo-3.1-fast/image-to-video 模型页面开始，确认当前的请求字段，然后将上面的两个函数提升到你自己的服务中。