HiAPI API（curl + Python）で veo-3.1-fast/image-to-video を呼び出す方法

このチュートリアルで構築するもの、事前に準備するもの

このチュートリアルでは、veo-3.1-fast/image-to-video を使った実行可能な HiAPI 統合を一通り実演します。先頭フレームの画像とプロンプトを送信し、非同期タスクを受け取り、完了するまでポーリングし、生成された動画とそのネイティブ音声をダウンロードします。

チュートリアルを終える頃には、2 つの等価な実装が手に入ります。ターミナルで実行できる curl コマンドと、スクリプトやサービスに組み込める Python クライアントです。どちらも同じ HiAPI エンドポイントにアクセスします。

前提条件

始める前に、以下を準備してください：

• HiAPI API キー。 ソースコード管理にコミットしないでください。以下の例では、直接インラインで貼り付けるのではなく、環境変数から読み取ります。
• HiAPI ベース URL。 HiAPI ダッシュボードまたは API リファレンスに表示されているアドレスを使用し、一度だけ設定すれば済むように変数として保存します。
• 先頭フレームの画像。 公開アクセス可能な URL が最も簡単な入力です。JPEG や PNG などの一般的な形式で、使用権のある実在の画像を使用してください。
• HTTP クライアント。 このデモでは curl で十分です。Python 版では requests だけが必要で、ベンダー SDK は不要です。
• ローカルディレクトリ（ダウンロードした動画を保存するため）。

環境のセットアップ

シェルと Python の両方の例で再利用できるよう、キーとベース URL をエクスポートします：

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 リクエストには、概念的には 2 つの部分があります：

• テキストプロンプト：動き、カメラの挙動、シーンの意図、つまり先頭フレームの後に何が起こるべきかを記述します。
• 参照画像：出力の先頭フレームを固定するためのものです。これが画像から動画とテキストから動画の違いです。モデルはゼロから冒頭フレームを作り出すのではなく、あなたの画像から続行します。

モデル名自体がトレードオフを物語っています。fast バリアントは画像から動画へのタスクの迅速なターンアラウンドに最適化されており、反復作業や、最高の忠実度よりも遅延が重要なパイプラインに最適です。

生成される出力にはネイティブ音声が含まれるため、最終ファイルは音声付きの完全な動画であり、別途音楽を当てる必要のある無音のクリップではありません。

タスクのライフサイクル

生成には時間がかかるため、やり取りは 3 つの段階に分かれます：

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 が得られたこと、そして状態が完了した動画ではないことです。

後続の手順を簡単にするため、シェルでその 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 リファレンスでフィールド名が異なると判明したら、1 行変更するだけで済みます。
• timeout を明示。 永遠にハングする送信は、素早く失敗するよりも悪いです。小さな JSON ボディには 60 秒で十分です。
• 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 は大規模では優れていますが、公開エンドポイントが必要です。ポーリングはどこでも機能します。
• 上限付きの指数バックオフ。 初期のポーリングは低コストで頻繁に行い、後期は間隔を広げます。上限により、長いタスクで間隔が暴走するのを防ぎます。約 5 秒から始めて徐々に増やしてください。
• 明示的な終了状態。 成功は 1 つの状態であり、失敗とキャンセルは別の状態です。それ以外は「待ち続ける」とみなしますが、決して無限に待ってはいけません。
• 実時間の締め切り。 テキストに比べて動画生成は遅いです。製品の忍耐度に合った寛大な全体タイムアウトを設定し、超えたら永遠にループするのではなく、きれいに失敗させてください。
• 状態のログ出力。 各状態遷移を出力または記録してください。問題が発生したとき、まず欲しくなる情報です。

シェルからのポーリング

デバッグ中の素早い手動確認には、1 回の curl 呼び出しで十分です：

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

数秒ごとに再実行するか、状態が終了状態に切り替わるまで小さな while ループで包んでください。正確なフィールド名とパス名は HiAPI API リファレンスで確認してください。ライフサイクルは安定していますが、実際の通信上の詳細が信頼できる情報源です。

poll_job が返ったら、レスポンスには完了した動画への参照が含まれているはずです。それが次にダウンロードするものです。

出力をダウンロードして検証する

タスクは完了し、レスポンスは生成されたファイルを指しています。ダウンロードは通常の HTTP GET ですが、2 つのことを確実に行ってください。ディスクへストリーミング書き込みし、取得した内容を検証します。

動画を取得する

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 モデルページから始めて現在のリクエストフィールドを確認し、上記の 2 つの関数を自分のサービスに取り込んでください。