Tutorial guide

Build a Cost-Aware Python Router for Chinese AI APIs

A hands-on guide to routing OpenAI-compatible calls across DeepSeek, Kimi, GLM and Qwen with budgets, retries, JSON validation and live price metadata.

Published 2026-08-0113 min readPython AI router

This tutorial is designed for a developer team that wants one API surface while still controlling cost. It uses OpenAI-compatible calls, environment variables and local validation. Replace model IDs with the exact names enabled in your AIWave account.

What the router must do

A useful multi-model router is not a random load balancer. It is a policy engine for model selection. The policy should know which jobs are cheap, which jobs are risky, which jobs require strict JSON, and which jobs can be retried without side effects. That is especially important when working with Chinese AI models because the providers expose different pricing shapes: DeepSeek publishes separate cache-hit and cache-miss input prices, Z.AI publishes cached-input prices for GLM models, Alibaba Model Studio has regional and tiered pricing for Qwen, and Kimi exposes model-list and pricing docs for Moonshot families.

The goal is to keep the application code boring. Your product should call a single function such as complete(task, messages). The router decides whether the request goes to DeepSeek V4 Flash, GLM-5.2, Qwen or a Kimi-family model. It records the decision, validates the response and falls back when the result fails the contract. That lets you change policy without rewriting the app.

The router in this guide has four features. It loads model metadata from a Python dictionary that you can later move to a database. It estimates cost before the call using dated rates. It validates JSON for extraction tasks. It retries with a fallback model only when the task is safe. For deeper context on provider-specific setup, keep the AIWave Python SDK guide, structured output guide and multi-model router guide open while implementing.

Price metadata

Price metadata needs more fields than most teams expect. A rate without a checked date is stale. A rate without a source URL cannot be audited. A Qwen rate without a region may be wrong. A long-context request without cache information may be over-budget even when the input price looks low. Use this shape as a starting point:

FieldPurposeExample
modelExact model ID requested at runtimedeepseek-v4-flash
providerBilling or upstream familyDeepSeek, Z.AI, Alibaba, Moonshot
input_per_mUncached input price per 1M tokens0.14 for DeepSeek V4 Flash official docs on Aug 1
cached_input_per_mCached input price per 1M tokens when applicable0.0028 for DeepSeek V4 Flash official docs on Aug 1
output_per_mOutput price per 1M tokens0.28 for DeepSeek V4 Flash official docs on Aug 1
regionRequired when provider pricing varies by geographySingapore, US, Germany, global
sourcePricing page or API sourceProvider documentation URL
checked_atDate you verified the number2026-08-01

When a provider only has discussion coverage for a new model, keep that in a separate note field and avoid using it for customer billing. For example, recent Kimi K3 articles are useful for understanding market attention and likely workload fit, but the router should verify the official platform before assigning hard production rates. This distinction prevents marketing pages and social posts from becoming accidental billing logic.

Runnable Python implementation

The following script runs as a single file. It uses the OpenAI Python SDK, accepts a task type, estimates cost, calls the selected model and validates JSON when required. To test it without spending much, start with a short prompt and a low max token cap.

# pip install openai
import json
import os
import time
from dataclasses import dataclass
from typing import Any
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["AIWAVE_API_KEY"],
    base_url=os.getenv("AIWAVE_BASE_URL", "https://api.aiwave.live/v1"),
)

RATES = {
    "deepseek-v4-flash": {
        "provider": "DeepSeek",
        "input_per_m": 0.14,
        "cached_input_per_m": 0.0028,
        "output_per_m": 0.28,
        "checked_at": "2026-08-01",
        "source": "https://api-docs.deepseek.com/quick_start/pricing/",
    },
    "glm-5.2": {
        "provider": "Z.AI",
        "input_per_m": 1.40,
        "cached_input_per_m": 0.26,
        "output_per_m": 4.40,
        "checked_at": "2026-08-01",
        "source": "https://docs.z.ai/guides/overview/pricing",
    },
    "qwen3.7-plus": {
        "provider": "Alibaba Model Studio",
        "input_per_m": None,  # fill from your selected region and tier
        "cached_input_per_m": None,
        "output_per_m": None,
        "checked_at": "2026-08-01",
        "source": "https://help.aliyun.com/zh/model-studio/model-pricing",
    },
}

@dataclass
class Job:
    kind: str
    prompt: str
    require_json: bool = False
    stable_prefix: bool = False
    max_output: int = 1000

def estimate_tokens(text: str) -> int:
    return max(1, len(text) // 4)

def choose_model(job: Job) -> str:
    tokens = estimate_tokens(job.prompt)
    if job.kind == "repo_agent" or tokens > 160_000:
        return "glm-5.2"
    if job.stable_prefix or job.require_json:
        return "deepseek-v4-flash"
    return "deepseek-v4-flash"

def estimate_usd(model: str, prompt_tokens: int, max_output: int, cached: bool) -> float | None:
    rate = RATES[model]
    input_rate = rate["cached_input_per_m"] if cached else rate["input_per_m"]
    output_rate = rate["output_per_m"]
    if input_rate is None or output_rate is None:
        return None
    return (prompt_tokens / 1_000_000 * input_rate) + (max_output / 1_000_000 * output_rate)

def call_model(model: str, job: Job) -> str:
    system = "You are a precise API assistant. Return valid JSON only." if job.require_json else "Be concise and verify claims."
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "system", "content": system}, {"role": "user", "content": job.prompt}],
        max_tokens=job.max_output,
        temperature=0.1,
    )
    return response.choices[0].message.content or ""

def complete(job: Job) -> dict[str, Any]:
    primary = choose_model(job)
    fallback = "glm-5.2" if primary != "glm-5.2" else "deepseek-v4-flash"
    prompt_tokens = estimate_tokens(job.prompt)
    estimate = estimate_usd(primary, prompt_tokens, job.max_output, job.stable_prefix)
    errors = []
    for attempt, model in enumerate([primary, fallback], start=1):
        try:
            text = call_model(model, job)
            parsed = json.loads(text) if job.require_json else None
            return {"model": model, "attempt": attempt, "estimate_usd": estimate, "text": text, "json": parsed}
        except Exception as exc:
            errors.append({"model": model, "error": str(exc)})
            time.sleep(0.5)
    raise RuntimeError(f"all routes failed: {errors}")

if __name__ == "__main__":
    job = Job(
        kind="extract",
        prompt="Return JSON with vendor, amount_usd and due_date from: Invoice from Example Cloud for USD 42 due 2026-08-15.",
        require_json=True,
        stable_prefix=True,
        max_output=300,
    )
    print(json.dumps(complete(job), indent=2))

Validation and retries

The script validates JSON by parsing it. In production, add a schema layer with Pydantic or JSON Schema. For an invoice extractor, require fields such as vendor, currency, amount and due date. For an agent, validate file changes and tests. For chat, validate policy boundaries and maximum answer length. The router should never assume that a successful HTTP 200 means the task succeeded.

Retries deserve strict rules. Retry network timeouts or 5xx responses on the same model first because the original request may have been fine. Retry validation failures only when the task is idempotent. Do not retry actions that charge a customer, send an email, place an order or mutate a database unless you have idempotency keys. When you fall back to a stronger model, log the reason. Fallbacks are expensive, but they are also a quality signal.

Cost control has three layers. Set per-request max output so a model cannot generate an unbounded answer. Set per-task budget so an agent cannot loop forever. Set per-user or per-project budgets so a broken workflow does not drain the account. If you expose this to developers, show the selected model and approximate budget before long jobs begin. A transparent budget screen converts better than a surprise balance drop.

Deployment checklist

  • Keep API keys in server-side environment variables and never ship them to browsers.
  • Record source URL and checked date for every model price used in estimates.
  • Use exact model IDs and test that each ID is enabled for the account.
  • Run a small fixture suite before changing routing policy.
  • Track input, cached input, output, retries, fallback model and validation result.
  • Expose a dry-run mode that estimates cost without calling the model.
  • Build alerts for sudden fallback spikes or token-per-request growth.

Once the router is stable, move the rate table to your database and add an admin-only screen for updating rates. Keep the router code deterministic. A team should be able to explain why a request used DeepSeek instead of GLM, Qwen or Kimi by reading one log line. That is the difference between a multi-model platform and a pile of provider keys.

Sources