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))

境界を越えた冪等性

• 送信：送信をリトライするときは常に同じ冪等キーを送ります。これが、すでにタスクを作成したかもしれないリクエストのタイムアウトから回復する唯一の安全な方法です。キーを一度生成し、意図したリクエストとともに保存し、再利用します。
• ポーリング：本質的に冪等——読み取り専用。時間予算内で自由にリトライできます。
• コールバック：配信は本質的に at-least-once です。タスク 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)

送信前の検証

予期せぬ過剰支出のほとんどは、送るべきでないリクエストから来ます：機能に必要な長さをはるかに超える duration パラメータ、まだ実行中のタスクを再送信するリトライループ、クライアントの不具合を引き起こす空のプロンプト。リクエストがプロセスを離れる前に、パラメータ——duration、解像度、プロンプト長、入力メディア参照——を検証し、明確なエラーで早く拒否しましょう。

エラーだけでなくタスク量を記録する

毎回の送信とそのパラメータ、リクエスト時間、ユーザー、冪等キーを記録します。機能別・ユーザー別に周期的なタスク数を集計します。集計されたタスク量こそが支出を本当に追跡する指標であり、異常なパターン——ある呼び出し元が通常の十倍のレートで生成している、など——をまだ小さいうちに捕捉できます。

同じタスクに二度課金しない

重複生成の失敗モードには三つのよくある原因があり、それぞれに防御策があります：

• 冪等キーなしで送信タイムアウトをリトライする。そのキーを再利用すること。
• Webhook の再配信が新しいダウンロードや新しいタスクを引き起こす。タスク ID で重複排除し、遷移を条件付きにすること。
• ポーラーが諦めて、元のタスクがまだ実行中なのに再送信する。代替タスクを作る前に既存タスクの状態を検証し、タスクテーブルで置換関係を追跡すること。

すべての再送信を、デフォルトの回復経路ではなく、理由を記録する必要のある決定として扱いましょう。

リリース前チェックリスト

統合をリリースする前に、これらの決定と失敗モードをエンドツーエンドで検証します。

取得とライフサイクル

• 取得モードはワークロードで選択し、ハイブリッドのフォールバックを備えて、逃したコールバックでタスクが失われないようにする。
• 送信後、他のどの副作用よりも先にタスク行を書き、パラメータ、冪等キー、所有者を記録する。
• すべてのタスクはちょうど一つの終態に達する。陳腐化しきい値が、まだ保留中のあらゆるものに決定を強制する。
• 状態遷移は現在の状態を条件とするため、重複配信や並行ポーリングは無害である。

ネットワークの挙動

• すべてのリクエストに接続と読み取りのタイムアウトを設定する（ダウンロードを含む）。
• 一時的エラーと終端エラーを分類し、一時的エラーだけをリトライする。
• リトライはジッタ付き指数バックオフを使い、試行回数と経過時間にハード上限を設ける。
• ポーリングのバックオフには間隔の上限と実時間予算の両方があり、ポーリングループはリクエストハンドラではなくワーカーで動く。

コールバックとストレージ

• コールバックエンドポイントは生のボディに対して送信元を検証し、素早く返し、処理をキューに入れる。
• すべてのコールバックは、何か行動する前に自分のタスク記録と突き合わせられる。
• 完成したアセットは自分のストレージへストリーミングされ、検証され、自分の URL からのみ配信される。ダウンロードステップは冪等である。

コストと運用

• 送信は同時実行制限と、duration、解像度、入力に対するパラメータ検証によってゲートされる。
• タスク量は機能別・ユーザー別に記録され、支出が突然ではなくトレンドとして見える。
• 重複生成の経路——送信リトライ、Webhook 再配信、ポーラーによる再送信——はすべて明示的に処理される。
• しきい値を超えても保留中のままのタスクと、継続的に同時実行制限に達している状況についてアラートがある。

各パスを通して実タスクを一つずつ実行しましょう——成功、失敗、逃したコールバック、重複コールバック——そして終態、ストレージ書き込み、カウンタが正しく見えることを確認します。上記のすべてがその演習で成り立てば、統合は HiAPI.ai 動画生成 API に対してリリース可能です。