API Timeoutの診断と対策：長文LLM生成における安全な再試行と接続設定

切断の発生箇所：HTTPリクエストの4つのフェーズ

LLM APIへのリクエストは4つの独立した段階で構成されており、それぞれに適したタイムアウト設定が必要です：

[クライアント] --- (1. Connect Timeout) ---> [API Gateway]
[クライアント] --- (2. Write Timeout)   ---> [プロンプト送信]
[モデル]       --- (3. Read / TTFT)     ---> [推論と回答生成]
[クライアント] <--- (4. Pool Timeout)    --- [ソケットプール待機]

1. Connect Timeout（接続タイムアウト）：TCP接続とTLSハンドシェイクの完了待機時間（5〜10秒推奨）。
2. Write Timeout（送信タイムアウト）：10万トークンを超える大規模コンテキスト送信時の所要時間。
3. Read Timeout（読み取りタイムアウト）：初回トークン返却（TTFT）およびSSEストリーム間隔の待機時間。
4. Pool Timeout（プールタイムアウト）：高並行リクエスト時にクライアントの空きソケットを待機する時間。

---

タイムアウト診断マトリクス

---

Python（HTTPX）での粒度の高いタイムアウト設定

標準ライブラリの既定値（10秒）では推論モデルの呼び出しで切断が頻発します。以下の安定した接続設定を推奨します：

import os
import httpx
from openai import OpenAI

API_KEY = os.environ.get("BETTERTOKEN_API_KEY", "your_api_key_here")

custom_timeout = httpx.Timeout(
    connect=5.0,    # ネットワーク不通時の高速フォールバック
    read=120.0,     # 長文推論に対応する十分な待機時間
    write=10.0,     # プロンプト送信時間
    pool=10.0       # ソケット獲得待機時間
)

http_client = httpx.Client(
    timeout=custom_timeout,
    limits=httpx.Limits(max_keepalive_connections=50, max_connections=100)
)

client = OpenAI(
    base_url="https://www.bettertoken.ai/v1",
    api_key=API_KEY,
    http_client=http_client
)

response = client.chat.completions.create(
    model="claude-3-7-sonnet-20250219",
    messages=[
        {"role": "system", "content": "You are a senior systems architect."},
        {"role": "user", "content": "Design a high-throughput distributed message broker."}
    ],
    stream=True
)

for chunk in response:
    delta = chunk.choices[0].delta.content or ""
    print(delta, end="", flush=True)
もっと見る

---

SSEストリームの切断防止と安全な再試行設計

接続切断時の二重課金やサーバー負荷を防ぐため、再試行処理には以下の3つの原則を適用します：

1. 受信開始後の再試行を避ける：既に一部のトークンを受信している場合、全プロンプトを再送すると二重課金となります。
2. ランダム指数バックオフ（Full Jitter）：接続失敗時の再試行間隔をランダムに分散させて同時リクエスト集中を防ぎます。
3. 冪等性キーの活用：バッチジョブでは一意のタスクIDを付与して重複実行を防止します。

import time
import random
import httpx

def execute_with_safe_retry(client, model, messages, max_retries=3):
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                stream=True
            )
            return response
        except (httpx.ConnectTimeout, httpx.ReadTimeout, httpx.NetworkError) as err:
            if attempt == max_retries - 1:
                raise err
            sleep_time = random.uniform(0, base_delay * (2 ** attempt))
            time.sleep(sleep_time)

動作確認項目

• ステータスコード 200 OK の確認。
• ストリーミングが途中で途切れることなく全トークンを受信完了できること。
• 通信完了後にソケットが正常にプールへ解放されること。

詳細なAPI仕様はBetterToken API Referenceをご覧ください。