AI APIタイムアウト：原因、リトライパターン、フォールバック設計

# 問題：大きな入力 + 高いmax_tokens = 遅いレスポンス
response = client.chat.completions.create(
    model="gpt-4o",
    messages=very_long_messages,  # 100K+ tokens
    max_tokens=4096  # 長い出力を要求
)

# 解決策：入力を減らすか出力を制限
response = client.chat.completions.create(
    model="gpt-4o",
    messages=trimmed_messages,  # 50Kトークンに削減
    max_tokens=1024  # 短い出力
)

# 同期 — クライアントが完全なレスポンスを待つ間にタイムアウトする可能性
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    max_tokens=2048
)

# ストリーミング — 最初のトークンがより早く到着し、接続を維持
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    max_tokens=2048,
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

import asyncio
import httpx

async def generate_video_async(client, prompt: str, timeout: int = 600):
    """動画生成ジョブを送信し、完了までポーリングする。"""

    # ステップ1：ジョブを送信
    submit_response = await client.post(
        "/v1/video/generations",
        json={"model": "veo-3-fast", "prompt": prompt}
    )
    job_id = submit_response.json()["id"]

    # ステップ2：完了までポーリング
    for _ in range(timeout // 5):  # 5秒ごとに確認
        status_response = await client.get(f"/v1/video/generations/{job_id}")
        status = status_response.json()

        if status["status"] == "completed":
            return status["result"]
        elif status["status"] == "failed":
            raise RuntimeError(f"Generation failed: {status.get('error')}")

        await asyncio.sleep(5)

    raise TimeoutError(f"Video generation did not complete within {timeout}s")

Status: queued → position 42
Status: queued → position 15
Status: processing → estimated 90s remaining
Status: completed → download URL available

import asyncio
import random

async def retry_with_timeout_awareness(
    make_request,
    max_retries: int = 3,
    base_timeout: float = 30.0
):
    """タイムアウトタイプを意識したリトライ。"""
    for attempt in range(max_retries):
        try:
            return await asyncio.wait_for(
                make_request(),
                timeout=base_timeout * (1.5 ** attempt)  # リトライごとにタイムアウトを延長
            )
        except asyncio.TimeoutError:
            if attempt == max_retries - 1:
                raise

            # サンダリングハード効果を防ぐためにジッターを追加
            delay = min(30, (2 ** attempt) + random.uniform(0, 1))
            await asyncio.sleep(delay)

        except Exception as e:
            # タイムアウト以外のエラー：自動リトライしない
            if "429" in str(e):
                delay = min(60, (2 ** attempt) + random.uniform(0, 2))
                await asyncio.sleep(delay)
            else:
                raise

async def call_with_fallback(messages, client, primary_model, fallback_model):
    """プライマリモデルを試し、タイムアウト時にはより高速なモデルにフォールバック。"""
    try:
        return await asyncio.wait_for(
            client.chat.completions.create(
                model=primary_model,
                messages=messages
            ),
            timeout=30.0
        )
    except asyncio.TimeoutError:
        # より高速な（場合によってはより小さな）モデルにフォールバック
        return await client.chat.completions.create(
            model=fallback_model,
            messages=messages
        )

from openai import OpenAI

client = OpenAI(
    api_key="your-evolink-key",
    base_url="https://api.evolink.ai/v1"
)

# Smart Routerがプロバイダー選択とフォールバックを処理
response = client.chat.completions.create(
    model="evolink/auto",
    messages=messages
)