Scorers
On this page
The hybrid scoring layer: three independent signal sources (SemanticScorer, KeywordScorer, LLMScorer) unified by HybridScorer into a single weighted score per route.
BaseScorer
class BaseScorer(ABC):
@abstractmethod
def score(self, query: str, route: Route, **ctx) -> float: ...
@abstractmethod
def batch_score(self, query: str, routes: List[Route], **ctx) -> List[float]: ...
def name(self) -> str: ...
Abstract contract every scorer implements, returning a float in [0.0, 1.0].
Related APIs: SemanticScorer, KeywordScorer, LLMScorer, HybridScorer
SemanticScorer
class SemanticScorer(BaseScorer):
def __init__(self, cfg: ScoringConfig): ...
Scores via similarity between a query embedding and each route's per-example embeddings, using the maximum similarity across all examples — a query only needs to match one example well for the route to score highly.
Similarity metrics
Controlled by cfg.similarity_metric (a SimilarityMetric):
Methods
score(self, query: str, route: Route, query_embedding: Optional[List[float]] = None, **_) -> float
Returns 0.0 immediately if route.embeddings is empty or query_embedding is None. Otherwise computes the similarity between query_embedding and every vector in route.embeddings, returning the maximum.
Complexity: O(n_examples · dim) — a linear scan of the route's example embeddings, each a fixed-cost similarity computation.
batch_score(self, query: str, routes: List[Route], query_embedding: Optional[List[float]] = None, **_) -> List[float]
Vectorized version: for each route, stacks its example embeddings into an (n_examples, dim) NumPy matrix and computes all similarities against the query in one operation via matrix multiplication (cosine/dot-product) or vectorized norm computation (Euclidean), then takes the max. Returns 0.0 for any route with no embeddings, and a list of all 0.0s if query_embedding is None.
Complexity: O(n_examples · dim) per route via vectorized NumPy ops — asymptotically the same as score() called in a loop, but with substantially lower constant-factor overhead due to BLAS-backed matrix operations instead of a Python-level loop.
Thread safety: Pure function of its inputs — no shared mutable state, safe to call concurrently.
Examples
from zerda.router.routing.scorers import SemanticScorer
from zerda.router import ScoringConfig, SimilarityMetric
scorer = SemanticScorer(ScoringConfig(similarity_metric=SimilarityMetric.COSINE))
score = scorer.score("what's the weather", route, query_embedding=q_emb)
scores = scorer.batch_score("what's the weather", [route1, route2, route3], query_embedding=q_emb)
Related APIs: EmbeddingProvider, Route.embeddings
KeywordScorer
class KeywordScorer(BaseScorer):
def score(self, query: str, route: Route, **_) -> float: ...
Lightweight, model-free keyword/pattern scorer using each route's RouteKeywords. Runs in O(n_keywords) with no model inference — "essentially free" per the source docstring.
Scoring algorithm
- Lowercase the query once.
- Required check: for every word in
keywords.required, if it's not a lowercase substring of the query, return0.0immediately (veto). - Excluded check: for every word in
keywords.excluded, if it is a substring of the query, return0.0immediately (veto). - Bonus accumulation: for every word in
keywords.any_offound in the query, addkeywords.boostto a runningbonus. - Normalization:
- If
keywords.any_ofis empty: return0.0ifrequiredorexcludedwere non-empty (the route has some keyword configuration, it just didn't add positive signal), else return a flat0.1base score (no keyword configuration at all — small positive default rather than 0). - Otherwise:
max_possible = len(any_of) * boost; returnmin(1.0, bonus / max_possible)(or0.0ifmax_possibleis somehow0).
- If
Matching is substring-based, not word-boundary-based — e.g. a keyword "cat" matches inside "category". This is a direct consequence of using Python's in operator on lowercased strings; the implementation does not tokenize or use regex word boundaries (re is imported at the top of the module but not used inside KeywordScorer).
Methods
batch_score(self, query: str, routes: List[Route], **_) -> List[float]
Simply [self.score(query, r) for r in routes] — no vectorization, since keyword matching is already cheap.
Examples
from zerda.router import Route, RouteKeywords
from zerda.router.routing.scorers import KeywordScorer
scorer = KeywordScorer()
route = Route(
name="weather", description="...", handler=h,
keywords=RouteKeywords(any_of=["weather", "forecast"], boost=0.2),
)
scorer.score("what's the weather today", route) # 1.0 (1 match / max_possible 0.2, capped at 1.0)
route2 = Route(
name="urgent", description="...", handler=h,
keywords=RouteKeywords(required=["urgent"], excluded=["test"]),
)
scorer.score("this is urgent", route2) # 0.0 → any_of empty but required present → 0.0 base
scorer.score("this is a test, urgent", route2) # 0.0 (vetoed by "test")
[!WARNING] A route with only
required/excluded(noany_of) never scores above 0 from keywords alone Per the normalization rule, ifany_ofis empty, the method returns exactly0.0wheneverrequired/excludedare configured (even if the query satisfies allrequiredkeywords and avoids allexcludedones) — it never returns a positive score in that branch. To get positive keyword signal,any_ofmust be non-empty.
Related APIs: RouteKeywords
LLMScorer
class LLMScorer(BaseScorer):
def __init__(self, model: str, api_key: Optional[str] = None): ...
Optional, costly scorer that classifies a query against all candidate route descriptions in a single LLM prompt. Invoked by HybridScorer only as a fallback when semantic+keyword signal is weak.
Constructor
Stores model and api_key, then calls self._build_client().
_build_client(self)
Attempts import openai and constructs openai.OpenAI(api_key=self._api_key) (or openai.OpenAI() if no key given). Catches ImportError and returns None — unlike EmbeddingProvider, a missing openai package does not raise; it silently degrades the scorer to a no-op (see below).
[!WARNING] Only ImportError is caught at construction
_build_clientonly guards againstImportError. Ifopenaiis installed but the client construction itself fails for another reason (e.g. malformedapi_keyargument causing some other exception type, misconfigured environment), that exception will propagate out ofLLMScorer.__init__uncaught.
Methods
_classify(self, query: str, routes: List[Route]) -> Dict[str, float] (private, documented for behavior clarity)
Returns {} immediately if self._client is None (missing openai dependency — a silent, cost-free no-op). Otherwise builds a prompt listing every route's name and description, asks the model to rate each from 0.0–1.0 as a JSON object, calls self._client.chat.completions.create(model=self._model, messages=[...], max_tokens=200, response_format={"type": "json_object"}), and parses the response content as JSON.
Exception handling: The entire API call + JSON parse is wrapped in a bare except Exception: return {} — any failure (network error, rate limit, malformed JSON, unexpected response shape) silently degrades to an empty score dict rather than raising or logging. There is no logging call inside this method in the analyzed source, so LLM scorer failures are invisible unless you separately instrument HybridScorer/HierarchicalRouter.
score(self, query: str, route: Route, **ctx) -> float
Reads ctx.get("all_routes", [route]) (defaulting to a single-route list containing just route if not provided), classifies against that list, and returns the score for route.name, defaulting to 0.0 if absent from the response.
batch_score(self, query: str, routes: List[Route], **_) -> List[float]
Classifies once against all routes in a single LLM call, then returns each route's score from the result dict (defaulting to 0.0 per-route if the LLM omitted it).
Cost and invocation behavior
HybridScorer.score_all only invokes LLMScorer.batch_score when the best raw semantic+keyword combined score across all candidates is below cfg.fallback_confidence_threshold — see HybridScorer.score_all below. This makes LLMScorer a genuine fallback: a single classification call per ambiguous query, not a call on every request.
Examples
1. Basic setup:
from zerda.router.routing.scorers import LLMScorer
scorer = LLMScorer(model="gpt-4o-mini", api_key="sk-...")
2. Direct scoring (rarely done outside HybridScorer):
scores = scorer.batch_score("book me a flight", [route_a, route_b, route_c])
3. Missing dependency — silent no-op:
# If `openai` is not installed:
scorer = LLMScorer(model="gpt-4o-mini")
scorer.score("anything", some_route) # 0.0 — no exception, no log line
Best practices
- Add your own logging/metrics around LLM scorer usage if visibility into fallback-classification failures matters to your operations — the implementation is silent by design (fail-open with a zero score) rather than fail-loud.
- Set
RouterConfig.llm_model(andllm_api_keyif not using an environment-basedOPENAI_API_KEY) only if you actually want the LLM fallback tier; otherwise leave itNoneto avoid constructing anopenai.OpenAIclient at all.
Related APIs: HybridScorer, RouterConfig.llm_model
HybridScorer
class HybridScorer:
def __init__(self, cfg: ScoringConfig, semantic: SemanticScorer, keyword: KeywordScorer, llm: Optional[LLMScorer] = None): ...
Combines semantic, keyword, and optional LLM scores using configurable, auto-normalized weights, plus each route's persistent score_bias.
Constructor
Stores the three (or two, if llm is None) scorers, then normalizes weights to sum to 1:
llm_w = cfg.llm_weight if llm else 0.0
total = cfg.semantic_weight + cfg.keyword_weight + llm_w
self._w_sem = cfg.semantic_weight / total
self._w_kw = cfg.keyword_weight / total
self._w_llm = llm_w / total
Raises: ZeroDivisionError — if cfg.semantic_weight + cfg.keyword_weight + llm_w == 0 (e.g. all three weights configured as 0.0). Not caught internally; validate your ScoringConfig weights are not all zero.
[!NOTE] llm_weight is silently zeroed if no LLMScorer is passed Even if
cfg.llm_weightis nonzero, passingllm=Noneto the constructor forcesllm_w = 0.0before normalization — the configured weight has no effect unless an actualLLMScorerinstance is supplied.
Methods
score_all(self, query: str, routes: List[Route], query_embedding: Optional[List[float]] = None) -> List[Tuple[Route, float, float, float]]
Scores all routes and returns (route, semantic_score, keyword_score, effective_score) tuples, sorted by effective_score descending.
Algorithm:
- Batch-score semantic (
self._semantic.batch_score) and keyword (self._keyword.batch_score) signals for all routes. - Compute
raw_combined[i] = w_sem * sem[i] + w_kw * kw[i]for every route (LLM excluded at this stage). best_raw = max(raw_combined)(or0.0ifroutesis empty).- Conditionally invoke the LLM scorer: only if
self._llm is not Noneandbest_raw < cfg.fallback_confidence_threshold. Otherwisellm_scoresstays all-zeros. - For each route:
combined = w_sem*sem + w_kw*kw + w_llm*llm;effective = min(1.0, combined + route.score_bias). - Append
(route, sem, kw, effective)— note the 4th tuple element is the bias-inclusiveeffectivescore, not the rawcombinedscore (the local variable namedcombinedis computed but the tuple storeseffective, per the source'sresults.append((route, sem, kw, effective))). - Sort by the 4th element descending; return.
Returns: List of 4-tuples, sorted best-first. Important: the 4th tuple element is documented in the source's own docstring as "combined" but is actually the score including route.score_bias — see RouteCandidate and its effective_score property for the related double-bias-application caveat.
Complexity: O(n_routes) for semantic/keyword batch scoring, plus one LLM call (if triggered) covering all routes in a single request rather than one call per route, plus an O(n_routes log n_routes) sort.
Side effects: None beyond scorer-internal caching (EmbeddingProvider's cache is not touched here directly; SemanticScorer/KeywordScorer are stateless). An LLMScorer call, when triggered, is a real network request with real cost.
Thread safety: HybridScorer itself holds no mutable state after construction (weights are fixed); safety depends on the injected scorers' thread safety.
Examples
1. Basic:
from zerda.router.routing.scorers import HybridScorer, SemanticScorer, KeywordScorer
from zerda.router import ScoringConfig
cfg = ScoringConfig()
scorer = HybridScorer(cfg, SemanticScorer(cfg), KeywordScorer(), llm=None)
ranked = scorer.score_all("what's the weather", [route1, route2], query_embedding=q_emb)
best_route, sem, kw, score = ranked[0]
2. With LLM fallback enabled:
from zerda.router.routing.scorers import LLMScorer
llm = LLMScorer(model="gpt-4o-mini", api_key="sk-...")
scorer = HybridScorer(cfg, SemanticScorer(cfg), KeywordScorer(), llm=llm)
# LLM is only actually called if semantic+keyword scores are all weak
ranked = scorer.score_all("some ambiguous query", routes, query_embedding=q_emb)
3. Error handling — all weights zero:
bad_cfg = ScoringConfig(semantic_weight=0, keyword_weight=0, llm_weight=0)
try:
HybridScorer(bad_cfg, SemanticScorer(bad_cfg), KeywordScorer(), llm=None)
except ZeroDivisionError:
print("At least one weight must be nonzero")
Best practices
- Don't rely on the 4th element of
score_all's tuples being bias-free — it already includesroute.score_bias. If you need the pure weighted-signal score without bias, recompute it from the 2nd/3rd elements and your own weight values. - Set
ScoringConfigweights as ratios you're comfortable with; they don't need to sum to 1 — normalization happens automatically — but all three (or two, if no LLM scorer) must not sum to exactly zero.
Related APIs: RoutingPipeline.run, RouteCandidate, Architecture → Scoring formula