HiAPI で Veo 3.1 Lite のイメージ・トゥ・ビデオを使う：curl と Python のチュートリアル

veo-3.1-lite/image-to-video にできること

veo-3.1-lite/image-to-video は、最初のフレームを動画に変換するモデルです。静止画像を渡すと、そのフレームをアニメーション化して短い動画を生成します。指定した画像はスタイルの参考でも、曖昧なインスピレーションの源でもありません——出力動画の文字通りの冒頭フレームです。そのため、構図、被写体の位置、フレーミングは運動の最初の数秒へ直接引き継がれます。

この設計上の選択が、入力をどう捉えるべきかを決めます。明るく、シャープで、適切にクロップされた画像は、期待した位置から始まる動画を生み出します。ぼやけた画像や妙なフレーミングの画像は、ぼやけて奇妙なフレーミングの冒頭から始まり、モデルが運動を生成するにつれてさらにずれていく動画を生み出します。このモデルをうまく使いこなすための作業の大半は、コードを書く前に完了しています。

出力側では、モデルはネイティブ音声付きの動画を返します。これが旧式のイメージ・トゥ・ビデオ・パイプラインとの違いです——旧来のフローでは、まず無音の映像を生成し、別途独立した音声トラックを後から合成していました。ここでは音声は同じタスクの一部として生成されるため、環境音と視覚的なシーンが一貫性を保ちます——手動でのミックスは不要で、効果音とそれに対応する動きの間にタイミングのずれも生じません。

生成は非同期です。タスクを送信して識別子を受け取り、完了するまでポーリングします。これは意図的なアーキテクチャ上の選択です。動画生成は一般的な HTTP リクエストのタイムアウトをはるかに超える時間がかかるため、API は送信と完了を切り離しています。このチュートリアルのすべての例は、この「まず送信、次にポーリング」という形式に従っています。

要するに、1 枚の画像を入力し、音声付きの動画を出力し、非同期タスクとして配信する。このメンタルモデルを覚えておけば、API の残りの部分はごく素直です。

最初のリクエストを送る前に

初回の呼び出しを成功させるには、3 つのものを用意する必要があります。

API キー

HiAPI.ai へのすべてのリクエストは API キーで認証されます。アカウントのダッシュボードで作成し、ソースファイルに直接貼り付けるのではなく環境変数から読み込みます：

export HIAPI_API_KEY="your_api_key_here"

キーが誤ってリポジトリにコミットされたことがある場合は、直ちにローテーションしてください。キーの交換コストはわずかですが、漏洩の代償は大きいです。

アクセス可能な最初のフレーム画像

モデルは自分で画像を取得するため、画像は URL 経由でアクセス可能でなければなりません——パブリックなオブジェクトストレージのリンク、CDN パス、またはインタラクティブな認証を必要としない任意の HTTP エンドポイントです。有効期限の長い署名付き URL は問題ありませんが、ログインセッションが必要な URL は不可です。PNG や JPEG などの一般的な形式を使用し、妥当な解像度を選んでください。クリアで適度なサイズの画像は、帯域を無駄にせず、モデルにきれいなディテールを提供します。

ソース画像が内部サービス内にある場合は、まずオブジェクトストレージにアップロードし、生成された URL を渡してください。これが初回の試みが失敗する最も一般的な原因です。

エンドポイントパス

このモデルは veo-3.1-lite/image-to-video で識別されます。この文字列はタスクとともに送信するモデル識別子として扱い、十数か所にハードコードしないでください——定数に入れておけば、後でモデルのバリアントを切り替える可能性が高いからです。

キーをエクスポートし、取得可能な画像 URL を用意し、モデル識別子を把握すれば、送信の準備は整います。

動作する curl リクエスト

送信呼び出しは生成タスクを作成します。完全に動作する例は次のとおりです：

curl -X POST "https://api.hiapi.ai/v1/video/generations" \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "veo-3.1-lite/image-to-video",
    "image": "https://your-bucket.example.com/first-frame.jpg",
    "prompt": "朝の光がシーンの中をゆっくりと移動し、カメラがゆっくりと前進する",
    "duration": 5
  }'

項目ごとに見ていきます：

• URL は HiAPI の生成エンドポイントです。すべてのモデルのトラフィックはこの 1 つのインターフェースを通るため、後でモデルを切り替えるにはエンドポイントではなく model フィールドを変更するだけです。
• Authorization ヘッダーはベアラートークンを運びます。$HIAPI_API_KEY という記法は環境変数から読み込むため、キーがシェルの履歴に残ることはありません。
• model は veo-3.1-lite/image-to-video を識別します。
• image は最初のフレームの URL です。このフィールドがあるからこそ、この呼び出しはテキスト・トゥ・ビデオではなくイメージ・トゥ・ビデオになります。
• prompt はオプションですが、強く推奨されます。モデルはすでにその画像を見ているため、静止画の説明ではなく、望む運動——カメラワーク、光の変化、被写体の動き——を記述します。
• duration はリクエストする動画の長さ（秒）で、モデルがサポートする範囲に制限されます。

レスポンスは id を含むタスクオブジェクトです。保存しておいてください：

{
  "id": "job_abc123",
  "status": "queued",
  "model": "veo-3.1-lite/image-to-video"
}

ステータスが queued または processing であることは想定どおりの結果です——まだ動画は得られておらず、約束を得ただけです。次のセクションでは、その約束をファイルに変える方法を説明します。

同じ呼び出しを Python で書く

この curl 呼び出しは requests を使って直接 Python に翻訳できます。次のバージョンは送信を小さな関数にまとめ、パイプラインで再利用できるようにしています：

import os
import requests

API_KEY = os.environ["HIAPI_API_KEY"]
BASE_URL = "https://api.hiapi.ai/v1"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

def submit_video_job(image_url: str, prompt: str, duration: int = 5) -> str:
    payload = {
        "model": "veo-3.1-lite/image-to-video",
        "image": image_url,
        "prompt": prompt,
        "duration": duration,
    }
    response = requests.post(
        f"{BASE_URL}/video/generations",
        headers=HEADERS,
        json=payload,
        timeout=30,
    )
    response.raise_for_status()
    data = response.json()
    print(f"Submitted job {data['id']} ({data['status']})")
    return data["id"]

if __name__ == "__main__":
    job_id = submit_video_job(
        image_url="https://your-bucket.example.com/first-frame.jpg",
        prompt="朝の光がシーンの中をゆっくりと移動し、カメラがゆっくりと前進する",
    )

いくつかの細部は残す価値があります：

• raise_for_status() は非 2xx レスポンスを早期に例外に変えます。これがないと、認証エラーや検証エラーが、成功したが予期しないレスポンスボディを返した呼び出しのように見えてしまいます。
• 30 秒のタイムアウトは送信にのみ適用されます。送信はすぐに戻るはずで、長時間の処理はサーバー側で行われます。送信自体がタイムアウトするなら、問題は生成速度ではなく、接続性かリクエストペイロードのサイズにあります。
• タスク id を返すことで、レスポンス全体ではなく関数を組み合わせ可能に保てます——そのタスクをどう扱うかは呼び出し側が決めます。

これで、この関数は変更なしでキューワーカー、Flask エンドポイント、またはノートブックのセルに組み込めます。

非同期タスクをポーリングする

動画生成には数ミリ秒ではなく数分かかります。人間の感覚に近い間隔でタスクステータスエンドポイントをポーリングし、タスクが終端状態に達したらすぐに停止します：

import time
import requests

def wait_for_video(job_id: str, timeout: int = 600, interval: int = 10) -> dict:
    deadline = time.time() + timeout
    url = f"{BASE_URL}/video/generations/{job_id}"
    while time.time() < deadline:
        response = requests.get(url, headers=HEADERS, timeout=30)
        response.raise_for_status()
        data = response.json()
        status = data["status"]
        print(f"Job {job_id}: {status}")
        if status == "completed":
            return data
        if status == "failed":
            raise RuntimeError(f"Generation failed: {data}")
        time.sleep(interval)
    raise TimeoutError(f"Job {job_id} did not finish within {timeout}s")

3 つの実践により、本番環境で確実に動作します：

1. 指数バックオフ。短い間隔から始めて徐々に増やし、妥当な上限を設定します。最初のポーリングは通常まだ queued を返すため、最初の数秒間にエンドポイントを連打しても得るものはありません。
2. ハードタイムアウト。総待機時間には常に上限を設定します。processing で止まったタスクは、無限ループではなくアラートを発するエラーとして現れるべきです。
3. タスク id を永続化する。ポーリングを開始する前にデータベースに書き込みます。プロセスが待機中にクラッシュした場合、再起動後にポーリングを再開でき、同じ作業に対して二重に課金されることも、再送信する必要もありません。

completed と failed のみをループを終了する状態と見なします。それ以外——queued、processing——は待機を続けることを意味します。

もう 1 点：このコードを Web サーバー上で実行する場合、リクエストハンドラー内でポーリングしないでください。送信してタスク id をクライアントに返し、バックグラウンドワーカーに待機を任せます。そうしないと、遅い生成が数分間接続を占有します。

出力動画をダウンロードする

タスクが完了すると、ステータスレスポンスに完成した動画の URL が含まれます。バイナリデータとしてダウンロードし、適切な拡張子でディスクに書き込みます：

def download_video(job: dict, path: str = "output.mp4") -> str:
    video_url = job["output"]["url"]
    response = requests.get(video_url, headers=HEADERS, timeout=120, stream=True)
    response.raise_for_status()
    with open(path, "wb") as f:
        for chunk in response.iter_content(chunk_size=1 << 20):
            f.write(chunk)
    print(f"Saved {path}")
    return path

いくつかの点を正しく行う必要があります：

• バイナリモードで書き込む。"wb" ではなく "w" で開くとファイルが静かに破損します。ダウンロードは成功したように見えても、動画は再生できません。
• 大きなファイルをストリーミング処理する。上記の stream=True とチャンク単位のコピーにより、動画のサイズに関係なくメモリを安定させられます。
• チェックサムを計算する——出力を大規模にアーカイブする場合。これは 1 行のコードですが、切り詰められたダウンロードがユーザーに届く前に捕捉できます。

音声について：返される動画にはすでにネイティブ音声トラックが含まれています。取得すべき別の音声ファイルも、ミックスするステップもありません。下流でファイルを再ラップする場合——たとえばコンテナ形式を変更する場合——は、音声ストリームも一緒に持っていってください。よくある間違いは、デフォルトで音声を剥がす変換を実行し、音声がどこに行ったのかと首をかしげることです。

最後に、出力 URL は通常、有効期限があります。URL を数日間保存してから取得するのではなく、できるだけ早くダウンロードして自分のバケットに保存してください。その頃にはリンクが期限切れになっている可能性があります。

よくある失敗ポイント

イメージ・トゥ・ビデオ呼び出しのトラブルシューティングで最も時間を費やすエラーを以下に示します：

画像 URL にアクセスできない

これは断トツで最も一般的な失敗原因です。モデルはサーバー側で画像を取得するため、ログイン済みのブラウザで機能する URL は何の意味もありません。タスクを送信する前に、認証なしの普通のリクエストでテストしてください。署名付き URL の有効期間は、生成時間より大幅に長くする必要があります。

モデルのエラーに見える認証エラー

Authorization ヘッダーの欠落や形式の誤りは通常、明確な 401 を返しますが、時折、開発者が初期化コードをステータスコードを飲み込むロジックで包み込み、その後「モデルの故障」と報告することがあります。リクエストペイロードをデバッグする前に、生のステータスコードを確認してください。

ポーリングが頻繁すぎる、またはまったくしない

100 ミリ秒ごとにポーリングするとクォータを浪費し、レート制限を引き起こす可能性があります。一度だけポーリングして processing を見て諦めると、タスクが成功したのに誰も知らないということになります。上記のバックオフとタイムアウトのパターンに従ってください。

送信を完了と勘違いする

送信の成功はタスクを返すのであって、動画ではありません。送信レスポンスから url フィールドを読み取るコードは何も見つけられません。出力を運ぶのは完了ステータスのレスポンスだけです。

プロンプトが画像を説明している

「窓辺に座る猫」のようなプロンプトは、モデルがすでに見ている内容を繰り返しているだけです。運動を記述するプロンプト——「雨が降り始めるとカメラがゆっくりと引いていく」——のほうが良い結果を生みます。内容ではなく、変化を記述してください。

クラッシュ時にタスク id を失う

送信と完了の間にプロセスが死んだ場合、永続化されていないタスク id は作業が復元できないことを意味します。永続化し、ワーカーから非同期にポーリングしてください。小さな設計上の選択が、最も長いデバッグの過程を避けられます。