Build a Pricing-Aware Request Queue for AI Gateways

Most AI cost controls are added after dispatch. A request enters the gateway, the SDK picks a model, the provider accepts or rejects the call, and only then does the team calculate what happened. That is too late for production AI systems that serve many tenants, task classes, and model families.

A better place to control spend is the request queue. The queue sees demand before tokens are spent. It can separate interactive work from batch work, attach dated rate cards, check concurrency limits, preserve cache affinity, and decide whether a job should run now, wait for a cheaper operating window, or be sent to a different approved route.

This article shows how to build that queue as a small production primitive for OpenAI-compatible AI gateways. The goal is not to turn a queue into a magic optimizer. The goal is to make dispatch decisions explicit, auditable, and tied to source-checked pricing.

All source and pricing checks below were refreshed on August 23, 2026. AIWave's public pricing page lists DeepSeek V4 Flash at $0.638 input, $1.914 output, and $0.0203 cache-hit input per 1M tokens; DeepSeek V4 Pro at $1.914 input, $5.742 output, and $0.0638 cache-hit input per 1M tokens. DeepSeek's official page lists peak and off-peak rows for V4 Flash and V4 Pro, plus account-level concurrency limits of 2500 for Flash and 500 for Pro. QwenCloud documents request-tiered token billing, Batch API rates at 50% of real-time pricing, context caching, thinking-token output billing, and tool fees. Z.AI lists GLM-5.3, GLM-5.2, and GLM-5.1 at $1.40 input, $0.26 cached input, and $4.40 output per 1M tokens. Kimi K3 public material lists $0.30/MTok cache-hit input, $3.00/MTok cache-miss input, and $15.00/MTok output for its API path.

The important lesson is not that one provider has a better table. The lesson is that route economics differ by model, cache state, output cap, request timing, and provider feature. A queue that ignores those fields will make expensive decisions in a hurry.

Why the Queue Should Know Pricing

A normal job queue knows priority, retry count, creation time, and worker capacity. An AI request queue also needs pricing context. The same prompt can produce very different exposure depending on whether it is fresh input or cache-hit input, whether the response cap is 256 or 8,000 tokens, whether the job is interactive or asynchronous, and whether a provider charges tool calls separately.

If the queue only stores model and priority, it cannot answer practical questions:

Question	Queue field needed
Can this request wait for a scheduled batch lane?	deadline_at, task_class, batch_allowed
Should it preserve a provider cache key?	cache_namespace, cache_affinity_required
Is the output budget larger than the tenant allows?	max_output_tokens, tenant_budget_usd
Can it use a Pro route during a capacity incident?	approved_route_set, quality_floor
Could a retry exceed the daily retry envelope?	attempt, estimated_retry_usd
Should it avoid a provider peak window?	rate_card_window, dispatch_after

These are queue decisions because they happen before the call. Once the request is in flight, the gateway can still retry or reroute, but it has already accepted latency, capacity, and cost exposure.

The Dispatch Contract

Start by making the request queue accept a dispatch contract instead of a raw prompt. A dispatch contract is a small document that says what the job is allowed to do.

{
  "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"
}

This contract should be created by your product or SDK layer, not invented by a worker after the request is already late. A support summary, coding patch, compliance review, and offline benchmark should not share one global policy.

The queue then adds operational metadata:

Field	Added by
rate_card_source_date	Pricing loader
estimated_fresh_input_usd	Queue estimator
estimated_cached_input_usd	Queue estimator
estimated_output_usd	Queue estimator
selected_route	Scheduler
dispatch_reason	Scheduler
queue_wait_ms	Queue runtime
capacity_bucket	Provider adapter

That final record becomes the explanation for why the request ran where it ran.

Route Classes

Do not build one queue. Build lanes.

Lane	Purpose	Typical policy
interactive_fast	Chat, support drafts, small extraction	Short wait, strict output cap, no batch delay
interactive_reasoning	Planning, coding review, complex analysis	Higher quality floor, lower concurrency share
batch_async	Evaluations, document conversion, nightly jobs	Deadline-based scheduling, batch route allowed
cache_warm	Repeated system prompts or reference packs	Preserve namespace and provider affinity
retry_repair	Bounded replay after transient failure	Small daily envelope, no blind escalation
manual_review	Risky or high-spend jobs	Human or policy approval before dispatch

The queue should move jobs between lanes only through policy. For example, an offline evaluation can move from batch_async to interactive_reasoning if the deadline is near and the budget allows it. A customer-facing chat request should not move into a delayed batch lane just because a cheaper path exists.

A Minimal Python Scheduler

The following example implements a compact pricing-aware scheduler. It uses AIWave's dated DeepSeek rows for a unified route and keeps official provider fields in the same shape so you can add DeepSeek peak windows, Qwen batch routes, GLM cache rows, or Kimi long-context routes without changing the queue 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_latency_policy",
        policy_version="dispatch-queue-2026-08-23",
        rate_card_source_date=row.source_date,
    )

def dispatch(contract: DispatchContract):
    decision = decide(contract)
    print(asdict(decision))
    if not decision.dispatch_now:
        return {"status": "queued", "decision": asdict(decision)}

    response = client.chat.completions.create(
        model=decision.route,
        messages=[{"role": "user", "content": contract.prompt}],
        max_tokens=contract.max_output_tokens,
    )
    return response

deadline = datetime(2026, 8, 23, 14, 5, tzinfo=timezone.utc).timestamp()
dispatch(
    DispatchContract(
        tenant_id="acme-support",
        task_class="support_summary",
        latency_class="interactive",
        prompt="Summarize the incident notes for a customer-facing reply.",
        max_output_tokens=900,
        estimated_input_tokens=12000,
        estimated_cached_tokens=8000,
        budget_ceiling_usd=0.01,
        batch_allowed=False,
        reroute_allowed=True,
        deadline_epoch=deadline,
    )
)

This code is intentionally small. In production you would replace the rough token estimates, store decisions in a durable queue, track in-flight provider capacity, and reconcile estimates with actual usage after the response. The core shape still holds: every dispatch decision returns a route, lane, estimated cost, source date, and reason.

Cache Affinity Is a Scheduling Constraint

Cache savings only exist if the provider sees the same reusable context in a compatible way. If your queue randomly moves repeated long-context work across providers, tenants, or cache namespaces, the rate card may say cache-hit input is available while the runtime rarely earns it.

Treat cache affinity like a scheduling constraint:

Workload	Queue behavior
Repeated coding-agent repository context	Keep a stable route unless quality policy says otherwise
Support bot with a large policy document	Use a named cache namespace and measure hit share
One-off customer prompt	Do not over-optimize for cache
Evaluation batch with repeated system prompt	Group jobs by prompt fingerprint
Retry after provider failure	Preserve cache namespace only if reroute policy allows

The queue should store cache_namespace, prompt_fingerprint, and cached_token_estimate. The worker should later record actual cache-hit tokens when the provider or aggregator returns them. That closes the loop between planning and billing.

Concurrency Is Not Just Throughput

DeepSeek's public rate-limit docs list account-level concurrency limits for V4 Flash and V4 Pro, and describe HTTP 429 behavior when limits are exceeded. That kind of field should live in the same scheduler as pricing. If the Pro lane has less concurrency than the Flash lane, the queue should avoid filling it with low-value work during a spike.

A simple capacity policy can reserve slices:

Capacity slice	Example use
50%	Interactive production traffic
20%	High-confidence reasoning jobs
15%	Async jobs near deadline
10%	Retry repair lane
5%	Internal tests and smoke checks

These percentages are not universal. The useful part is the isolation. A bad evaluation batch should not consume the same lane that support traffic needs. Retry storms should have a small envelope, not unlimited access to every worker.

Batch Windows Need Deadlines

QwenCloud's pricing docs describe Batch API rates at 50% of real-time pricing for eligible async workloads. That is exactly the kind of feature a request queue should understand. A job cannot choose batch mode by itself; the product has to declare whether delay is acceptable.

Use a deadline-based rule:

Deadline	Dispatch choice
Under 60 seconds	Interactive lane only
1 to 15 minutes	Interactive or short queue wait
15 minutes to 24 hours	Batch lane allowed
No deadline	Require a product owner to set one

The phrase "batch allowed" is not enough. A daily report due at 23:00 UTC and a customer chat response due now are both asynchronous in code, but only one can wait.

Tool Fees Belong in the Queue

Tool calls change dispatch economics. QwenCloud lists built-in tool fees for some tools, and states that tool descriptions can count as input tokens for function calling and MCP. Z.AI lists web search as a per-use tool cost. Even if your gateway abstracts those details, the queue should receive a tool plan before dispatch.

Add fields such as:

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

Then decide whether the job still fits its budget. A request with a small prompt and a large tool schema can be more expensive than expected because tool descriptions and tool outputs become part of the model context. A queue that sees only user text will underestimate it.

Observability Fields

Every queue decision should produce fields that can be joined with the final usage ledger.

Field	Example
ai.queue.lane	interactive_fast
ai.queue.reason	within_budget_and_latency_policy
ai.route.model	deepseek-v4-flash
ai.route.family	deepseek
ai.estimate.usd	0.004842
ai.actual.usd	0.004501
ai.rate_card.source_date	2026-08-23
ai.cache.estimated_tokens	8000
ai.cache.actual_tokens	7912
ai.deadline.ms_remaining_at_dispatch	48000
ai.policy.version	dispatch-queue-2026-08-23

These fields let finance and engineering debug the same event. If estimates are consistently above actual usage, improve token prediction. If estimates are consistently below actual usage, inspect output caps, tool schemas, and cache assumptions. If one lane dominates retry spend, isolate it before it becomes a platform incident.

Rollout Plan

Start with shadow mode. Keep dispatch behavior unchanged, but make the queue calculate lane, route, estimated spend, and dispatch reason for every request. Compare those estimates with actual usage for one week.

Next, enforce only the safest rule: output caps and tenant budget ceilings. If a request estimates above its contract, send it to review or ask the caller to lower max output. This prevents the most obvious budget surprises without changing model quality.

Then add lane isolation. Give interactive traffic, async jobs, retries, and tests separate capacity buckets. This reduces incident blast radius even before you optimize routing.

After that, add batch scheduling for declared async jobs. Require a real deadline. Store the reason when the queue waits instead of dispatching immediately.

Finally, add model rerouting. Do this last because it can change quality, latency, privacy posture, and cost. Every approved route pair should have replay fixtures, acceptance thresholds, and source-dated rate cards.

Source Links for Implementation

If you are building this on AIWave, start with the pricing page, predictable-pricing calculator, model catalog, and Chat Completions documentation. Those pages provide the rate-card and OpenAI-compatible request context a queue needs.

For provider checks, keep dated links to DeepSeek pricing, DeepSeek rate limits, QwenCloud pricing, Z.AI pricing, and the Kimi K3 public page. Recheck them before changing route budgets or queue policy.

Final Checklist

Before you ship a pricing-aware queue, make sure it can answer these questions:

Requirement	Pass condition
Dated rate cards	Every route stores source URL and source date
Lane isolation	Interactive, async, retry, and test traffic have separate buckets
Budget estimates	Fresh input, cached input, output, and tool exposure are estimated separately
Cache affinity	Repeated contexts keep namespace and route metadata
Deadline handling	Batch-eligible jobs have explicit deadlines
Capacity awareness	Provider concurrency limits are part of scheduling
Decision logging	Every dispatch has a policy version and human-readable reason
Reconciliation	Estimated and actual usage are joined after completion

The queue is the right place to slow down before a request becomes spend. Once pricing, cache state, concurrency, deadlines, and route policy are visible before dispatch, the gateway can make boring decisions under pressure: run now when the contract allows it, wait when the job can wait, preserve cache when it matters, and stop when the request does not fit its budget.