How to run shadow traffic during an LLM provider migration

Migrating to a new LLM provider without disrupting production is risky. Shadow traffic testing for LLM migration lets you send copies of real requests to the candidate model while still serving responses from the incumbent, so you can measure quality and latency before committing. This guide walks through a concrete setup you can run this week, from request interception to canary cutover.

Step 1: Establish baseline metrics and success criteria

You cannot judge a migration without a reference point. Pull the last 30 days of production LLM calls from your logging store and compute p50 and p95 latency per endpoint, token consumption, and error rate. If you don’t already record this, add a thin wrapper around your client that emits timing and token counts to your metrics pipeline.

Define hard pass/fail thresholds for the shadow candidate. These should be explicit numbers written down before you look at any shadow data. A practical config:

{
  "baseline_model": "openai/gpt-4o-mini",
  "shadow_model": "anthropic/claude-3.5-sonnet",
  "max_latency_ms_p95": 1200,
  "min_similarity": 0.85,
  "max_error_rate": 0.005,
  "sample_rate": 1.0
}

The sample_rate lets you mirror a subset if volume is high. Start at 1.0 for a small product; drop to 0.1 if you process millions of calls daily.

Step 2: Intercept production requests

The core of shadow traffic testing for LLM migration is duplication. In a Python service using the OpenAI SDK, wrap the client so every chat.completions.create also fires a background shadow call. The user must never wait on the shadow.

import asyncio
import openai
from typing import Any, Dict

primary = openai.AsyncOpenAI(api_key=PRIMARY_KEY)
shadow = openai.AsyncOpenAI(api_key=SHADOW_KEY, base_url=SHADOW_BASE)

async def routed_create(correlation_id: str, **payload):
    primary_resp = await primary.chat.completions.create(**payload)
    asyncio.create_task(_shadow(correlation_id, payload))
    return primary_resp

async def _shadow(cid: str, payload: Dict[str, Any]):
    try:
        shadow_resp = await shadow.chat.completions.create(**payload)
        await persist(cid, payload, None, shadow_resp)
    except Exception as e:
        await persist_error(cid, e)

If your request volume exceeds what a single process can handle, push the shadow payload to a Redis queue and let a worker pool make the calls. The principle is identical: primary path stays synchronous, shadow path is decoupled.

Step 3: Isolate shadow credentials and routing

Never point shadow traffic at the same API key as production. Create a separate project with independent rate limits. Exceeding quota on the shadow side should surface as a metric, not a user error.

A gateway simplifies this. For example, n4n.ai exposes one OpenAI-compatible endpoint that addresses 240+ models and automatically falls back when a provider is rate-limited or degraded. It also provides per-token usage metering, so you can attribute shadow cost separately from production. Your shadow client then becomes:

shadow = openai.AsyncOpenAI(
    api_key=GATEWAY_KEY,
    base_url="https://api.n4n.ai/v1"
)
SHADOW_HEADERS = {"x-n4n-model": "anthropic/claude-3.5-sonnet"}

Pass SHADOW_HEADERS via the default_headers argument. The gateway forwards the directive and handles provider failover; your code stays unchanged if you later switch the candidate.

Step 4: Mirror payloads exactly

A shadow test is only valid if the candidate receives the same input. Copy the entire messages array, including system prompts and few-shot examples. Match temperature, max_tokens, top_p, stop, and seed (if the model honors it). Strip provider-only extensions like logit_bias unless both sides support the same format.

def normalize(payload: Dict[str, Any]) -> Dict[str, Any]:
    keep = {"model", "messages", "temperature", "max_tokens",
            "top_p", "stop", "seed", "frequency_penalty", "presence_penalty"}
    return {k: v for k, v in payload.items() if k in keep}

Log the normalized payload. You will replay it later if you need to debug a specific failure.

Step 5: Persist request/response pairs

Use a schema that keeps the correlation ID, normalized payload, primary response (optional, you may already have it in prod logs), shadow response, and timestamps. A JSONL file is enough for most migrations; a table is better if you want SQL queries.

import json, time

async def persist(cid, payload, primary, shadow):
    record = {
        "cid": cid,
        "payload": normalize(payload),
        "primary": primary.model_dump() if primary else None,
        "shadow": shadow.model_dump() if shadow else None,
        "ts": time.time()
    }
    with open("/var/log/shadow/pairs.jsonl", "a") as f:
        f.write(json.dumps(record) + "\n")

Include the same cid in your primary response logs. That lets you join production latency with shadow quality after the fact.

Step 6: Run offline comparison

After a day of mirrored traffic, load the pairs. Start with an automated similarity signal. Embedding cosine similarity is cheap and catches gross divergence.

from sentence_transformers import SentenceTransformer
import numpy as np, json

emb = SentenceTransformer("all-MiniLM-L6-v2")

def similarity(a, b):
    ea, eb = emb.encode([a, b])
    return float(np.dot(ea, eb) / (np.linalg.norm(ea) * np.linalg.norm(eb)))

records = [json.loads(l) for l in open("/var/log/shadow/pairs.jsonl")]
scores = []
for r in records:
    p = r["primary"]["choices"][0]["message"]["content"]
    s = r["shadow"]["choices"][0]["message"]["content"]
    scores.append(similarity(p, s))

print("median similarity:", sorted(scores)[len(scores)//2])

python score_similarity.py

Any pair below min_similarity goes to a human review queue. Look for patterns: does the shadow model refuse more often? Does it hallucinate numbers? Those are migration blockers.

Step 7: Check latency and error distributions

Shadow traffic testing for LLM migration must capture tail latency. Compute percentiles from the shadow response timestamps.

import statistics
ms = [r["shadow_ms"] for r in records if "shadow_ms" in r]
p95 = statistics.quantiles(ms, n=20)[-1]

Categorize errors: rate limit (429), timeout, content filter, malformed response. If the gateway used fallback, separate those retries from pure model latency so you don’t penalize the candidate for network hiccups.

Step 8: Verify success

Create a verification script that asserts your Step 1 criteria against the collected data.

def verify(records, cfg):
    errs = sum(1 for r in records if r.get("error"))
    rate = errs / len(records)
    assert rate <= cfg["max_error_rate"], f"error rate {rate}"
    assert p95 <= cfg["max_latency_ms_p95"]
    assert median_sim >= cfg["min_similarity"]
    return True

If verify passes, you have data-backed confidence. If it fails, adjust the shadow model, prompt template, or parameters and re-run. Do not skip this step because the new model “feels better.”

Step 9: Cut over with a canary

Flip traffic gradually. With a gateway that honors client routing directives, set a header per request based on a random rollout weight.

import random
def route_headers():
    if random.random() < 0.05:
        return {"x-n4n-model": "anthropic/claude-3.5-sonnet"}
    return {"x-n4n-model": "openai/gpt-4o-mini"}

Because the gateway forwards provider cache-control hints, cached system prompts keep working during the canary. Watch error dashboards for the 5% slice. If clean for 24 hours, move to 25%, then 100%.

Step 10: Tear down and keep the harness

After a stable week, remove the shadow duplicate calls. Compress the JSONL to an object store for regression testing. Revoke shadow credentials.

Shadow traffic testing for LLM migration is a repeatable discipline, not a one-off project. Provider behavior drifts; keep the interception code behind a feature flag so you can re-enable it before the next switch.