AIエージェントと長時間APIコール：ポーリング vs ウェブフック

エージェント向け非同期処理APIの設計とテスト

エージェントが動画トランスコードのエンドポイントを呼び出すと、エンドポイントは 202 Accepted とジョブIDを返します。しかし、エージェントが 202 の意味を理解していないと、トランスコードが完了したと誤認し、まだ存在しないファイルを読み込もうとします。

今すぐApidogを試す

時間のかかる操作は、エージェント特有の問題を引き起こします。同期呼び出しには「送信し、待ち、回答を得る」という明確な契約があります。一方、非同期呼び出しは処理を開始と終了に分割するため、その間にエージェントが混乱します。

その結果、次のような問題が起こります。

• 早期に成功を宣言する
• 密なループで何千回もポーリングする
• 会話のターンを開いたまま数分間ブロックする
• ジョブIDを忘れる

この記事では、エージェントが従いやすい非同期契約、ポーリングとWebhookの使い分け、ツールのレスポンス設計、低速・失敗ケースのテスト方法を説明します。エージェントのエラー回復に関する当社の投稿がAPI呼び出しの失敗を扱うのに対し、この記事では「時間はかかるが最終的には成功する」ケースを扱います。

Apidogは、4分かかって失敗するジョブをエージェントが処理できるか検証する場合に役立ちます。本番環境で初めて発見するべき問題ではありません。

なぜエージェントは非同期処理を誤るのか

主な原因は、次の3つです。

2xxを完了と解釈する

202 は、リクエストが処理のために受け付けられたことを示します。処理が完了したことを意味しません。HTTPセマンティクス仕様にも、処理が完了していない可能性が明記されています。

通常の要求・応答トラフィックで学習したモデルは、レスポンスが明示的に否定しない限り、任意の 2xx を成功と読み取る傾向があります。

ポーリングが高コストになる

エージェントの推論ループ内でポーリングすると、各チェックがモデルのターンと過去の会話トークンを消費します。

4分間のジョブを2秒ごとに確認すると、120ターン必要です。コンテキストを使い切ったり、予算を超過したりする可能性があります。コンテキストウィンドウからツール応答を除外する方法については、こちらの記事で詳しく説明しています。

ジョブを見失う

ジョブを開始してジョブIDを返すツールは、エージェントが後で引き継ぐべき状態を作ります。長い会話の途中にジョブIDが現れると、コンテキスト圧縮によって失われ、進行中のジョブ自体を忘れることがあります。

モデルが誤読しにくいレスポンスを設計する

最も効果的な対策は、アーキテクチャよりも明確な言葉遣いです。ステータスコードだけに頼らず、レスポンスボディに「何が起きたか」と「次に何をすべきか」を記載します。

{
  "status": "processing",
  "job_id": "job_7f21c",
  "message": "The transcode has STARTED and is NOT complete. Do not report success. Check status with getJobStatus(job_id) after at least 30 seconds.",
  "poll_after_seconds": 30,
  "estimated_duration_seconds": 240,
  "status_url": "/v1/jobs/job_7f21c"
}

人間向けAPIとしては過剰に見えるかもしれません。しかし、モデルはステータスコードから意味を推測するより、レスポンスボディに書かれた明示的な指示に従いやすい傾向があります。

特に次の3点が重要です。

1. 「未完了」と明記する
2. 次に呼び出すツールを指定する
3. 最小待機時間を示す

Googleの長時間実行操作に関するAIP-151は、done、error、response フィールドを持つ単一の Operation オブジェクトを定義しています。この形を採用すれば、遅いエンドポイント全体で一貫したインターフェースを提供できます。

ステータスレスポンスも、直接的な形式にします。

{
  "job_id": "job_7f21c",
  "status": "processing",
  "done": false,
  "progress_percent": 45,
  "elapsed_seconds": 108,
  "poll_after_seconds": 45,
  "message": "Still processing. Do not proceed to the next step."
}

完了時に結果が小さい場合は、インラインで返します。これにより、エージェントが不要な3回目の呼び出しを行わずに済みます。

{
  "job_id": "job_7f21c",
  "status": "succeeded",
  "done": true,
  "result": { "output_url": "https://cdn.example.com/out/7f21c.mp4", "duration_seconds": 372 }
}

ポーリングはモデルの外側で行う

最も重要な実装上の選択は、待機処理をエージェントの推論ループではなく、ツールのラッパー内に置くことです。

import time

def start_and_await_transcode(client, source_url, max_wait=600):
    job = client.post("/v1/transcode", json={"source_url": source_url}).json()
    job_id = job["job_id"]
    delay = job.get("poll_after_seconds", 5)
    waited = 0

    while waited < max_wait:
        time.sleep(delay)
        waited += delay
        status = client.get(f"/v1/jobs/{job_id}").json()

        if status.get("done"):
            if status["status"] == "succeeded":
                return {"status": "succeeded", "result": status["result"]}
            return {"status": "failed", "error": status.get("error")}

        delay = min(int(delay * 1.5), 60)

    return {
        "status": "timed_out",
        "job_id": job_id,
        "message": f"Still running after {max_wait}s. Job {job_id} continues in the background.",
    }

モデルから見ると、これは時間のかかる処理を実行して最終結果を返す、1回のツール呼び出しです。コンテキスト内のポーリングループ、忘れられたジョブID、120ターンの消費は発生しません。

バックオフによってリクエスト数を抑え、上限によって停止したジョブが実行を永久にハングアップすることを防ぎます。数値を調整する前に、タイムアウト、リトライ、ジッター付きバックオフに関するAmazonの解説も参照してください。

安全に運用するには、次のルールを守ります。

• 待機時間には必ず上限を設定する
• タイムアウト時には必ずジョブIDを返す
• 曖昧な結果を返さない
• succeeded、failed、timed_out を別の状態として扱う

分単位ではなく時間単位で実行されるジョブでは、ラッパー内ポーリングは適切ではありません。その場合は、開始用と確認用の2つのツールを用意し、進行中のジョブを会話の外部に永続化します。

保存すべき情報は次のとおりです。

• job_id
• 所属するタスク
• 開始時刻

エージェントには、各実行の開始時にこの一覧を読み取らせます。コンテキスト圧縮が起きても、進行中のジョブを失わずに済みます。

Webhookを使うべきケース

ポーリングは単純で、どこでも利用できます。Webhookは効率的ですが、実装に必要な要素が増えます。Webhookとポーリングの比較も参考にしてください。

ポーリングを使う場合

次の条件ではポーリングが適しています。

• ジョブが数秒から数分で完了する
• エージェントが結果を待ってから続行する
• パブリックエンドポイントをホストできない

多くのエージェントワークロードはこの範囲に収まります。

Webhookを使う場合

次の条件ではWebhookを検討します。

• ジョブが数時間かかる
• エージェントがジョブ開始後に次の処理へ進む
• 多数のジョブが同時に実行され、個別のポーリングが無駄になる

Webhookでは、次の仕組みが必要です。

• パブリックな受信エンドポイント
• 署名検証
• リトライ処理
• コールバック到着時にエージェントを起動する仕組み

信頼性の高いWebhookの設計方法と、Webhook署名検証のガイドが基礎を説明しています。

SSEという選択肢

サーバー送信イベント（SSE）でジョブの進行状況をストリーミングすれば、クライアントが接続を保持したままプッシュ型の通知を受け取れます。パブリックなWebhookエンドポイントを用意できない、対話型エージェントに適しています。

実装方法は、SSEでAPIレスポンスをストリーミングする方法を参照してください。

完了処理は冪等にする

ポーリングでは競合が起き、Webhookではリトライが発生します。「成功」を2回受け取っても、後続ステップを2回実行してはいけません。

AIエージェントの冪等性キーを使い、完了処理を安全に再実行できるようにします。

高速パスだけでなく低速パスもテストする

非同期処理のバグは、テスト環境が速すぎると隠れます。本番で4分かかるジョブがローカルスタブでは200ミリ秒で完了すると、エージェントは実際の状態を経験できません。

次の4つのシナリオを意図的に作成します。

1. 実際に遅いジョブ

ステータスエンドポイントをモックし、最初の数回は processing、その後に succeeded を返します。ラッパーがポーリングし、バックオフし、最終的に結果を返すことを検証できます。

Apidogでは、リクエスト数や制御パラメーターに応じて変化するモックを作成できます。同じテストを毎回同じ条件で実行できます。

2. 遅れて失敗するジョブ

processing を3回返した後、エラーボディとともに failed を返します。エージェントが、ポーリングが完了したこととジョブが成功したことを混同せず、失敗として報告することを確認します。

これは、エラーを誤って成功として扱った場合に、サイレントなデータ損失につながるケースです。

3. タイムアウト

ラッパーの上限を超えて processing を返し続けます。ツールが例外や偽の成功ではなく、ジョブIDを保持した timed_out を返すことをアサートします。

4. 重複する完了

Webhookのリトライや競合するポーリングによって、成功通知を2回渡します。ダウンストリームのステップが1回だけ実行されることを確認します。

この4つのシナリオを保存し、CIで実行してください。再実行のコストが低いため、誰かがタイムアウトを短縮したり、エラーを飲み込んだりする回帰を検出できます。

より広範なテスト戦略については、API契約テストのガイドを参照してください。

問題を露呈させる3種類のジョブ

レポート生成

財務エージェントが四半期レポートをリクエストし、生成に90秒かかるとします。

素朴なツールでは、エージェントはジョブIDを受け取ると、レポートが準備できたと発表し、壊れたダウンロードリンクを返します。

ブロッキングラッパーなら、90秒待ってから実際のURLを返します。同じAPIでも、待機場所を変えるだけで結果は正反対になります。

一括インポート

運用エージェントが20,000件のレコードをアップロードし、8分後に14,000行目で部分的に失敗するとします。

ジョブ自体は終了しているため done は true になります。しかし、結果には拒否された行の一覧が含まれます。部分的な結果をカウントとともに明示し、エージェントが次に進む前に確認できるようにします。

モデルとビルドパイプライン

トレーニング実行やCIビルドに40分かかる場合、ラッパー内ポーリングは不適切です。会話のターンを長時間ブロックしてしまいます。

この場合は、次の流れにします。

1. ジョブを開始する
2. IDを永続ストレージに記録する
3. 会話のターンを終了する
4. スケジュールされたチェックやコールバックで後続処理を起動する

マルチエージェントの引き継ぎとコンテキスト受け渡しでは、実行間で状態を失わずに移動する方法を説明しています。

部分的な結果に明確な形を与える

長時間ジョブは、完全な成功と完全な失敗の中間で終了することがあります。成功と失敗の2状態だけでは、実際の結果を正しく表現できません。

3番目の状態を明示します。

{
  "job_id": "job_a11f",
  "status": "completed_with_errors",
  "done": true,
  "summary": { "processed": 20000, "succeeded": 19860, "failed": 140 },
  "errors_url": "/v1/jobs/job_a11f/errors?limit=50",
  "message": "Import finished. 140 rows failed and were not written. Review errors before reporting success."
}

このペイロードのポイントは2つです。

• 件数をインラインで返すため、追加の呼び出しなしで判断できる
• 失敗した行の詳細は制限付きURLに置き、140個のエラーオブジェクトでコンテキストを圧迫しない

滞ったジョブには所有者を割り当てる

タイムアウト時には、ジョブIDと「まだ実行中なので後で確認する」というメッセージを返します。これは正しい戻り値ですが、人に届かなければ役に立ちません。

エージェントが自社サービスなら、チームが監視しているキューへルーティングします。割り当てられたタスクを処理するコーディングランタイムなら、実行プラットフォーム上の適切な場所へ記録します。

Sharklyでは、ブロックされて終了した実行が、その実行状態と結果とともにタスクに残ります。受信トレイでは、人間の返信やレビューが必要な項目を通常の更新から分離できます。

重要なのは特定のツールではありません。「まだ実行中。後で確認してください」という状態には所有者が必要です。所有者がいなければ、実質的には「誰も確認しなかった」状態になります。

チェックリスト

• すべての遅いエンドポイントが、ジョブID、ステータスURL、未完了であることを示す平易なメッセージを返す
• ステータスレスポンスに、文字列だけでなくブール値の done フィールドを含める
• ポーリングを、指数バックオフと厳密な上限を持つツールラッパー内で実行する
• タイムアウト時にジョブIDを返し、後から再開できるようにする
• 成功、失敗、タイムアウトを別の戻り値として扱う
• 数分を超えるジョブを会話の外部に記録する
• ポーリングとコールバックのどちらで完了しても、処理を冪等にする
• 遅いジョブ、遅れて失敗するジョブ、タイムアウト、重複完了のテストを保存する

レスポンスの文言とラッパーを適切に設計すれば、長時間実行される操作はエージェントにとって特別なケースではなくなります。

「ツールを呼び出し、待ち、回答を得る」という契約は、エージェントが最も扱いやすい形です。テストと並行して低速ジョブのモックを作成するには、Apidogをダウンロードしてください。

よくある質問

非同期開始の場合、APIは202または200を返すべきですか？

202 Accepted は、処理が完了していないことを標準的なクライアントへ正直に伝えるコードです。

ただしンドポイントの現実的な最悪ケースより少し長い上限を設定します。一般的には2〜10分です。

それを超える場合は、ラッパーで会話を長時間ブロックせず、「後で確認する」ツールへ切り替えます。

どのポーリング間隔を使うべきですか？

サーバーが poll_after_seconds を返す場合は、その値から開始します。その後は約1.5倍ずつバックオフし、上限を約60秒に設定します。

固定の1秒ポーリングはリクエストを無駄にし、レート制限にかかる可能性があります。詳しくはレート制限超過ガイドを参照してください。

エージェントは待機中に別の作業をできますか？

オーケストレーターが並行ツール呼び出しをサポートしている場合に限り可能です。その場合は、ジョブを開始し、独立した作業を実行してからステータスを確認します。

サポートしていない場合は、手動で作成したスケジューラより、ブロッキングラッパーの方がシンプルでエラーも少なくなります。

エージェントが早期に成功を主張するのを防ぐには？

次の対策を組み合わせます。

• レスポンスボディに未完了であることを明記する
• ブール値の done フィールドを公開する
• 完了ツールを、結果が表示される唯一の場所にする
• 開始レスポンスに結果を含めない

開始レスポンスに結果がなければ、モデルがそれを成功結果として報告することもできません。

ラップトップ上で実行しているエージェントでもWebhookを使えますか？

パブリックエンドポイントがないため、直接は利用できません。

開発時はトンネルを使うか、エージェントがアドレス指定可能な環境で動作するまでポーリングを使います。WebhookサービスでローカルホストAPIをテストするガイドも参照してください。