The fastest way to A/B test prompts without duplicate codepaths is to treat prompt variants as data, not logic. When you scatter if experiment == "variant_b" checks across your request handlers, you couple deployment lifecycle to marketing guesses and make rollback painful; a single wrapper that resolves the active variant from a config store keeps your business logic oblivious to the experiment.
Step 1: Model prompt variants as configuration
Start by extracting every prompt string out of source code into a versioned config file. JSON, YAML, or a row in a feature-flag DB all work. The shape matters more than the storage: each experiment has a name, a set of variants, and a traffic split.
{
"experiments": {
"checkout_summary": {
"variants": [
{
"id": "control",
"prompt": "Summarize the order:\n{{order}}",
"model": "gpt-4o-mini"
},
{
"id": "variant_a",
"prompt": "You are a concise clerk. Summarize the following order:\n{{order}}",
"model": "gpt-4o-mini"
}
],
"split": { "control": 0.5, "variant_a": 0.5 }
}
}
}Keep the template syntax minimal. I use {{variable}} because it is trivial to render with str.replace or a tiny Jinja environment. Do not pull in a full templating engine unless you need loops or conditionals—those belong in code, not prompts.
Store this file in your repo or a config service. If you use a runtime flag service (LaunchDarkly, Unleash, or a homegrown KV), mirror the same schema. The key win: changing a prompt or a split weight is a config edit, not a pull request to your inference path. Version the file so you can audit which prompt shipped on which day.
One more rule: never embed business logic in the variant. If variant_a needs a different output parser, that parser difference is a code change, not a prompt change. Keep variants limited to text and model name.
Step 2: Assign variants with deterministic hashing
Random assignment per request causes the same user to see different variants on reload, poisoning your metrics. Hash the user ID (or session ID) so assignment is stable.
import hashlib
def assign_variant(user_id: str, split: dict) -> str:
h = int(hashlib.sha256(user_id.encode()).hexdigest(), 16)
bucket = h % 100
cumulative = 0
for variant, weight in split.items():
cumulative += int(weight * 100)
if bucket < cumulative:
return variant
return list(split.keys())[0]This function is pure and side-effect free. It returns "control" or "variant_a" based on the hashed bucket. You can unit test it with known IDs to assert the split approximates your weights.
Avoid random.choice weighted by split at request time. Determinism also lets you replay logs and debug exactly what a given user saw last Tuesday. If you worry about hash inversion, append a static salt to the user ID before hashing; the salt lives in config alongside the experiment.
Step 3: Build a single inference wrapper
Write one function that takes the experiment name, a stable user identifier, and the template variables. It loads config, assigns the variant, renders the prompt, and calls the model. That is the only place where LLM calls for experiments live.
from openai import OpenAI
import os, json
client = OpenAI(
base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
api_key=os.getenv("LLM_API_KEY")
)
def load_config() -> dict:
with open("experiments.json") as f:
return json.load(f)
def render(template: str, vars: dict) -> str:
out = template
for k, v in vars.items():
out = out.replace("{{" + k + "}}", str(v))
return out
def run_experiment(exp_name: str, user_id: str, vars: dict) -> dict:
cfg = load_config()["experiments"][exp_name]
variant_id = assign_variant(user_id, cfg["split"])
variant = next(v for v in cfg["variants"] if v["id"] == variant_id)
prompt = render(variant["prompt"], vars)
resp = client.chat.completions.create(
model=variant["model"],
messages=[{"role": "user", "content": prompt}],
extra_headers={
"x-experiment": exp_name,
"x-variant": variant_id
}
)
return {
"text": resp.choices[0].message.content,
"variant": variant_id,
"model": variant["model"],
"usage": resp.usage.model_dump()
}The extra_headers are optional but useful: they let your logging proxy tag requests without parsing the body. If you route through n4n.ai, the same OpenAI-compatible call addresses 240+ models and automatically falls back when a provider is rate-limited, so the wrapper never needs provider-specific branches.
Notice there is no if exp_name == "checkout_summary" inside your route handler. The handler calls run_experiment(...) and gets a result. That is how you A/B test prompts without duplicate codepaths.
If you need streaming, wrap the stream iterator similarly and attach the same headers. Do not create a second function for the experiment case.
Step 4: Instrument and log outcomes
An experiment is worthless without measurement. Wrap the call with structured logging that captures variant, latency, token usage, and the downstream business metric.
import logging, time
logger = logging.getLogger("exp")
def run_and_log(exp_name: str, user_id: str, vars: dict, outcome_fn):
start = time.time()
try:
res = run_experiment(exp_name, user_id, vars)
res["latency_ms"] = int((time.time() - start) * 1000)
res["outcome"] = outcome_fn()
logger.info("experiment_result", extra=res)
return res
except Exception as e:
logger.error("experiment_failed", extra={"exp": exp_name, "err": str(e)})
raiseoutcome_fn should return whatever you optimize for: a boolean purchased, a rating, or a parsed JSON score. Push these logs to your warehouse or a simple JSONL file for offline analysis. Gateways like n4n.ai return per-token usage metering in the standard usage object, so your cost analysis needs no extra instrumentation.
If your gateway honors provider cache-control hints, add extra_headers={"cache-control": "max-age=300"} to the create call for read-heavy experiments. This reduces redundant spend on identical prompts without changing your code structure.
Step 5: Evaluate and promote
After a few days of traffic, pull the logs and compute per-variant performance.
import pandas as pd
from statsmodels.stats.proportion import proportions_ztest
df = pd.read_json("experiment_logs.jsonl", lines=True)
agg = df.groupby("variant").agg(
conversions=("outcome", "mean"),
p50_latency=("latency_ms", lambda x: x.quantile(0.5)),
total_tokens=("usage", lambda s: sum(u.get("total_tokens", 0) for u in s))
)
print(agg)
control = df[df.variant == "control"]["outcome"]
variant = df[df.variant == "variant_a"]["outcome"]
count = [len(control), len(variant)]
success = [control.sum(), variant.sum()]
stat, pval = proportions_ztest(success, count)
print(f"z={stat:.3f}, p={pval:.4f}")Look at the conversion difference and its p-value. For binary outcomes, a two-proportion z-test is enough; don’t overcomplicate with ML. If variant_a beats control at p < 0.05 and latency is acceptable, promote it by editing the config: set split to {"variant_a": 1.0} or simply delete the experiment and inline the winning prompt if the test is done.
Rollback is equally a config change. If a variant shows regressions, flip the split back to control in seconds.
How to verify success
You have cleanly implemented the pattern when:
- Code grep is empty. Search your handlers for the experiment name or variant IDs. You should find zero conditionals. All references live in
experiments.jsonand the wrapper. - Both variants appear in logs. Query your log store for
x_variantvalues; both should have traffic proportional to the split. - Metrics compute without joins to code. Your analysis script reads only logs, not a separate hardcoded map of variants.
- No provider forks. The same
client.chat.completions.createcall serves all variants. If a provider degrades, your gateway handles fallback; your code does not.
Following these steps lets you A/B test prompts without duplicate codepaths in an afternoon. The discipline is boring on purpose: prompts become data, assignment becomes a hash, and your service code stays deaf to the experiment. That is the only scalable way to run continuous prompt optimization against production traffic.