How to A/B test models inside a live agent pipeline

When you run a production agent, swapping the underlying LLM is a high-risk change. To A/B test models agent pipeline safely, you need a routing shim, consistent evaluation, and per-request telemetry—not a fork in your codebase. This guide walks through a concrete implementation you can drop into an existing async agent without rewriting your tool-calling logic.

Step 1: Define the experiment and variants

Start by writing down the hypothesis and the traffic split. Pick a control model (usually your current production model) and a treatment model (the candidate). Keep the split small—5–10% to the treatment—until you have confidence in the candidate’s behavior on real inputs.

Store this in a config file your pipeline loads at startup. Keep it declarative so ops can change weights without a deploy:

{
  "experiment": "model-swap-2025-03",
  "control": "gpt-4o-mini",
  "treatment": "claude-3-5-haiku",
  "treatment_weight": 0.1,
  "metrics": ["latency_ms", "tokens_total", "task_success", "cost_usd"]
}

The weights must sum to 1 across control and treatment. If you later expand to multiple candidates, use a map of model name to weight and normalize at assignment time. Resist the urge to run a multi-armed bandit on day one; a fixed split is easier to reason about and debug.

Step 2: Add a routing shim to your agent pipeline

The shim is a pure function that decides which model handles a given request. Determinism matters: the same user or session should hit the same variant across calls, or you poison your metrics with cross-talk between models.

Hash a stable identifier (user_id, session_id, or request_id if you want per-request randomization) and mod it into a bucket. For agent pipelines, sticky assignment by user_id is usually correct—it prevents a single conversation from bouncing between model personalities.

import hashlib

def assign_variant(id_str: str, treatment_weight: float) -> str:
    # returns "control" or "treatment"
    h = hashlib.sha256(id_str.encode()).hexdigest()
    bucket = int(h[:8], 16) / 0xffffffff  # 0..1
    return "treatment" if bucket < treatment_weight else "control"

def resolve_model(id_str: str, cfg: dict) -> str:
    variant = assign_variant(id_str, cfg["treatment_weight"])
    return cfg["treatment"] if variant == "treatment" else cfg["control"]

Wire resolve_model into the agent’s model-selection step. Do not branch on model name anywhere else; the rest of the pipeline calls a generic complete(). If you use a framework like LangGraph or Autogen, override the model binding at the node level with the resolved name.

Step 3: Call models through a single OpenAI-compatible endpoint

Your agent should speak one protocol. If you use a gateway that exposes an OpenAI-compatible endpoint, the only difference between variants is the model string. Using a gateway such as n4n.ai gives you one OpenAI-compatible endpoint for 240+ models and automatic fallback if a provider is degraded, so your shim only changes the model name rather than the client.

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.n4n.ai/v1",  # OpenAI-compatible gateway
    api_key="sk-...",
)

async def complete(model: str, messages: list, **kwargs):
    resp = await client.chat.completions.create(
        model=model,
        messages=messages,
        **kwargs,
    )
    return resp

If you need to force a specific provider or pass cache hints, forward them as extra headers. The gateway honors client routing directives and forwards provider cache-control hints, which keeps prompt caching working across both variants.

headers = {
    "x-n4n-route": "provider:anthropic",  # optional explicit route
    "x-n4n-cache": "ttl:3600",
}
resp = await client.chat.completions.create(
    model=model,
    messages=messages,
    extra_headers=headers,
)

For streaming responses, count tokens via the final usage chunk rather than the input tokenizer; some providers only report usage at stream end. Your complete() wrapper should aggregate that before returning.

Step 4: Emit per-request telemetry with variant tag

Every request must log the assigned variant, model, latency, token counts, and a success signal. Without this, you cannot A/B test models agent pipeline with any rigor. The variant tag is the pivot for all downstream analysis.

Use structured logging and include a correlation id that flows to the gateway:

import time, structlog, uuid
log = structlog.get_logger()

async def agent_turn(user_id: str, messages: list, cfg: dict):
    model = resolve_model(user_id, cfg)
    corr_id = uuid.uuid4().hex
    t0 = time.monotonic()
    try:
        resp = await complete(model, messages, temperature=0.2,
                              extra_headers={"x-correlation-id": corr_id})
        dt = (time.monotonic() - t0) * 1000
        usage = resp.usage
        log.info("agent.turn",
            experiment=cfg["experiment"],
            variant="treatment" if model == cfg["treatment"] else "control",
            model=model,
            correlation_id=corr_id,
            latency_ms=dt,
            prompt_tokens=usage.prompt_tokens,
            completion_tokens=usage.completion_tokens,
            total_tokens=usage.total_tokens,
            success=True,
        )
        return resp.choices[0].message.content
    except Exception as e:
        log.error("agent.turn.failed",
            experiment=cfg["experiment"],
            model=model,
            correlation_id=corr_id,
            error=str(e),
            success=False,
        )
        raise

Feed these logs to your metrics backend (Prometheus, BigQuery, ClickHouse). Make sure both variants emit identical fields; a missing total_tokens on one side will silently break cost comparison.

Step 5: Collect and analyze results

After a few days of traffic, pull the aggregated numbers. Compute per-variant:

• Median and p95 latency
• Total and per-success token cost
• Task success rate (defined by your agent’s eval, e.g., JSON parse rate, user thumbs-up)

A minimal pandas slice:

import pandas as pd

df = pd.read_parquet("agent_turns.parquet")
grp = df.groupby("variant").agg(
    n=("success", "size"),
    success_rate=("success", "mean"),
    p95_latency=("latency_ms", lambda s: s.quantile(0.95)),
    avg_tokens=("total_tokens", "mean"),
)
print(grp)

Cost normalization across tokenizers

Token counts are not comparable across model families. Normalize by billed cost: multiply total_tokens by the per-token price for that specific model from your provider invoice or gateway metering. If your gateway provides per-token usage metering, use those numbers rather than the API’s raw counts to avoid surprises.

If the treatment shows equal or better success at lower cost or latency, promote it. If not, kill the experiment. The point of choosing to A/B test models agent pipeline is to get this evidence before a full rollout. For statistical significance on success rate, use a two-proportion z-test. Don’t trust a 1% lift on 50 requests.

Step 6: Roll out or rollback

To promote, set treatment_weight to 1.0 and rename treatment to control in config. Keep the shim for the next test—model swapping is never one-and-done.

If metrics degrade, set weight to 0. The gateway’s automatic fallback already covers provider outages; your experiment rollback covers model-quality regressions. Run a canary check post-rollout: keep 1% on the old model for 24 hours and alert if error rate diverges.

Step 7: Automate the experiment lifecycle

Once the manual loop works, automate it. A cron job can shift treatment_weight based on intermediate metrics, and a CI check can reject configs where weights don’t sum to 1. Emit an experiment report to Slack when the z-test crosses significance.

# example github action trigger
name: ab-report
on:
  schedule:
    - cron: "0 6 * * *"
jobs:
  report:
    runs-on: ubuntu-latest
    steps:
      - run: python scripts/ab_summary.py --experiment model-swap-2025-03

Automation turns a one-off test into a durable capability. When the next model drops, you flip a weight instead of rewriting call sites.

How to verify success

You have a working A/B test models agent pipeline setup when:

1. Traffic split matches config. Query your logs: df.variant.value_counts(normalize=True) should show ~10% treatment if that was the weight.
2. Both variants log identical fields. Missing fields in one variant invalidate comparisons.
3. Latency and cost per success are computable. You can produce the table from Step 5 without joins to external systems.
4. No error-rate spike. The treatment’s exception rate is within 0.5% of control.
5. Decision recorded. You write down the win/lose call and the config change that enacts it.

Run the shim in staging against recorded production prompts first. Replay a thousand turns through both variants and confirm the telemetry emits before touching live users.

Practical caveats

Model outputs are non-deterministic. Set temperature=0 for the eval slice if you need reproducibility, but keep production temperature as intended—you are testing the real experience.

Prompt caching behaves differently per provider. When you A/B test models agent pipeline across vendors, confirm cache hit rates via the usage telemetry; a slower treatment might just be missing cache hints.

Finally, watch token accounting. Some gateways meter per-token usage; discrepancies in reported total_tokens between models are expected because tokenizers differ. Normalize cost by your actual billed usage, not by raw token counts alone.

That’s the whole loop: config, shim, single endpoint, telemetry, analysis, rollout, automation. Do it for every model change and your agent stops breaking when the LLM landscape shifts.