OpenNave 文档中心

外观

频率限制 ​

OpenNave 通过频率限制保障公平使用和平台稳定性。本文说明 RPM、TPM 的限制维度、触发机制及应用侧的限流处理方式。

限制维度 ​

维度说明
RPM每分钟请求数，即每分钟允许的 API 请求数量。
TPM每分钟 Token 数，即每分钟允许的输入与输出 Token 总量。

某一维度的数值为 0 时，表示该维度不设置限制。

限制层级 ​

请求必须同时通过以下两个独立层级的检查。

用户级别限制 ​

账户所属用户组决定用户级 RPM 和 TPM 限制。平台管理员可以对不同用户组设置不同额度。

通道级别限制 ​

每个上游通道也可以设置独立的 RPM 和 TPM，以保护提供商端点免受突发并发冲击。某个通道达到限制时，网关可能将请求路由至同一模型的其他可用通道。

限制执行机制 ​

OpenNave 使用固定窗口算法。

RPM 执行流程 ​

  1. 网关在请求处理前增加 RPM 计数器。
  2. 计数器超过限制时，请求被拒绝并返回 429 Too Many Requests。
  3. 计数器在下一个分钟窗口开始时重置。

TPM 执行流程 ​

  1. 网关在调用前预估输入 Token，优先使用 Tokenizer，无法计算时按“文本长度除以 4”估算。
  2. 将预估值与 TPM 限制比较。
  3. 响应完成后，使用提供商返回的实际 Token 用量修正计数器。

预检阶段的 TPM 值是近似值，但最终计数会以实际用量校正。

频率限制响应 ​

触发限制时，接口返回 429 Too Many Requests，错误格式与所用协议保持一致。

OpenAI 格式：

json
{
  "error": {
    "message": "rate limit exceeded: user RPM limit",
    "type": "rate_limit_error"
  }
}

Anthropic 格式：

json
{
  "type": "error",
  "error": {
    "type": "rate_limit_error",
    "message": "rate limit exceeded: user RPM limit"
  }
}

Gemini 格式：

json
{
  "error": {
    "code": 429,
    "message": "rate limit exceeded: user RPM limit",
    "status": "RESOURCE_EXHAUSTED"
  }
}

处理频率限制 ​

指数退避 ​

收到 429 后使用指数退避与随机抖动，避免集中重试：

python
import random
import time
from openai import APIStatusError, OpenAI

client = OpenAI(
    base_url="https://api.opennave.top/v1",
    api_key="YOUR_OPENNAVE_API_KEY",
)

def call_with_backoff(max_retries=5):
    for attempt in range(max_retries):
        try:
            return client.responses.create(model="gpt-5.4", input="你好！")
        except APIStatusError as error:
            if error.status_code != 429 or attempt == max_retries - 1:
                raise
            delay = (2 ** attempt) + random.random()
            time.sleep(delay)

均匀分布请求 ​

不要在窗口开始时集中发出所有请求。RPM 为 60 时，目标速率约为每秒 1 个请求。对于批处理，使用受控速率分发请求：

python
import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.opennave.top/v1",
    api_key="YOUR_OPENNAVE_API_KEY",
)

async def process_batch(prompts: list[str], rps: float = 1.0):
    results = []
    interval = 1.0 / rps
    for prompt in prompts:
        results.append(
            await client.responses.create(model="gpt-5.4", input=prompt)
        )
        await asyncio.sleep(interval)
    return results

使用请求队列 ​

高吞吐场景使用带有并发上限的队列：

python
import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.opennave.top/v1",
    api_key="YOUR_OPENNAVE_API_KEY",
)

async def process_with_concurrency(prompts: list[str], max_concurrent: int = 5):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_request(prompt: str):
        async with semaphore:
            return await client.responses.create(model="gpt-5.4", input=prompt)

    return await asyncio.gather(*(limited_request(prompt) for prompt in prompts))

流式响应可降低用户感知的首字延迟，但不会绕过 RPM 或 TPM 限制。

最佳实践 ​

  • 从应用设计阶段就加入队列、并发控制和退避策略。
  • 为不同服务或环境使用独立 API Key，以隔离用量和配额。
  • 缓存可复用响应，减少不必要请求。
  • 对 429 错误率建立监控和告警。

关联文档 ​