Shadow traffic testing for LLM models lets you send copies of live production requests to a candidate model while users still get answers from the incumbent. This catches prompt regressions, schema breaks, and latency surprises before a risky cutover.
Why staging lies to you
A staging environment with seeded fixtures tells you the happy path works. It does not tell you what happens when a user pastes a 40-page PDF, uses slang from a community you have never heard of, or hits the endpoint with a malformed tool call. LLM behavior is data-dependent in a way traditional unit tests are not. The only representative dataset is production traffic itself.
Shadow traffic testing for LLM models solves this by mirroring real requests to a second model and recording both outputs. Users never see the shadow response. You get a side-by-side corpus to evaluate before promoting the candidate.
Step 1: Identify the request boundary to fork
Pick the exact layer where you can intercept the completed request payload headed to your primary model. In a typical Python service this is middleware or a wrapper around your inference client. You need the final messages array, temperature, max_tokens, and any tool definitions—not the raw HTTP before your prompt builder runs.
The fork must be asynchronous and must never block the user path. If the shadow call throws, log and drop it.
# fastapi middleware style fork
import asyncio
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
class ShadowFork(BaseHTTPMiddleware):
def __init__(self, app, shadow_sender):
super().__init__(app)
self.shadow_sender = shadow_sender
async def dispatch(self, request: Request, call_next):
body = await request.json()
# user path proceeds normally
response = await call_next(request)
# fire-and-forget shadow after response started
asyncio.create_task(self.shadow_sender.enqueue(body))
return responseVerification: deploy this with the shadow sender disabled (no-op). Confirm p99 latency of the primary path is unchanged under load.
Step 2: Build a shadow client with model override
Your shadow sender takes the captured payload and swaps the model field for the candidate. Keep every other parameter identical. Use an OpenAI-compatible client so you can point at any provider or gateway.
If you route through an OpenAI-compatible gateway such as n4n.ai, you get access to 240+ models behind one endpoint and automatic fallback when a provider is rate-limited—useful when your candidate sits on a flaky beta endpoint. The code below uses the standard openai SDK.
from openai import AsyncOpenAI
import os
client = AsyncOpenAI(
base_url=os.environ["INFERENCE_BASE_URL"], # e.g. https://api.n4n.ai/v1
api_key=os.environ["INFERENCE_KEY"],
)
CANDIDATE_MODEL = "anthropic/claude-3.5-sonnet" # example candidate
async def send_shadow(payload: dict):
shadow_payload = {**payload, "model": CANDIDATE_MODEL}
# drop stream if you only want final text for diffing
shadow_payload["stream"] = False
try:
resp = await client.chat.completions.create(**shadow_payload)
return resp.model_dump()
except Exception as e:
logger.warning("shadow failed: %s", e)
return NoneVerification: run the sender against a small captured batch. Confirm you receive well-formed completions with usage blocks.
Step 3: Sanitize and redact before mirroring
Production payloads contain PII, auth tokens, and customer data. You are obligated to redact before sending to a model you do not fully control. At minimum, strip explicit headers and mask emails, SSNs, and anything in a system prompt that looks like a secret.
import re
EMAIL_RE = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
SSN_RE = re.compile(r"\b\d{3}-\d{2}-\d{4}\b")
def redact_messages(messages):
for m in messages:
if isinstance(m.get("content"), str):
m["content"] = EMAIL_RE.sub("[EMAIL]", m["content"])
m["content"] = SSN_RE.sub("[SSN]", m["content"])
return messages
async def enqueue(self, payload):
payload["messages"] = redact_messages(payload.get("messages", []))
await shadow_queue.put(payload)Verification: inspect 50 redacted payloads manually. Assert zero raw emails or SSNs remain and that the structure is intact for the model call.
Step 4: Store paired responses for offline diff
Do not try to evaluate inline. Write the primary response (already returned to user) and the shadow response to a durable store keyed by a shared request ID. JSONL on disk or a SQLite table works for most teams.
import json, time, sqlite3
conn = sqlite3.connect("shadow_pairs.db")
conn.execute("""CREATE TABLE IF NOT EXISTS pairs (
id TEXT, ts REAL, primary_model TEXT, shadow_model TEXT,
primary_resp TEXT, shadow_resp TEXT)""")
def persist(req_id, primary_resp, shadow_resp, pmodel, smodel):
conn.execute("INSERT INTO pairs VALUES (?,?,?,?,?,?)",
(req_id, time.time(), pmodel, smodel,
json.dumps(primary_resp), json.dumps(shadow_resp)))
conn.commit()Verification: after a day of traffic, query the table. You should have pairs for >95% of production requests (excluding ones where shadow failed transiently).
Step 5: Run evaluation diffs on the pairs
Now the real work: compare. For structured output, validate both responses against your JSON schema. For free text, compute token overlap, length ratios, and run a smaller LLM-as-judge on a sample to flag tone or instruction-following drift.
import json
from jsonschema import validate
schema = {"type": "object", "properties": {"answer": {"type": "string"}}}
def check_pair(primary_resp, shadow_resp):
issues = []
try:
validate(json.loads(primary_resp["choices"][0]["message"]["content"]), schema)
except Exception as e:
issues.append(f"primary schema: {e}")
try:
validate(json.loads(shadow_resp["choices"][0]["message"]["content"]), schema)
except Exception as e:
issues.append(f"shadow schema: {e}")
# latency / token diff
pu = primary_resp["usage"]["total_tokens"]
su = shadow_resp["usage"]["total_tokens"]
if su > pu * 1.5:
issues.append(f"shadow token blowup {pu}->{su}")
return issuesRun this nightly over the new pairs. Emit a report with pass rate and distribution of issue types.
Verification: successfully generate a report showing schema compliance for both models and token delta histogram. If shadow compliance is below 99%, do not proceed.
Step 6: Define launch gates and alerting
Set hard thresholds before you look at the data—otherwise you will rationalize a bad candidate. Concrete gates we use:
- Schema validity: shadow ≥ 99.5% of primary.
- Mean token increase: < 20%.
- p95 latency: shadow < 1.3× primary.
- Judge rejection rate on 200-sample: < 5%.
Wire these into CI or a cron that posts to Slack. If any gate fails, the launch is blocked automatically.
GATES = {
"schema_ok": 0.995,
"token_increase": 0.20,
"p95_latency_ratio": 1.3,
"judge_reject": 0.05,
}
def gate_ok(metrics):
return (metrics["schema_ok"] >= GATES["schema_ok"]
and metrics["token_increase"] <= GATES["token_increase"]
and metrics["p95_latency_ratio"] <= GATES["p95_latency_ratio"]
and metrics["judge_reject"] <= GATES["judge_reject"])Verification: inject a deliberately broken candidate (wrong model name causing schema errors) and confirm the gate fails and alerts fire.
Handling streaming and tool calls
If your production path streams, do not stream the shadow call to the user—but you can still stream internally and collect the full text. For tool-calling models, compare the sequence of function calls, not just final text. Capture tool_calls from each choice and diff arity and argument schemas.
def extract_calls(resp):
msg = resp["choices"][0]["message"]
return [tc["function"] for tc in msg.get("tool_calls", [])]Shadow traffic testing for LLM models with tool use surfaces mismatch in argument naming that would silently break your backend.
Cost and rate limit reality
Shadow traffic doubles your inference spend on the mirrored slice. Sample if needed: fork 10% of requests via a deterministic hash on request ID. You still get statistical power. Also respect provider rate limits—use the gateway’s fallback or queue with backpressure so shadow traffic never competes with real users for quota.
Verifying end-to-end success
A successful shadow run means: (1) production latency is flat, (2) redaction logs show zero leaks, (3) paired corpus covers the sampled traffic, (4) nightly diff report passes all gates, and (5) a staged canary using the candidate at 1% real traffic confirms the shadow numbers hold. Only then promote the model to default.
If you skip the verification steps after each stage, you will discover the regression in production that shadow traffic was supposed to prevent.