Running a controlled experiment between two model providers used to mean writing a custom load balancer. A/B testing models with n4n routing collapses that into a single OpenAI-compatible endpoint and a small client wrapper—you point at n4n.ai, tag each request with a variant, and read per-token usage to compare cost and latency.
Prerequisites
- Python 3.11+
openai>=1.30- An n4n.ai API key exported as
N4N_API_KEY - Two candidate models from the 240+ available (we’ll use
openai/gpt-4o-miniandanthropic/claude-3-haiku) pandasandscipyif you want significance testing
Step 1: Configure the client
Point the OpenAI SDK at the gateway. One key, one base URL.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.n4n.ai/v1",
api_key=os.environ["N4N_API_KEY"],
)Step 2: Define the experiment
We hash the user ID to assign a stable variant. This keeps a user in the same bucket across requests and avoids per-call randomness that corrupts join keys.
import hashlib
MODELS = {
"control": "openai/gpt-4o-mini",
"treatment": "anthropic/claude-3-haiku",
}
def assign_variant(user_id: str) -> str:
bucket = int(hashlib.sha256(user_id.encode()).hexdigest(), 16) % 100
return "treatment" if bucket < 50 else "control"Why hashing instead of random
Random assignment per call skews metrics when a user issues multiple queries. Hashing gives you persistent buckets without a stateful store. If you later join with user-level outcomes (retention, conversion), the bucket must be deterministic.
Step 3: Send tagged requests
n4n.ai honors client routing directives, so we forward the variant as a routing hint. The gateway passes it through and applies any server-side rules mapped to that tag.
def chat(user_id: str, prompt: str) -> dict:
variant = assign_variant(user_id)
model = MODELS[variant]
resp = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
extra_headers={"x-n4n-route": variant}, # client routing directive
temperature=0.2,
)
return {
"variant": variant,
"model": model,
"content": resp.choices[0].message.content,
"usage": resp.usage.model_dump(),
}Expected output for chat("user_42", "Summarize: ..."):
{
"variant": "control",
"model": "openai/gpt-4o-mini",
"content": "The text describes a distributed system...",
"usage": {"prompt_tokens": 12, "completion_tokens": 34, "total_tokens": 46}
}Step 4: Log metrics for analysis
Per-token usage metering lets you attribute spend precisely. Wrap the call with latency timing and append to a ledger.
import time, json
ledger = []
def logged_chat(user_id: str, prompt: str) -> dict:
start = time.perf_counter()
row = chat(user_id, prompt)
row["lat_ms"] = (time.perf_counter() - start) * 1000
row["ts"] = time.time()
ledger.append(row)
return row
logged_chat("user_42", "Explain CAP theorem in one sentence")After a few hundred requests you have a dataset with variant, model, token counts, and latency. Dump it:
with open("ab_ledger.jsonl", "w") as f:
for r in ledger:
f.write(json.dumps(r) + "\n")Step 5: Compare variants
Aggregate in Python or load into pandas.
from collections import defaultdict
stats = defaultdict(lambda: {"calls": 0, "prompt_tok": 0, "comp_tok": 0, "lat_ms": 0})
for row in ledger:
s = stats[row["variant"]]
s["calls"] += 1
s["prompt_tok"] += row["usage"]["prompt_tokens"]
s["comp_tok"] += row["usage"]["completion_tokens"]
s["lat_ms"] += row["lat_ms"]
for v, s in stats.items():
print(f"{v:10} n={s['calls']:4} avg_prompt={s['prompt_tok']/s['calls']:.1f} "
f"avg_comp={s['comp_tok']/s['calls']:.1f} avg_lat={s['lat_ms']/s['calls']:.0f}ms")Sample output:
control n= 512 avg_prompt=18.2 avg_comp=42.1 avg_lat=820ms
treatment n= 488 avg_prompt=17.9 avg_comp=39.7 avg_lat=910msChecking significance
If you collected binary success labels (e.g., user thumbs-up), run a two-proportion z-test.
from scipy.stats import norm
def z_test(success_a, n_a, success_b, n_b):
p_a, p_b = success_a / n_a, success_b / n_b
pool = (success_a + success_b) / (n_a + n_b)
se = (pool * (1 - pool) * (1/n_a + 1/n_b)) ** 0.5
z = (p_b - p_a) / se
return 2 * (1 - norm.cdf(abs(z))) # two-sided p-value
# example: 210/512 vs 230/488
p = z_test(210, 512, 230, 488)
print(f"p-value={p:.3f}")A p-value under 0.05 means the treatment’s win is not noise.
Step 6: Wire fallback safety
One provider will degrade. n4n.ai automatically falls back when a provider is rate-limited or degraded, but you should still catch errors and record them per variant to avoid biased results.
from openai import APIError
def safe_chat(user_id, prompt):
try:
return logged_chat(user_id, prompt)
except APIError as e:
return {"variant": assign_variant(user_id), "error": str(e), "usage": None, "lat_ms": 0}
# errors show up in the ledger as None usage; filter before aggregating
clean = [r for r in ledger if r.get("usage")]Step 7: Optional server-side split
If you don’t want the assignment logic in every service, configure a routing rule on the gateway that maps a virtual model name to a weighted split. Then the client just requests that route:
resp = client.chat.completions.create(
model="route:ab-50-50", # gateway resolves to control/treatment
messages=[{"role": "user", "content": prompt}],
)The gateway stamps the resolved model in the usage metadata, so your ledger still gets the real backend name. This keeps A/B testing models with n4n routing entirely declarative.
Step 8: Promote the winner
Once treatment beats control on your success metric, switch the default model and drop the routing header. Because you used the gateway’s routing, cutover is a config change, not a redeploy.
# post-experiment
MODELS = {"default": "anthropic/claude-3-haiku"}
def chat_final(user_id: str, prompt: str) -> dict:
resp = client.chat.completions.create(
model=MODELS["default"],
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
)
return {"model": MODELS["default"], "content": resp.choices[0].message.content}Gotchas
- Cache hints: If you send provider cache-control hints, n4n.ai forwards them. Don’t let cached responses mask latency differences—measure cache hit rate separately or tag experimental traffic to bypass cache.
- Token accounting: Completion token counts vary by model tokenizer. Compare cost in dollars using the per-token metering export, not raw token sums.
- Sticky buckets: Never assign variant by
random.random()inside the request path if you later join with user-level outcomes. - Sample ratio mismatch: Monitor the actual count of control vs treatment. If it drifts from 50/50, a routing rule or header drop is silently failing.
Full harness
import os, hashlib, time, json
from openai import OpenAI, APIError
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key=os.environ["N4N_API_KEY"])
MODELS = {"control": "openai/gpt-4o-mini", "treatment": "anthropic/claude-3-haiku"}
def assign_variant(uid):
return "treatment" if int(hashlib.sha256(uid.encode()).hexdigest(),16)%100<50 else "control"
def chat(uid, prompt):
v = assign_variant(uid)
r = client.chat.completions.create(model=MODELS[v], messages=[{"role":"user","content":prompt}],
extra_headers={"x-n4n-route": v}, temperature=0.2)
return {"variant":v,"model":MODELS[v],"content":r.choices[0].message.content,"usage":r.usage.model_dump()}
ledger=[]
def logged(uid, prompt):
s=time.perf_counter()
try:
row=chat(uid,prompt); row["lat_ms"]=(time.perf_counter()-s)*1000; row["ts"]=time.time()
except APIError as e:
row={"variant":assign_variant(uid),"error":str(e),"usage":None,"lat_ms":0}
ledger.append(row); return row
# run experiment, then analyze with the snippets aboveYou now have a repeatable harness. The same pattern works for prompt variants, temperature sweeps, or provider migrations. A/B testing models with n4n routing is just disciplined logging plus a stable assignment function.