Uncensored API 错误与速率限制实用指南

错误信封

每次失败都会返回一个包含单个顶层 error 对象的 JSON 正文。两个字段很重要：code 是简短的机器字符串，message 是用于日志和人类阅读的文本。

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

按此顺序判断。先看 HTTP 状态码，因为它能告诉你问题的类别。仅在状态码相同的情况下使用 code 来区分情况，并且不要将 message 用于逻辑判断，因为措辞可能会更改。有关成功响应的形状，请参阅 API 参考。

状态码、原因与操作

状态	代码	典型原因	重试？	操作
400	请求错误	JSON 格式错误，或提示词 token 加上 max_tokens 超过 100,000	否	修复负载；截断历史消息或降低 max_tokens
401	密钥无效	缺少标头、拼写错误或密钥已重新生成	否	从密钥库重新加载密钥
402	no_credit	余额耗尽或免费试用已过期	否	为预付余额充值，然后恢复
403	content_blocked	请求涉及未成年人性内容	否	不要重新发送；显示你自己的策略消息
404	未找到	路径错误，例如缺少 /v1 前缀	否	检查基础 URL
429	速率限制	单个密钥在一分钟内超过 300 个请求	是	带抖动退避；添加客户端节流
503	upstream_busy	临时容量压力	是	几秒后重试

403 错误值得强调：这是一个策略决定，而非临时故障。重试相同的请求会浪费配额，并且会返回相同的答案。最让人惊讶的 400 错误情况是上下文总和。一个 60,000 token 的提示词加上 max_tokens 为 8,000 时，会溢出 4,000 并失败，尽管每个数字单独看都是有效的。

一个表现良好的重试策略

使用一个小型且有限制的策略，仅在重试有帮助的地方应用。

1. 将可重试的故障列入白名单： 429、503、连接重置和超时。其他所有错误应立即抛出。
2. 带随机退避的指数退避：等待 1、2、4、8 秒加上随机的小数秒，上限为 30。随机退避可防止许多工作进程同步重试。
3. 将尝试次数限制为五次。 超过此限制后，服务中断的可能性大于运气不佳，队列中的工作应明显失败。
4. 当你运行自己的循环时，否则（max_retries=0）禁用 SDK 自动重试，否则尝试次数会叠加，导致你的节流计数失效。
5. 从头开始重试流式输出。 中途失败的流式输出无法恢复；丢弃部分文本并重新发出请求，或者保留它并要求继续。

对于 503，建议等待几秒钟，因此将第一次延迟设置为一两秒左右，而不是立即频繁请求。对于 429，解决方法是减慢速度，下一节将自动化这一点。

保持在每分钟 300 个请求以下

限制是每个密钥每分钟 300 个请求，平均每秒五个。令牌桶精确执行该限制：它最多容纳一小批 token，每秒补充五个，每个请求消耗一个 token。如果桶为空，调用者会睡眠直到存在 token。

Python 版本使用 asyncio。所有二十个任务共享一个桶和一个锁，因此并发协程不会超额使用它。请注意，桶计算每次尝试，包括重试。

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 后运行它。二十个请求在突发量加上几秒的补充量范围内，因此你会看到它很快完成；将计数提高到 600，运行时间将拉伸至大约两分钟，这正是限制在发挥作用。

Node 版本使用 ES 模块中的官方包，通过 promise 队列链接获取，以便按顺序提供服务。

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 的突发允许短时间的尖峰，同时持续速率保持在每秒五个。如果几个进程共享一个密钥，请在它们之间分配速率，例如三个工作器每个每分钟 100，或者将所有流量路由到一个队列。

在产品中优雅地处理 402

带有代码 no_credit 的 402 错误意味着预付余额已耗尽或免费试用额度已过期。这是你的终端用户绝不应该看到的堆栈跟踪错误，它通常意味着只有你能修复的问题：充值。

构建三层架构。首先，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、完成时的 id（如果存在）、HTTP 状态、错误 code、尝试次数和 token 使用情况。永远不要记录 Authorization 标头或包含个人数据的完整提示词。

有用的警报，按优先级顺序：任何 402（立即），持续的 401 运行（重新生成的密钥未部署到所有地方），429 率高于呼叫的百分之几（你的限流器大小不正确）和 503 集群（退避更宽，考虑队列）。每分钟状态计数的仪表板使所有四个一目了然。处理得当，这些代码成为常规信号而不是故障。

在发布前测试你的故障路径

从未运行过的错误处理是一种猜测。你可以使用便宜的请求和你自己的测试模拟来执行大多数分支，而无需等待真正的故障。

• 401: 调用 GET /v1/models 并使用故意错误的密钥，确认你的应用向操作员显示配置错误，而不是向客户显示。
• 404: 请求 /v1/chat/completion（单数）一次，并确认日志行命名了路径，以便在几分钟内发现基础 URL 中的拼写错误。
• 400: 发送 max_tokens 为 32,000，提示词填充至大约 50,000 token。总和超过 100,000，你可以观察验证路径触发。
• 429 和 503： 你无法按需触发这些状态，因此需要模拟它们。将 HTTP 客户端包裹在一个假对象中，前两次调用抛出该状态，随后成功，并断言你的循环会休眠、重试并返回最终答案。
• 402: 也模拟它，并确认 UI 显示暂停界面，警报触发，且没有密集循环重试。

将这些编写为在持续集成中运行的自动化测试。使用表格驱动测试（表格中的每个状态对应一行）只需几分钟即可编写，并能防止善意的重构破坏策略。最后，使用 API 模拟对自身的节流进行简短的负载测试：发起 1,000 次获取，并断言在任何一个 60 秒窗口内，授予次数不超过 300 次加上突发量。这证明了在真实流量依赖该算法之前的算术正确性。当你稍后阅读 agent loop guide 时，请在那里重用相同的包装器，因为代理会成倍增加请求计数并首先触及这些限制。