Uncensored API 向けエラーとレート制限の実践プレイブック

エラーエンベロープ

すべての失敗は、単一のトップレベルのerrorオブジェクトを含むJSONボディを返します。重要なのは2つのフィールドです：code（短い機械用文字列）と、ログや人間向けのmessageです。

Copy{
  "error": {
    "code": "no_credit",
    "message": "Balance used up or trial expired."
  }
}

この順序で判断してください。まずHTTPステータスを確認し、問題のクラスを把握してください。同じステータスを共有するケースを区別するためにのみcodeを使用し、言葉の表記が変わる可能性があるため、ロジックのためにmessageを解析しないでください。成功レスポンスの形式については、APIリファレンスを参照してください。

ステータスコード、原因と対応策

ステータス	コード	一般的な原因	リトライ？	アクション
400	リクエストエラー	不正なJSON、またはプロンプトトークン数とmax_tokensの合計が100,000を超えた場合	いいえ	ペイロードを修正し、履歴を短くするか、max_tokensを減らしてください
401	無効なキー	ヘッダーの欠如、タイプミス、または再生成されたキー	いいえ	シークレットストアからキーを再読み込みしてください
402	no_credit	残高が枯渇したか、無料トライアルが終了しました	いいえ	前払い残高をチャージし、再開してください
403	content_blocked	リクエストに未成年者の性的コンテンツが含まれています	いいえ	再送信せず、独自のポリシーメッセージを表示してください
404	見つかりません	間違ったパス（例：/v1プレフィックスの欠落）	いいえ	ベースURLを確認してください
429	レート制限	1つのキーで1分間に300件を超えるリクエスト	はい	ジッター付きでバックオフし、クライアントのスロットリングを追加してください
503	upstream_busy	一時的な容量逼迫	はい	数秒後にリトライしてください

403は強調に値します：これは一時的な障害ではなくポリシーの決定です。同一のリクエストをリトライするとクォータが無駄になり、同じ回答が返されます。人々を最も驚かせる400のケースはコンテキストウィンドウの合計です。max_tokensが8,000の60,000トークンのプロンプトは4,000オーバーフローし、各数値が単独では有効であっても失敗します。

適切に動作する再試行ポリシー

小さく制限されたポリシーを使用し、再試行が役立つ場合にのみ適用してください。

1. 再試行可能な障害をホワイトリストに登録する： 429、503、接続のリセット、タイムアウト。それ以外は直ちにエラーを発生させます。
2. ジッター付き指数関数的バックオフ： 1、2、4、8 秒待機し、1 秒未満のランダムな分数を加算し、最大 30 秒に制限します。ジッターにより、多くのワーカーが同時に再試行するのを防ぎます。
3. 試行回数を 5 回に制限する。 それ以上の場合、障害の可能性の方が高いため、キューイングされた作業は明示的に失敗させるべきです。
4. 独自のループを実行する場合は SDK の自動再試行を無効にする（max_retries=0）。そうしないと試行回数が倍増し、スロットルがカウントを失います。
5. ストリームを最初から再試行する。 途中で失敗したストリームは再開できません。部分的なテキストを破棄してリクエストを再送信するか、そのままにして継続をリクエストしてください。

503の場合、ガイドラインは数秒待つことなので、すぐに攻撃するのではなく、最初の遅延を1〜2秒程度に設定してください。429の場合の解決策は速度を落とすことで、次のセクションでこれを自動化します。

1 分あたり 300 リクエスト以下に抑える

制限はキーごとに 1 分あたり 300 リクエストで、平均すると 1 秒あたり 5 リクエストです。トークンバケットはこれを正確に適用します。バケットは少量のトークンのバーストを保持し、1 秒あたり 5 つの割合で補充され、各リクエストが 1 つのトークンを消費します。バケットが空の場合、呼び出し元はトークンが生成されるまで待機します。

Python 版では asyncio を使用します。20 のタスクは 1 つのバケットとロックを共有するため、並行コルーチンがバケットを使い果たすことはありません。バケットは試行回数（再試行を含む）をすべてカウントすることに注意してください。

Copyimport asyncio, os, random, time
from openai import AsyncOpenAI, APIStatusError, APIConnectionError

class TokenBucket:
    """Allows `rate` acquisitions per `per` seconds, with a small burst."""
    def __init__(self, rate=300, per=60.0, burst=10):
        self.fill = rate / per          # tokens added per second (5.0)
        self.cap = burst
        self.tokens = burst
        self.stamp = time.monotonic()
        self.lock = asyncio.Lock()

    async def acquire(self):
        async with self.lock:
            while True:
                now = time.monotonic()
                self.tokens = min(self.cap, self.tokens + (now - self.stamp) * self.fill)
                self.stamp = now
                if self.tokens >= 1:
                    self.tokens -= 1
                    return
                await asyncio.sleep((1 - self.tokens) / self.fill)

client = AsyncOpenAI(base_url="https://api.uncensoredapis.com/v1",
                     api_key=os.environ["API_KEY"], max_retries=0)
bucket = TokenBucket()

RETRYABLE = {429, 503}

async def ask(prompt, attempts=5):
    for n in range(attempts):
        await bucket.acquire()
        try:
            r = await client.chat.completions.create(
                model="uncensored",
                messages=[{"role": "user", "content": prompt}],
                max_tokens=200,
            )
            return r.choices[0].message.content
        except APIStatusError as e:
            if e.status_code not in RETRYABLE or n == attempts - 1:
                raise
        except APIConnectionError:
            if n == attempts - 1:
                raise
        await asyncio.sleep(min(30, 2 ** n) + random.uniform(0, 1))

async def main():
    prompts = [f"Give one-line fact number {i} about tides." for i in range(1, 21)]
    results = await asyncio.gather(*(ask(p) for p in prompts))
    print(len(results), "answers; first:", results[0])

asyncio.run(main())

API_KEY をエクスポートして実行します。20 件のリクエストはバーストと数秒の補充内に収まるため、すぐに完了するのが確認できます。カウントを 600 に上げると、実行時間は約 2 分になり、これは制限が正常に機能していることを示しています。

Node 版は公式パッケージを ES モジュールとして使用し、プロミスキューを通じて取得をチェーン化し、呼び出し元を順番に処理します。

Copyimport OpenAI from "openai";

class TokenBucket {
  constructor(rate = 300, perMs = 60000, burst = 10) {
    this.fillPerMs = rate / perMs;
    this.cap = burst;
    this.tokens = burst;
    this.stamp = Date.now();
    this.queue = Promise.resolve();
  }
  acquire() {
    // Serialize callers so the bucket math never races.
    this.queue = this.queue.then(async () => {
      for (;;) {
        const now = Date.now();
        this.tokens = Math.min(this.cap, this.tokens + (now - this.stamp) * this.fillPerMs);
        this.stamp = now;
        if (this.tokens >= 1) { this.tokens -= 1; return; }
        await new Promise((r) => setTimeout(r, Math.ceil((1 - this.tokens) / this.fillPerMs)));
      }
    });
    return this.queue;
  }
}

const client = new OpenAI({
  baseURL: "https://api.uncensoredapis.com/v1",
  apiKey: process.env.API_KEY,
  maxRetries: 0,
});
const bucket = new TokenBucket();

async function ask(prompt, attempts = 5) {
  for (let n = 0; n < attempts; n++) {
    await bucket.acquire();
    try {
      const r = await client.chat.completions.create({
        model: "uncensored",
        messages: [{ role: "user", content: prompt }],
        max_tokens: 200,
      });
      return r.choices[0].message.content;
    } catch (err) {
      const retryable = err.status === 429 || err.status === 503 || err.status === undefined;
      if (!retryable || n === attempts - 1) throw err;
      await new Promise((r) => setTimeout(r, Math.min(30000, 2 ** n * 1000) + Math.random() * 1000));
    }
  }
}

const prompts = Array.from({ length: 20 }, (_, i) => `Give one-line fact number ${i + 1} about tides.`);
const answers = await Promise.all(prompts.map((p) => ask(p)));
console.log(answers.length, "answers; first:", answers[0]);

バーストを保守的に設定してください。バーストを 10 に設定すると、持続レートが 1 秒あたり 5 のまま短いスパイクを許容します。複数のプロセスが 1 つのキーを共有する場合は、レートをそれらで分割します（例：3 つのワーカーでそれぞれ 1 分あたり 100）。または、すべてのトラフィックを 1 つのキューにルーティングしてください。

プロダクトでの 402 の適切な処理

コードが no_credit の 402 は、前払い残高がゼロになったか、無料トライアルが期限切れになったことを意味します。これはエンドユーザーにスタックトレースとして表示すべきではない唯一のエラーであり、通常はトップアップなどあなたにしかできない修正を意味します。

3つのレイヤーを構築します。まず、detectを中央のラッパーで一元化し、呼び出し元が覚える必要がないようにします。次に、translateします：ブラウザやモバイルアプリはcapacity_pausedのような中立な状態を受け取り、親しみやすい画面を表示し、アップストリームのメッセージを直接表示しません。最後に、alertしてください。すべてのユーザーが同時に影響を受けるため、アカウント所有者に通知を送り、機能フラグを切り替えてUIがリクエストを送信しないようにします。

Copy# server side: translate the upstream 402 into a product state
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from openai import AsyncOpenAI, APIStatusError
import os

app = FastAPI()
client = AsyncOpenAI(base_url="https://api.uncensoredapis.com/v1", api_key=os.environ["API_KEY"])

@app.post("/api/reply")
async def reply(body: dict):
    try:
        r = await client.chat.completions.create(
            model="uncensored", messages=body["messages"], max_tokens=400)
        return {"text": r.choices[0].message.content}
    except APIStatusError as e:
        if e.status_code == 402:
            # Tell the client app to show a "temporarily unavailable" screen,
            # and page yourself to top up the prepaid balance.
            return JSONResponse({"state": "capacity_paused"}, status_code=503)
        if e.status_code == 429 or e.status_code == 503:
            return JSONResponse({"state": "busy_retry"}, status_code=503)
        raise

予防的な保護も追加してください。リクエストごとの usage.total_tokens を独自の台帳に合計し、選択した閾値で警告を出してください。これにより、ユーザーがブロックされる前にトップアップが行われます。トライアルクレジットは $0.50 で 7 日間有効であるため、お金が残っていても期限切れだけで 402 に到達する可能性があります。計画のためのレートは 価格ページ に記載されています。

残高が再び正になったら、一時停止中にキューに入れた作業を再生します。ラッパーは最初の成功時に一時停止フラグをクリアすべきです。

ログとアラートの対象

適切な記録により、インシデントを短く抑えることができます。リクエストごとに、独自のリクエスト ID、存在する場合は completion id、HTTP ステータス、エラー code、試行回数、トークン使用量をログに記録します。Authorization ヘッダーや個人データを含む完全なプロンプトをログに記録しないでください。

優先度順の有用なアラート：任意の402（即時）、401の継続的な発生（再生成されたAPI キーがすべてにデプロイされていない）、呼び出しの数％以上の429レート（スロットルサイズが不適切）、503のクラスター（より広くバックオフ、キューを検討）。1分あたりのステータス数のダッシュボードにより、これら4つを一目で確認できます。適切に処理されれば、これらのコードは障害ではなく通常のシグナルになります。

ローンチ前に障害パスをテストする

実行されたことがないエラー処理は推測にすぎません。安価なリクエストと独自のテストダブルを使用して、実際の障害を待たずにほとんどの分岐を実行できます。

• 401: で GET /v1/models を呼び出し、意図的に間違ったキーを使用して、アプリが顧客ではなくオペレーターに構成エラーを表示することを確認します。
• /v1/chat/completion（単数形）に対して1回だけ404:リクエストを実行し、ログ行がパスを指定していることを確認してください。これにより、Base URLのタイプミスは数分で明らかになります。
• 400: でプロンプトを約 50,000 トークンにパディングして max_tokens を 32,000 に送信します。合計が 100,000 を超えるため、検証パスが起動するのを見ることができます。
• 429と503：これらを要求に応じて発生させることはできないため、スタブしてください。HTTPクライアントをラップし、最初の2回の呼び出しでステータスを返すフェイクを作成し、その後成功するようにし、ループが待機、リトライ、そして最終的な回答を返すことをアサートしてください。
• 402:これもスタブし、UIが一時停止画面を表示し、アラートが発生し、ループ処理で何もリトライしないことを確認してください。

これらを継続的インテグレーションで実行される自動化テストとして記述してください。上記の表の各ステータスにつき1行のテーブル駆動テストは、記述に数分しか要せず、ポリシーを善意のリファクタリングから保護します。最後に、APIをスタブした状態で自分のスロットルに対して短い負荷テストを実行してください：1,000回の取得を発火し、60秒のウィンドウ内に300の許可とバーストを超えるものがないことをアサートしてください。これにより、実際のトラフィックがそれに基づいて依存する前に算術が証明されます。後でエージェントループガイドを読む際、エージェントがリクエスト数を増幅し、これらの制限に最初に到達するため、同じラッパーを再利用してください。