DEV Community

Mattias chaw
Mattias chaw

Posted on

Build a Migration Harness for OpenAI-Compatible Chinese AI APIs

Build a Migration Harness for OpenAI-Compatible Chinese AI APIs

Changing one SDK setting can make a prototype feel migrated:

client = OpenAI(
    api_key=os.environ["AIWAVE_API_KEY"],
    base_url="https://aiwave.live/v1",
)
Enter fullscreen mode Exit fullscreen mode

That line is useful, but it is not a migration plan.

For teams in the United States, the United Kingdom, Germany, the Netherlands, Japan, Singapore, Canada, Australia, and other Tier 1/2 markets, the real question is operational: can an existing OpenAI SDK workflow move to a Chinese AI model gateway without breaking response parsing, cost controls, data policy, observability, or rollback?

The answer should come from a migration harness. A harness is a small set of probes, fixtures, policy checks, and release rules that proves a route is ready before customer traffic moves. It is not a benchmark suite. It is not a model leaderboard. It is a release artifact for engineering teams that need to test DeepSeek, Qwen, GLM, Kimi, ERNIE, or other Chinese model families through an OpenAI-compatible gateway such as AIWave.

This article shows a practical harness design. The examples use environment variables and placeholder values only.

Why a harness matters

OpenAI compatibility reduces integration work, but it does not remove all migration risk. Different model families can vary in:

  • accepted request parameters;
  • streaming chunk shape;
  • JSON mode behavior;
  • tool call behavior;
  • usage metadata;
  • context-window boundaries;
  • cache accounting;
  • output-token growth;
  • account-level access;
  • regional and privacy review requirements.

A harness catches those differences before the router becomes the default path. It also gives product, security, finance, and support teams a shared artifact. Instead of asking "does the model work?", the team can ask "did this workflow pass the same probes, price checks, policy gates, and rollback rules as the route it is replacing?"

Pricing facts checked on August 13, 2026

Use dated pricing checks in the harness. Do not hard-code a provider table once and treat it as permanent infrastructure. Provider pages, model aliases, cache rules, and regional billing terms can change.

The public sources below were checked on August 13, 2026 for example harness inputs:

Provider family Example public rate shape Harness implication
DeepSeek V4 Flash $0.0028 per 1M cache-hit input tokens, $0.14 per 1M cache-miss input tokens, $0.28 per 1M output tokens; 1M context and 384K max output are listed on the official pricing page. Cache-hit visibility and output caps must be tested together.
DeepSeek V4 Pro $0.003625 per 1M cache-hit input tokens, $0.435 per 1M cache-miss input tokens, $0.87 per 1M output tokens. Planning and reasoning routes need separate ceilings from execution routes.
Qwen3.7 Flash QwenCloud lists context-tiered pricing for qwen3.7 text models; public model pages show 1M context and tool/caching capabilities. The harness must record input band and tool usage, not only total tokens.
GLM-5.1 / GLM-5 family Z.AI lists GLM-5.1 at $1.40 input, $0.26 cached input, and $4.40 output per 1M tokens; GLM-5 is listed at $1.00, $0.20, and $3.20. Cached input and output growth need separate budget gates.
Kimi K3 Kimi lists $0.30 cache-hit input, $3.00 cache-miss input, and $15.00 output per 1M tokens, with a 1,048,576-token context window. Long-context routes need prefix reuse tests and strict output limits.

These numbers are source facts for a harness example, not a promise about any account's final bill. A production release should check the pricing source that the application actually bills through, then store the checked date with the route config.

Sources:

Harness structure

Keep the harness boring and versioned. A simple directory is enough:

ai-route-harness/
  fixtures/
    support_reply.json
    invoice_summary.json
    code_review.json
  routes/
    support_reply.aiwave.glm.yaml
    code_review.aiwave.deepseek.yaml
  probes/
    chat.py
    json_mode.py
    streaming.py
    tools.py
  policy/
    customers.yaml
  reports/
    latest.json
Enter fullscreen mode Exit fullscreen mode

Fixtures are stable requests. Routes describe candidate models and prices. Probes execute the behavior checks. Policy files decide whether a route is allowed for a customer workflow. Reports become release evidence.

Route config

The route config should include model identity, gateway details, budget ceilings, pricing source, and rollback information.

route_id: support_reply_aiwave_glm5
checked_on: 2026-08-13
gateway:
  name: aiwave
  base_url: https://aiwave.live/v1
  endpoint: chat_completions
model:
  family: glm
  name: glm-5
workflow:
  owner: platform-ai
  task_class: support_reply
  customer_visible: true
  requires_json: true
  requires_streaming: false
budget:
  max_input_tokens: 18000
  max_output_tokens: 900
  max_request_usd: "0.020"
pricing:
  source_url: https://docs.z.ai/guides/overview/pricing
  input_usd_per_1m: "1.00"
  cached_input_usd_per_1m: "0.20"
  output_usd_per_1m: "3.20"
policy:
  retention_class: metadata_only
  personal_data_possible: true
rollback:
  feature_flag: ai_route_support_reply_glm5
  previous_route_id: support_reply_current_provider
Enter fullscreen mode Exit fullscreen mode

Money values are strings so they can be parsed with Decimal. The file stores pricing source and checked date together. If the pricing check is stale, the route should fail the release gate.

Probe the actual client path

The first probe should use the same SDK path as production. This catches base URL, auth, parameter, and response-shape issues.

import os
from openai import OpenAI


def make_client() -> OpenAI:
    return OpenAI(
        api_key=os.environ["AIWAVE_API_KEY"],
        base_url=os.environ.get("AIWAVE_BASE_URL", "https://aiwave.live/v1"),
    )


def probe_chat(model: str) -> dict:
    client = make_client()
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "Be concise."},
            {"role": "user", "content": "Return one deployment risk for an API migration."},
        ],
        temperature=0.2,
        max_tokens=80,
    )

    choice = response.choices[0]
    return {
        "model": model,
        "finish_reason": choice.finish_reason,
        "has_text": bool((choice.message.content or "").strip()),
        "has_usage": response.usage is not None,
    }
Enter fullscreen mode Exit fullscreen mode

This probe is intentionally small. It does not try to rank models. It verifies that the route can answer through the production client shape and that usage metadata exists.

Probe JSON mode

Many SaaS workflows do not consume prose. They consume structured output. If your existing route relies on JSON mode, the migration harness must test JSON mode directly.

import json


def probe_json_mode(model: str) -> dict:
    client = make_client()
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "Return valid JSON only."},
            {
                "role": "user",
                "content": "Return {\"status\":\"ok\",\"risk\":\"medium\"} with the same keys.",
            },
        ],
        response_format={"type": "json_object"},
        temperature=0,
        max_tokens=80,
    )
    text = response.choices[0].message.content or ""
    parsed = json.loads(text)
    return {
        "valid_json": True,
        "keys": sorted(parsed.keys()),
        "has_usage": response.usage is not None,
    }
Enter fullscreen mode Exit fullscreen mode

If a route cannot reliably satisfy the response contract, do not bury that result behind a fallback. Mark it as a failed probe and keep the current route in place.

Estimate request ceilings before release

The harness should reject routes that can exceed workflow budgets under allowed token limits.

from decimal import Decimal


def estimate_request_usd(
    input_tokens: int,
    output_tokens: int,
    input_usd_per_1m: str,
    output_usd_per_1m: str,
) -> Decimal:
    input_cost = Decimal(input_tokens) * Decimal(input_usd_per_1m) / Decimal(1_000_000)
    output_cost = Decimal(output_tokens) * Decimal(output_usd_per_1m) / Decimal(1_000_000)
    return input_cost + output_cost


def budget_gate(route: dict) -> tuple[bool, str]:
    budget = route["budget"]
    pricing = route["pricing"]
    ceiling = estimate_request_usd(
        budget["max_input_tokens"],
        budget["max_output_tokens"],
        pricing["input_usd_per_1m"],
        pricing["output_usd_per_1m"],
    )
    if ceiling > Decimal(budget["max_request_usd"]):
        return False, f"estimated ceiling {ceiling} exceeds policy"
    return True, str(ceiling)
Enter fullscreen mode Exit fullscreen mode

Cache rates can be added as a separate scenario, but do not let a cache-hit estimate be the only release gate. The first request, changed system prompts, and failed cache reuse can all behave like cache misses.

Add customer policy

The harness should evaluate customer policy before model selection. This is especially important for Tier 1/2 buyers that ask about privacy, data retention, billing records, and regional review.

def policy_gate(route: dict, customer: dict) -> tuple[bool, str]:
    family = route["model"]["family"]
    if family not in customer["allowed_model_families"]:
        return False, "model family blocked"

    if route["policy"]["personal_data_possible"]:
        if not customer["allows_personal_data_routes"]:
            return False, "personal data route blocked"

    if route["policy"]["retention_class"] != customer["required_retention_class"]:
        return False, "retention class mismatch"

    return True, "approved"
Enter fullscreen mode Exit fullscreen mode

This gate is deliberately plain. A production version may read from your tenant database or policy engine, but the output should still be auditable: route, customer class, decision, reason, and timestamp.

Log metadata, not prompts

A migration harness should also test the ledger shape it expects in production. Store metadata that helps debug routing and billing without storing raw prompts.

from datetime import datetime, timezone
import hashlib


def tenant_hash(tenant_id: str) -> str:
    return hashlib.sha256(tenant_id.encode("utf-8")).hexdigest()[:16]


def ledger_entry(
    tenant_id: str,
    route_id: str,
    model: str,
    probe_name: str,
    result: str,
    estimated_usd: str,
    pricing_source: str,
    checked_on: str,
) -> dict:
    return {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "tenant_hash": tenant_hash(tenant_id),
        "route_id": route_id,
        "model": model,
        "probe_name": probe_name,
        "result": result,
        "estimated_usd": estimated_usd,
        "pricing_source": pricing_source,
        "pricing_checked_on": checked_on,
        "retention_class": "metadata_only",
    }
Enter fullscreen mode Exit fullscreen mode

In a real application, add request ID, retry count, fallback route, prompt-token estimate, returned usage, cached-input tokens when available, output tokens, latency bucket, and validation status. Avoid raw prompts, customer emails, payment identifiers, real keys, and private document text.

Release rules

A route can move from test to production only when the harness report passes hard gates:

Gate Pass condition
SDK path The route works through the same OpenAI-compatible client shape as production.
Behavior Required probes pass for chat, JSON, streaming, or tools.
Pricing Pricing source and checked date are stored in the route config.
Budget Cache-miss request ceiling fits the workflow budget.
Policy Customer region, personal-data, and retention rules approve the route.
Ledger Accepted and rejected decisions produce metadata rows.
Rollback A feature flag or config switch can restore the previous route without deployment.

Do not waive rollback. A model route can fail because of capacity, provider changes, changed aliases, unexpected output growth, or account-level limits. A migration harness should assume rollback is part of normal operations.

Applying this to AIWave

AIWave's value is that a team can use one OpenAI-compatible API surface for 25+ Chinese AI models instead of maintaining separate integrations for every provider. That helps with experimentation, but production still needs release discipline.

A practical sequence looks like this:

  1. Keep the existing OpenAI SDK.
  2. Move a single non-critical workflow to https://aiwave.live/v1 behind a feature flag.
  3. Run chat, JSON, streaming, and tool probes for candidate models.
  4. Store dated pricing evidence for the route.
  5. Apply customer policy before model selection.
  6. Log metadata for accepted, rejected, retried, and fallback requests.
  7. Compare output quality and operational behavior on real fixtures.
  8. Move traffic gradually, then keep the previous route available until the new route has enough production evidence.

This approach works for support drafting, code review assistants, long-context document summarization, internal knowledge search, and agent planning/execution splits. It also keeps the migration conversation grounded in facts: request behavior, budget ceilings, policy decisions, and rollback paths.

Final rule

Do not treat an OpenAI-compatible base URL change as a completed migration.

Treat it as the first line in a migration harness. The route is ready only when the same client path, required response behavior, dated pricing check, customer policy, metadata ledger, and rollback control all pass in the same release window.

That is the difference between testing a model and operating a model route.

Top comments (0)