为AI网关构建支持定价的请求队列

大多数 AI 成本控制都是在请求发送之后才添加的。一个请求进入网关，SDK 选择模型，提供商接受或拒绝调用，然后团队才计算发生了什么。对于服务于多个租户、任务类别和模型家族的生产级 AI 系统来说，这太晚了。

更好的成本控制位置是请求队列。队列在 tokens 消耗之前就能看到需求。它可以将交互式工作与批处理工作分离，附加带日期的费率卡，检查并发限制，保留缓存亲和性，并决定一个任务应该立即运行、等待更便宜的运行窗口，还是发送到不同的已批准路由。

本文展示如何将该队列构建为 OpenAI 兼容 AI 网关的一个小型生产级原语。目标不是把队列变成一个神奇的优化器。目标是使调度决策变得明确、可审计，并与来源可查的定价挂钩。

以下所有来源和定价检查均在 2026 年 8 月 23 日更新。AIWave 的公开定价页面列出 DeepSeek V4 Flash 为每 1M tokens $0.638 输入、$1.914 输出、$0.0203 缓存命中输入；DeepSeek V4 Pro 为每 1M tokens $1.914 输入、$5.742 输出、$0.0638 缓存命中输入。DeepSeek 官方页面列出了 V4 Flash 和 V4 Pro 的高峰和非高峰行，以及账户级别并发限制：Flash 为 2500，Pro 为 500。QwenCloud 文档记录了请求级分层 token 计费、Batch API 费率为实时定价的 50%、上下文缓存、思考 token 输出计费以及工具费用。Z.AI 列出了 GLM-5.3、GLM-5.2 和 GLM-5.1，每 1M tokens $1.40 输入、$0.26 缓存输入、$4.40 输出。Kimi K3 公开资料列出 API 路径为每 1M tokens $0.30 缓存命中输入、$3.00 缓存未命中输入、$15.00 输出。

重要的教训不是某个提供商的表格更好。教训是路由经济学因模型、缓存状态、输出上限、请求时间和提供商功能而异。一个忽略这些字段的队列会迅速做出昂贵的决策。

为什么队列应该知道定价

一个普通的任务队列知道优先级、重试次数、创建时间和 worker 容量。AI 请求队列还需要定价上下文。同一个提示可能会产生非常不同的风险敞口，这取决于它是新鲜输入还是缓存命中输入，响应上限是 256 还是 8,000 tokens，任务是交互式还是异步的，以及提供商是否单独收取工具调用费用。

如果队列只存储模型和优先级，它无法回答实际问题：

这些是队列决策，因为它们发生在调用之前。一旦请求正在飞行中，网关仍然可以重试或重新路由，但它已经承担了延迟、容量和成本敞口。

调度契约

首先，让请求队列接受一个调度契约，而不是原始提示。调度契约是一个小文档，说明任务允许做什么。

{
  "tenant_id": "acme-support",
  "task_class": "support_summary",
  "latency_class": "interactive",
  "model_preferences": ["deepseek-v4-flash", "glm-5.3"],
  "max_input_tokens": 12000,
  "max_output_tokens": 900,
  "cache_namespace": "acme-support-policy-v3",
  "batch_allowed": false,
  "reroute_allowed": true,
  "deadline_at": "2026-08-23T14:05:00Z",
  "budget_ceiling_usd": 0.01,
  "policy_version": "dispatch-queue-2026-08-23"
}

这个契约应该由你的产品或 SDK 层创建，而不是在请求已经迟到之后由 worker 发明。支持摘要、代码补丁、合规审查和离线基准测试不应该共享一个全局策略。

队列然后添加操作元数据：

最终这条记录成为请求为什么在那个地方运行的解释。

不要构建一个队列。构建车道。

队列应该只通过策略在车道之间移动任务。例如，一个离线评估可以在截止日期临近且预算允许的情况下从 batch_async 移动到 interactive_reasoning。一个面向客户的聊天请求不应该仅仅因为存在更便宜的路径就移动到延迟批处理车道。

一个最小的 Python 调度器

以下示例实现了一个紧凑的定价感知调度器。它使用 AIWave 的带日期 DeepSeek 行作为统一路由，并将官方提供商字段保持相同的形状，这样你就可以添加 DeepSeek 高峰窗口、Qwen 批处理路由、GLM 缓存行或 Kimi 长上下文路由，而无需更改队列 API。

import os
import time
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Literal

from openai import OpenAI

client = OpenAI(
    base_url="https://aiwave.live/v1",
    api_key=os.environ.get("AIWAVE_API_KEY", "YOUR_API_KEY_HERE"),
)

LatencyClass = Literal["interactive", "async"]

@dataclass(frozen=True)
class RateCard:
    route: str
    family: str
    input_per_m: float
    cached_input_per_m: float | None
    output_per_m: float
    concurrency_limit: int | None
    source_url: str
    source_date: str
    window: str

@dataclass(frozen=True)
class DispatchContract:
    tenant_id: str
    task_class: str
    latency_class: LatencyClass
    prompt: str
    max_output_tokens: int
    estimated_input_tokens: int
    estimated_cached_tokens: int
    budget_ceiling_usd: float
    batch_allowed: bool
    reroute_allowed: bool
    deadline_epoch: float

@dataclass(frozen=True)
class DispatchDecision:
    route: str
    lane: str
    estimated_usd: float
    dispatch_now: bool
    reason: str
    policy_version: str
    rate_card_source_date: str

RATE_CARDS = {
    "deepseek-v4-flash": RateCard(
        route="deepseek-v4-flash",
        family="deepseek",
        input_per_m=0.638,
        cached_input_per_m=0.0203,
        output_per_m=1.914,
        concurrency_limit=2500,
        source_url="https://aiwave.live/pricing",
        source_date="2026-08-23",
        window="aiwave_all_day",
    ),
    "deepseek-v4-pro": RateCard(
        route="deepseek-v4-pro",
        family="deepseek",
        input_per_m=1.914,
        cached_input_per_m=0.0638,
        output_per_m=5.742,
        concurrency_limit=500,
        source_url="https://aiwave.live/pricing",
        source_date="2026-08-23",
        window="aiwave_all_day",
    ),
}

def estimate_usd(contract: DispatchContract, row: RateCard) -> float:
    cached = min(contract.estimated_cached_tokens, contract.estimated_input_tokens)
    fresh = contract.estimated_input_tokens - cached
    cached_rate = row.cached_input_per_m if row.cached_input_per_m is not None else row.input_per_m
    total = (
        fresh / 1_000_000 * row.input_per_m
        + cached / 1_000_000 * cached_rate
        + contract.max_output_tokens / 1_000_000 * row.output_per_m
    )
    return round(total, 6)

def choose_lane(contract: DispatchContract) -> str:
    if contract.latency_class == "async" and contract.batch_allowed:
        return "batch_async"
    if contract.task_class in {"planning", "code_review", "reasoning"}:
        return "interactive_reasoning"
    return "interactive_fast"

def decide(contract: DispatchContract) -> DispatchDecision:
    lane = choose_lane(contract)
    candidates = ["deepseek-v4-flash", "deepseek-v4-pro"] if contract.reroute_allowed else ["deepseek-v4-flash"]
    scored = []
    for route in candidates:
        row = RATE_CARDS[route]
        cost = estimate_usd(contract, row)
        scored.append((cost, route, row))

    scored.sort(key=lambda item: item[0])
    cost, route, row = scored[0]

    now = time.time()
    if cost > contract.budget_ceiling_usd:
        return DispatchDecision(
            route=route,
            lane="manual_review",
            estimated_usd=cost,
            dispatch_now=False,
            reason="estimated_cost_exceeds_contract_ceiling",
            policy_version="dispatch-queue-2026-08-23",
            rate_card_source_date=row.source_date,
        )

    if lane == "batch_async" and contract.deadline_epoch - now > 900:
        return DispatchDecision(
            route=route,
            lane=lane,
            estimated_usd=cost,
            dispatch_now=False,
            reason="batch_job_can_wait_for_async_window",
            policy_version="dispatch-queue-2026-08-23",
            rate_card_source_date=row.source_date,
        )

    return DispatchDecision(
        route=route,
        lane=lane,
        estimated_usd=cost,
        dispatch_now=True,
        reason="within_budget_and_policy",
        policy_version="dispatch-queue-2026-08-23",
        rate_card_source_date=row.source_date,
    )

这段代码有意做得很小。在生产环境中，你会替换粗糙的 token 估算，将决策存储在持久化队列中，追踪飞行中的提供商容量，并在响应后对估算值与实际使用量进行对账。核心形态仍然成立：每个调度决策都返回一个路由、车道、估算成本、来源日期和原因。

缓存亲和性是一个调度约束

缓存节省只有在其提供商以兼容方式看到相同的可重用上下文时才能存在。如果你的队列随机地将重复的长上下文工作跨提供商、租户或缓存命名空间移动，费率卡可能会显示缓存命中输入可用，而运行时却很少能获得它。

像对待调度约束一样对待缓存亲和性：

队列应该存储 cache_namespace、prompt_fingerprint 和 cached_token_estimate。worker 稍后应该在提供商或聚合器返回时记录实际的缓存命中 tokens。这就闭环了规划和计费之间的回路。

并发不仅仅是吞吐量

DeepSeek 的公开速率限制文档列出了 V4 Flash 和 V4 Pro 的账户级别并发限制，并描述了超过限制时的 HTTP 429 行为。这类字段应该与定价存在于同一个调度器中。如果 Pro 车道的并发量少于 Flash 车道，队列应该避免在高峰期间用低价值工作填满它。

一个简单的容量策略可以保留分片：

这些百分比不是通用的。有用的是隔离。一个糟糕的评估批次不应该消耗支持流量需要的车道。重试风暴应该有一个小 envelope，而不是无限制地访问每个 worker。

批处理窗口需要截止日期

QwenCloud 的定价文档描述了符合条件的异步工作负载的 Batch API 费率为实时定价的 50%。这正是请求队列应该理解的功能类型。一个任务不能自己选择批处理模式；产品必须声明是否接受延迟。

使用基于截止日期的规则：

"batch allowed" 这个短语是不够的。一份在 UTC 23:00 到期的每日报告和一份需要立即回复的客户聊天在代码中都是异步的，但只有一个可以等待。

工具费用属于队列

工具调用改变调度经济学。QwenCloud 列出了某些工具的内置工具费用，并说明工具描述可以计入函数调用和 MCP 的输入 tokens。Z.AI 将网络搜索列为按次使用的工具成本。即使你的网关抽象了这些细节，队列也应该在调度前收到一个工具计划。

{
  "tools_requested": ["web_search", "code_interpreter"],
  "tool_call_cap": 3,
  "tool_budget_ceiling_usd": 0.02,
  "tool_descriptions_token_estimate": 1800
}

然后决定任务是否仍然符合其预算。一个带有小提示和大工具 schema 的请求可能比预期更昂贵，因为工具描述和工具输出成为模型上下文的一部分。只看到用户文本的队列会低估它。

每个队列决策都应该产生可以与最终使用台账联接的字段。

这些字段让财务和工程可以调试同一个事件。如果估算值始终高于实际使用量，改进 token 预测。如果估算值始终低于实际使用量，检查输出上限、工具 schema 和缓存假设。如果某个车道主导了重试支出，在它成为平台事件之前将其隔离。

首先使用影子模式。保持调度行为不变，但让队列为每个请求计算车道、路由、估算支出和调度原因。将这些估算值与实际使用量比较一周。

然后，仅执行最安全的规则：输出上限和租户预算上限。如果请求估算高于其契约，将其发送到审查或要求调用者降低最大输出。这可以防止最明显的预算意外，而无需更改模型质量。

然后添加车道隔离。为交互流量、异步作业、重试和测试提供独立的容量桶。这减少了事件爆炸半径，即使在你优化路由之前。

之后，为声明的异步作业添加批处理调度。需要一个真实的截止日期。当队列等待而不是立即调度时存储原因。

最后，添加模型重新路由。最后做这个，因为它可以改变质量、延迟、隐私姿态和成本。每个批准的路由对应该有回放 fixtures、接受阈值和带来源日期的费率卡。

实现来源链接

如果你在 AIWave 上构建，从定价页面、可预测定价计算器、模型目录和 Chat Completions 文档开始。这些页面提供了队列需要的费率卡和 OpenAI 兼容的请求上下文。

对于提供商检查，保留指向 DeepSeek 定价、DeepSeek 速率限制、QwenCloud 定价、Z.AI 定价和 Kimi K3 公开页面的带日期链接。在更改路由预算或队列策略之前重新检查它们。

在你发布定价感知队列之前，确保它能回答这些问题：

队列是在请求变成支出之前放慢脚步的正确位置。一旦定价、缓存状态、并发、截止日期和路由策略在调度前可见，网关就可以在压力下做出无聊的决策：在契约允许时立即运行，在任务可以等待时等待，在重要时保留缓存，并在请求不符合其预算时停止。