Build eval gate component with pluggable evaluator registry

Introduction

Teams that ship GenAI features into production quickly find that response quality is not a property of the model alone — it emerges from the pipeline that wraps every call. When eval gates are bolted on as an afterthought, hallucinations leak into customer-facing surfaces, expensive judge calls run on requests a cheap regex would have rejected, and there is no controlled path to add a new check without touching every caller. By the end of this lesson you'll be able to build an evaluator registry that wires rule-based, embedding-based, and LLM-as-judge checks behind a single pluggable contract and orders them in a cost-ascending cascade with early termination on failure.

Key Terminology

• Eval gate: A load-bearing pipeline component that inspects a prompt-response pair and emits a pass/fail verdict before the response is allowed to propagate downstream.
• Evaluator registry: A central object that maps string names to evaluator instances, enforces uniqueness, and orchestrates cost-ordered execution of evaluator chains.
• Cost-ascending cascade: An execution strategy that runs the cheapest evaluator first and short-circuits on the first failure, so expensive LLM-as-judge calls only run on responses that survived rule-based and embedding-based checks.

Concepts

Core ideas the lesson teaches: a uniform evaluator contract (BaseEvaluator returning EvalResult), a cost-tier taxonomy that orders evaluators from rule-based to LLM-as-judge, and a registry that composes them into cascades with a latency budget and an explicit over-budget policy.

Wiring the Registry

Wiring the registry means three concrete steps that the Code Walkthrough below makes operational. First, every evaluator family — rule-based, embedding-based, LLM-as-judge — subclasses BaseEvaluator and declares a cost_tier, so the registry can sort callers' name lists into cheapest-first order regardless of how they were specified. Second, EvalGateRegistry.register binds an evaluator instance to a unique name, giving callers a stable handle that survives swapping the underlying implementation (e.g. moving from a Haiku-class judge to an Opus-class judge without touching call sites). Third, run_cascade drives the chain end-to-end: it consults cost_tier for ordering, applies the budget_ms ceiling between gates, and breaks on the first failing EvalResult so the expensive tiers never run on responses that have already been rejected.

Code Walkthrough

Core Abstractions: EvalResult and BaseEvaluator

Before building the registry, you need stable contracts for what an evaluator accepts and returns. The two foundational abstractions are the EvalResult dataclass that carries verdict metadata and the BaseEvaluator abstract class that all evaluator families implement. Every evaluator — regardless of whether it executes a regex or calls GPT-4 — must return the same EvalResult structure so that downstream routing logic remains decoupled from evaluator internals.

The following code defines the EvalResult dataclass with fields for the binary pass/fail verdict, a continuous confidence score, the evaluator name for traceability, and an optional metadata dictionary. It also defines the BaseEvaluator abstract base class with an evaluate method that accepts a prompt-response pair and returns an EvalResult, plus a cost_tier property that the registry uses to order evaluators within a cascade.

1from abc import ABC, abstractmethod
2from dataclasses import dataclass, field
3from enum import IntEnum
4from typing import Optional
5
6class CostTier(IntEnum):
7    NEGLIGIBLE = 0   # rule-based: regex, length checks
8    LOW = 1          # embedding: vector similarity
9    MEDIUM = 2       # small LLM judge (Haiku-class)
10    HIGH = 3         # large LLM judge (Opus-class)
11
12@dataclass(frozen=True)
13class EvalResult:
14    passed: bool
15    score: float                          # 0.0 to 1.0 continuous
16    evaluator_name: str
17    latency_ms: float = 0.0
18    metadata: dict = field(default_factory=dict)
19
20    def __post_init__(self):
21        if not 0.0 <= self.score <= 1.0:
22            raise ValueError(f"Score {self.score} outside [0, 1]")
23
24class BaseEvaluator(ABC):
25    def __init__(self, name: str, threshold: float = 0.5):
26        self.name = name
27        self.threshold = threshold
28
29    @property
30    @abstractmethod
31    def cost_tier(self) -> CostTier:
32        ...
33
34    @abstractmethod
35    async def evaluate(
36        self, prompt: str, response: str, context: Optional[dict] = None
37    ) -> EvalResult:
38        ...
39
40    def make_result(self, score: float, **meta) -> EvalResult:
41        return EvalResult(
42            passed=score >= self.threshold,
43            score=score,
44            evaluator_name=self.name,
45            metadata=meta,
46        )

Evaluator Registry Architecture

The registry serves as the single point of registration, resolution, and orchestration for all evaluators in the system. It maps string names to evaluator instances, enforces uniqueness constraints, and provides a run_cascade method that executes evaluators in cost-tier order with early termination on failure.

Loading diagram...

This cascade pattern is the core value proposition of a registry-based design. When a rule-based evaluator detects an obvious format violation — a JSON response missing required fields, a response exceeding length limits, or a safety keyword trigger — the system short-circuits immediately without incurring embedding computation or LLM judge latency. In production systems processing thousands of requests per second, this early termination can reduce average eval overhead by 60–80% because the majority of failures are catchable by cheap heuristics.

The following implementation of EvalGateRegistry provides the register method for adding evaluators, a get method for resolving evaluators by name, and the critical run_cascade method that orchestrates evaluation in cost-ascending order. The cascade accepts an optional budget_ms parameter that enforces a latency ceiling — if cumulative evaluation time approaches the budget, remaining expensive evaluators are skipped and the gate defaults to the configured over_budget_policy.

1import asyncio
2import time
3from typing import Dict, List
4
5class EvalGateRegistry:
6    def __init__(self, over_budget_policy: str = "pass"):
7        self._evaluators: Dict[str, BaseEvaluator] = {}
8        self.over_budget_policy = over_budget_policy  # "pass" or "fail"
9
10    def register(self, evaluator: BaseEvaluator) -> None:
11        if evaluator.name in self._evaluators:
12            raise KeyError(f"Evaluator '{evaluator.name}' already registered")
13        self._evaluators[evaluator.name] = evaluator
14
15    def get(self, name: str) -> BaseEvaluator:
16        if name not in self._evaluators:
17            raise KeyError(f"No evaluator registered as '{name}'")
18        return self._evaluators[name]
19
20    def _sorted_evaluators(self, names: List[str]) -> List[BaseEvaluator]:
21        evals = [self.get(n) for n in names]
22        return sorted(evals, key=lambda e: e.cost_tier)
23
24    async def run_cascade(
25        self,
26        names: List[str],
27        prompt: str,
28        response: str,
29        context: dict = None,
30        budget_ms: float = float("inf"),
31    ) -> List[EvalResult]:
32        ordered = self._sorted_evaluators(names)
33        results: List[EvalResult] = []
34        elapsed_ms = 0.0
35
36        for evaluator in ordered:
37            if elapsed_ms >= budget_ms:
38                default_passed = self.over_budget_policy == "pass"
39                results.append(EvalResult(
40                    passed=default_passed, score=0.0,
41                    evaluator_name=evaluator.name,
42                    metadata={"skipped": True, "reason": "budget_exceeded"},
43                ))
44                continue
45
46            t0 = time.perf_counter()
47            result = await evaluator.evaluate(prompt, response, context)
48            dt = (time.perf_counter() - t0) * 1000
49            elapsed_ms += dt
50            results.append(EvalResult(
51                passed=result.passed, score=result.score,
52                evaluator_name=result.evaluator_name,
53                latency_ms=dt, metadata=result.metadata,
54            ))
55
56            if not result.passed:
57                break  # early termination on first failure
58
59        return results

Concrete Evaluator Implementations

With the registry and base abstractions in place, the three evaluator families become straightforward implementations. A rule-based evaluator executes deterministic checks — JSON schema validation, response length bounds, regex pattern matching for PII or prohibited content. An embedding-based evaluator computes cosine similarity between the response embedding and a reference distribution, catching semantic drift from expected answer patterns. An LLM-as-judge evaluator sends the prompt-response pair to a judge model with a structured rubric and parses the verdict from the judge's output.

Each family subclasses BaseEvaluator and declares its cost_tier. RuleBasedEvaluator accepts a list of Callable[[str], bool] predicates (length bounds, regex matches, PII detectors) at construction, reports CostTier.NEGLIGIBLE, and scores as the fraction of predicates that return True against the response — feeding make_result so threshold comparison stays centralized. EmbeddingEvaluator stores a precomputed reference centroid (the mean embedding of known-good responses) normalized once at construction, takes an async embed_fn that maps text to a numpy vector (so OpenAI, Cohere, or local backends are swappable), and at evaluation time embeds the response, normalizes it, and returns the dot product (cosine similarity for unit vectors) at CostTier.LOW — a similarity of ~0.7 typically indicates semantic alignment with the reference distribution. LLMJudgeEvaluator stores an async judge_fn callable plus a rubric string, formats a judge prompt that interpolates the rubric and prompt-response pair and requests structured JSON (score plus reasoning), parses the verdict, and reports CostTier.HIGH. Separating the rubric from the evaluator class means the same judge class scores faithfulness, coherence, or safety by swapping the rubric — and the reasoning string flows through EvalResult.metadata to support precision-recall analysis when triaging false positives and false negatives on the eval dashboard. You'll know the cascade works when registering all three families and calling run_cascade short-circuits the LLM judge whenever a cheap rule check fails first.

Do's and Don'ts

Do's

1. ✓Do implement the cost_tier abstract property on every BaseEvaluator subclass — The EvalGateRegistry reads this CostTier integer value to sort evaluators before building the cascade; a subclass that omits it raises TypeError at instantiation, and one that declares the wrong tier may execute an LLM-as-judge (CostTier.HIGH) before a rule-based check (CostTier.NEGLIGIBLE), burning large-model tokens on requests a regex would have terminated with early failure.
2. ✓Do call self.make_result(score, **meta) inside every evaluate() implementation rather than constructing EvalResult directly — make_result centralizes the score >= self.threshold comparison so that all evaluator families share identical pass/fail logic, and adjusting an evaluator's operating point requires changing only the threshold constructor argument rather than editing subclass bodies.
3. ✓Do normalize every evaluator's output to a continuous score in [0.0, 1.0] before returning — EvalResult.__post_init__ raises ValueError for out-of-range values, catching miscalibrated evaluators at result-creation time rather than silently corrupting downstream routing; the normalized range also lets you compare rule-based, embedding, and LLM-judge evaluators on the same ROC axis when tuning cascade thresholds.

Don'ts

1. ✗Don't run evaluators outside cost-ascending CostTier order in the registry cascade — Placing a CostTier.HIGH LLM-as-judge before a CostTier.NEGLIGIBLE rule-based check defeats early termination on failure: every request pays large-model latency and token cost even when a cheap regex would have rejected it first, which is the primary failure mode the cascade architecture exists to prevent.
2. ✗Don't encode pass/fail logic inside the evaluate() method itself — Evaluators must compute and return only a continuous score; the binary passed field is derived exclusively by make_result()'s score >= self.threshold check. Duplicating that comparison inside evaluate() breaks the single-knob threshold guarantee and causes evaluators sharing a threshold to diverge silently when the operating point is tuned.
3. ✗Don't register two evaluators under the same string name in EvalGateRegistry — The registry enforces uniqueness constraints precisely because EvalResult.evaluator_name is the traceability key in production logs and dashboards; a collision overwrites one evaluator's entry with another's, making it impossible to attribute a gate rejection to the correct evaluator family when diagnosing quality regressions.