ZerdaZerda

ab_testing — Offline Prompt Variant Comparison

On this page

Compares two or more prompt variants against the same set of queries using an LLM-as-judge approach. This module has no import dependency on the rest of the prompt package — it works with any object exposing a generate(prompt: str) -> str method (duck-typed llm), and optionally a rag_system exposing retrieve(query) and (optionally) context_manager.build(query, chunks).

Table of contents

PromptVariant

Python · 5 lines
@dataclass
class PromptVariant:
    name:        str
    template:    Any      # str | Callable[[str, str], str]
    description: str = ""

build()

Python
def build(self, context: str, question: str) -> str

If self.template is callable, calls template(context, question). Otherwise treats it as a format string and calls .format(context=context, question=question); if that raises KeyError (i.e. the template uses {query} instead of {question}), retries with .format(context=context, query=question). Any other KeyError-unrelated placeholder (e.g. a typo'd {contxt}) will propagate as an uncaught KeyError from the second .format() call.

Python · 6 lines
v = PromptVariant(
    name="cot",
    template="Context: {context}\n\nThink step by step, then answer: {question}\nAnswer:",
    description="Chain-of-thought prompting",
)
prompt_str = v.build(context="...", question="What is RAG?")

TrialResult

Python · 8 lines
@dataclass
class TrialResult:
    variant_name:  str
    query:         str
    answer:        str
    score:         float
    reasoning:     str
    latency_sec:   float

Plain data container — one per (variant, query) pair, produced by ABTestRunner.run().

QueryComparison

Python · 5 lines
@dataclass
class QueryComparison:
    query:        str
    ground_truth: Optional[str]
    results:      List[TrialResult]

best_variant()

Python
def best_variant(self) -> Optional[str]

Returns the variant_name of the highest-score TrialResult, or None if results is empty. Ties are broken by max()'s stable behavior — the first result achieving the max score in list order wins, not a random or explicit tie-break rule.

ABExperiment

Python · 6 lines
@dataclass
class ABExperiment:
    experiment_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    variant_names: List[str] = field(default_factory=list)
    comparisons:   List[QueryComparison] = field(default_factory=list)
    created_at:    float = field(default_factory=time.time)

experiment_id is an 8-character prefix of a UUID4 — collision risk exists (not cryptographically namespaced), though vanishingly unlikely for typical usage volumes.

mean_score(variant_name)

Python
def mean_score(self, variant_name: str) -> Optional[float]

statistics.mean() of all scores across all comparisons for that variant, rounded to 4 decimals; None if there are zero matching scores (avoids statistics.mean([]) raising StatisticsError).

win_rate(variant_name)

Python
def win_rate(self, variant_name: str) -> float

Fraction of queries where this variant achieved the max score for that query. Ties split evenly: if two variants tie for the max on a query, each gets 1 / len(winners) credit rather than 0 or 1. Returns 0.0 if there are no comparisons (not None) — this differs from mean_score's None-on-empty behavior, so callers should not assume a uniform "empty" sentinel across these two methods.

all_scores()

Python
def all_scores(self) -> Dict[str, Dict[str, Any]]

Returns, per variant name: {"mean_score", "win_rate", "std", "n"}. std uses statistics.stdev() (sample standard deviation) and is None unless at least 2 scores exist (guards StatisticsError from stdev() on fewer than 2 data points).

winner()

Python
def winner(self) -> Optional[str]

max(variant_names, key=lambda n: (stats[n]["win_rate"], stats[n].get("mean_score") or 0.0)) — primary sort key is win rate, tie-broken by mean score. Returns None only if variant_names is empty.

to_dict()

Serializes experiment_id, variant_names, created_at, num_queries (len(comparisons)), scores (all_scores()), and winner.

Prints a formatted table to stdout with box-drawing characters, one row per variant showing mean score / win rate / std, and a trophy-emoji winner line. Pure side-effecting console output — returns None.

ABTestRunner

Python · 7 lines
ABTestRunner(
    llm: Any,
    variants: List[PromptVariant],
    rag_system: Optional[Any] = None,
    judge_llm: Optional[Any]  = None,
    batch_delay_sec: float    = 0.0,
) -> None

Constructor validation

Raises ValueError if len(variants) < 2 — "A/B testing requires at least 2 variants." This is the only validation error raised anywhere in this module.

ParameterNotes
llmMust expose .generate(prompt: str) -> str. Duck-typed; no isinstance check
variants≥ 2 PromptVariant instances
rag_systemOptional. Must expose .retrieve(query); if it also has a .context_manager attribute with .build(query, chunks), that's used to format retrieved chunks into context
judge_llmDefaults to llm if not given — the same model can generate and judge, or you can pass a stronger model as judge
batch_delay_secSleep (via time.sleep) between queries in run(), for rate-limit pacing

_get_context(query) (internal)

Python
def _get_context(self, query: str) -> str

Returns "" if rag_system is None. Otherwise calls rag_system.retrieve(query); if the result is truthy and rag_system has a context_manager attribute, delegates formatting to rag_system.context_manager.build(query, chunks) — note this call signature (build(query, chunks)) does not match ContextManager.build(request) from context_manager.py, which takes a single PromptRequest. This implies ABTestRunner expects a different, externally-supplied RAG system object with its own, differently-shaped context_manager, not zerda.prompt.ContextManager directly — using this module against zerda.prompt.ContextManager as rag_system.context_manager would raise a TypeError for a mismatched signature. If context_manager isn't present, falls back to joining getattr(c, "text", str(c)) for the first 5 (chunk, score) tuples — this fallback path itself assumes chunks is a list of 2-tuples, which is a further, different assumption about rag_system.retrieve()'s return shape. Any exception during retrieval is caught, logged as a warning, and "" is returned instead of propagating.

_generate(variant, context, query) (internal)

Builds the prompt via variant.build(context, query), times a call to self.llm.generate(prompt), and returns (answer, latency_sec). If .generate() raises, the exception is caught, logged as an error, and answer="" is used instead (latency is still recorded as time-to-failure).

_judge(query, answer, ground_truth) (internal)

Python
def _judge(self, query, answer, ground_truth) -> Tuple[float, str]
  1. If answer is empty, returns (0.0, "Empty answer.") immediately — no judge call.
  2. Detects Arabic via _is_arabic(query) (checks for any Unicode codepoint in U+0600–U+06FF) and selects the bilingual judge prompt template (_JUDGE_PROMPT_AR / _JUDGE_PROMPT_EN).
  3. Computes a heuristic fallback score via token-overlap (_tok_overlap) between query/answer, blended 50/50 with answer/ground-truth overlap if a ground truth was supplied.
  4. Calls self.judge.generate(judge_prompt), parses the response via _parse_score() (see below), using the heuristic as the default if parsing fails.
  5. Returns a blended score: 0.7 * judge_score + 0.3 * heuristic, rounded to 4 decimals.
  6. If self.judge.generate() itself raises, catches it, logs a warning, and returns (heuristic, "Heuristic (LLM judge unavailable)") — i.e. the system degrades gracefully to pure heuristic scoring rather than failing the whole experiment.

run()

Python · 7 lines
def run(
    self,
    queries: List[str],
    ground_truths: Optional[List[Optional[str]]] = None,
    contexts: Optional[List[str]] = None,
    show_progress: bool = False,
) -> ABExperiment

For each query (optionally paired with a ground truth and/or pre-built context), generates an answer and judges it for every variant, appends a TrialResult per variant, and appends one QueryComparison per query to the returned ABExperiment. Sleeps batch_delay_sec seconds between queries if configured.

Parameters

NameTypeNotes
queriesList[str]Required
ground_truthsOptional[List[Optional[str]]]Must align positionally with queries if provided; defaults to a list of None
contextsOptional[List[str]]If provided, overrides automatic retrieval per-query — used to bypass rag_system entirely (e.g. in tests)
show_progressboolPrints one line per query to stdout if True

Raises: no explicit validation of len(ground_truths) == len(queries) — zip() silently truncates to the shortest list if the lengths mismatch, meaning trailing queries could be silently dropped from the experiment without any error or warning.

Complexity: O(len(queries) × len(variants)) LLM calls total (one generation + one judge call per pair, unless the answer is empty).

Side effects: LLM API calls (external, via injected llm/judge_llm), optional time.sleep, optional stdout printing, logger.debug/logger.info calls.

compare_two()

Python · 3 lines
def compare_two(
    self, query: str, ground_truth: Optional[str] = None, context: Optional[str] = None,
) -> QueryComparison

Convenience wrapper: calls self.run(queries=[query], ...) and returns exp.comparisons[0].

Python · 2 lines
cmp = runner.compare_two("What is the refund policy?")
print(cmp.best_variant(), cmp.results[0].score, cmp.results[1].score)

Thread safety

Not thread-safe for concurrent run() calls on the same ABTestRunner instance sharing one llm/judge_llm object, since correctness depends entirely on the thread-safety of the injected llm/judge_llm/rag_system objects, which this module has no control over or knowledge of.

Module-level helpers

_tok_overlap(a, b) -> float

Jaccard similarity between the \w+-regex word sets of two strings (lowercased). Returns 0.0 if either set is empty.

_parse_score(text, default=0.5) -> Tuple[float, str]

  1. Strips ```json/``` fences, attempts json.loads, and if successful reads obj.get("score", default) clamped to [0.0, 1.0] via max(0.0, min(1.0, score)), and obj.get("reasoning", "").
  2. On any parse failure, falls back to a regex scan for bare decimal/integer scores in [0, 1] (\b0?\.\d+\b|\b1\.0\b|\b[01]\b) and returns the first valid match found, with the entire original text as the "reasoning".
  3. If nothing matches, returns (default, text.strip()).

Examples

1. Basic — two static-template variants

Python · 6 lines
from zerda.prompt.ab_testing import PromptVariant, ABTestRunner

v1 = PromptVariant(name="concise", template="Context: {context}\n\nQ: {question}\nA (brief):")
v2 = PromptVariant(name="detailed", template="Context:\n{context}\n\nQuestion: {question}\nDetailed answer:")

runner = ABTestRunner(llm=my_llm, variants=[v1, v2])

2. Intermediate — run against explicit contexts (no RAG system)

Python · 5 lines
exp = runner.run(
    queries=["What is the refund policy?"],
    contexts=["Refunds are processed within 14 business days of the return request."],
)
exp.print_report()

3. Advanced — with a RAG system and ground truths

Python · 5 lines
exp = runner.run(
    queries=["ما هي سياسة الإجازات؟", "What is the refund policy?"],
    ground_truths=["30 يوم في السنة", "14 business days"],
)
print(exp.winner(), exp.all_scores())

4. Error handling — fewer than 2 variants

Python · 4 lines
try:
    ABTestRunner(llm=my_llm, variants=[v1])
except ValueError as e:
    print(e)  # "A/B testing requires at least 2 variants."

5. Production — a callable variant with structured judging

Python · 8 lines
def cot_template(context: str, question: str) -> str:
    return f"Context:\n{context}\n\nThink step by step.\n\nQ: {question}\nA:"

v_cot = PromptVariant(name="cot", template=cot_template)
runner = ABTestRunner(llm=production_llm, judge_llm=stronger_judge_llm, variants=[v1, v_cot])
exp = runner.run(queries=eval_queries, ground_truths=eval_answers, show_progress=True)
if exp.mean_score("cot") and exp.mean_score("cot") > exp.mean_score("concise"):
    promote_variant("cot")

Best practices

  • Pass judge_llm as a stronger/more reliable model than the one under test, to reduce judge bias toward whatever style the generating model favors.
  • Use contexts= explicitly in CI/test environments rather than relying on rag_system — it avoids the signature-mismatch pitfall described in _get_context above and makes tests deterministic.

Common mistakes

  • Passing zerda.prompt.ContextManager as rag_system.context_manager and expecting _get_context to work — the method signatures don't match (ContextManager.build(request) vs. the build(query, chunks) this module expects).
  • Assuming ground_truths/contexts length mismatches with queries will raise an error — they don't; zip() truncates silently.

Related pages