ZerdaZerda

Pipeline

On this page

Orchestrates the full routing decision: embed → select group → score routes → evaluate confidence → return ranked candidates + a trace. Also provides the EmbeddingProvider wrapper and the ConfidenceEvaluator.

ConfidenceEvaluator

Python · 2 lines
class ConfidenceEvaluator:
    def __init__(self, cfg: ScoringConfig): ...

Translates a raw numeric combined score into a ConfidenceLevel and decides what action the pipeline should take.

Constructor

Reads high_confidence_threshold, medium_confidence_threshold, and fallback_confidence_threshold from the given ScoringConfig and stores them.

Methods

evaluate(self, score: float) -> ConfidenceLevel

Text · 4 lines
score >= high_threshold      → ConfidenceLevel.HIGH
score >= medium_threshold    → ConfidenceLevel.MEDIUM
score >= fallback_threshold  → ConfidenceLevel.LOW
otherwise                    → ConfidenceLevel.NONE

Thresholds are checked in descending order — the first satisfied condition wins.

should_use_llm(self, score: float) -> bool

Returns score < self._fallback (i.e. fallback_confidence_threshold). Used by HybridScorer to decide whether the (optional) LLM scorer is worth invoking.

should_route(self, level: ConfidenceLevel) -> bool

Returns True for HIGH, MEDIUM, or LOW; False only for NONE. This is the single gate HierarchicalRouter.route_query/route_async use to decide between executing a route and calling _handle_no_match.

Examples:

Python · 8 lines
from zerda.router.config import ScoringConfig
from zerda.router.routing.pipeline import ConfidenceEvaluator

evaluator = ConfidenceEvaluator(ScoringConfig())
evaluator.evaluate(0.90)  # ConfidenceLevel.HIGH
evaluator.evaluate(0.50)  # ConfidenceLevel.MEDIUM
evaluator.evaluate(0.35)  # ConfidenceLevel.NONE (below default 0.40 fallback threshold)
evaluator.should_route(ConfidenceLevel.NONE)  # False

Related APIs: ScoringConfig, RoutingPipeline

EmbeddingProvider

Python · 2 lines
class EmbeddingProvider:
    def __init__(self, model_name: str, batch_size: int = 32): ...

Thin wrapper around a sentence_transformers.SentenceTransformer model, adding an in-memory embedding cache keyed by MD5 hash of the input text.

Constructor

Parameters:

NameTypeDefaultDescription
model_namestr—Passed directly to SentenceTransformer(model_name).
batch_sizeint32Used as the batch_size argument in batched SentenceTransformer.encode calls.

Raises: ImportError — if sentence_transformers cannot be imported (message: "sentence-transformers is required. Install with: pip install sentence-transformers"), chained from the original ImportError via raise ... from exc.

Side effects: Loads the sentence-transformer model into memory immediately (constructor-time, not lazy) — this can be a slow, potentially network-dependent operation (model download on first use) and a significant memory allocation.

Methods

encode(self, text: str, use_cache: bool = True) -> List[float]

Encodes a single string into an embedding vector.

Algorithm: Compute hashlib.md5(text.encode()).hexdigest() as the cache key. If use_cache and the key is cached, return the cached vector. Otherwise call self._model.encode(text, convert_to_numpy=True, normalize_embeddings=True).tolist(), cache it (if use_cache), and return it.

Complexity: O(1) on cache hit; O(model forward pass) on cache miss.

Side effects: Mutates self._cache on a cache miss (if use_cache=True).

Thread safety: The internal self._cache dict is a plain Python dict with no locking — concurrent calls from multiple threads with overlapping cache misses on the same key are not guaranteed safe (a classic read-check-write race), though CPython's GIL makes outright corruption unlikely for simple dict assignment; treat as not explicitly thread-safe.

encode_batch(self, texts: List[str], use_cache: bool = True) -> List[List[float]]

Encodes multiple texts efficiently, checking the cache for each and only sending genuinely uncached texts to the model in one batched encode(..., batch_size=self._batch_size) call.

Algorithm: Compute an MD5 key per text. Partition into already-cached (filled in directly) vs. uncached (collected for a single batched model call). Batch-encode only the uncached texts, then scatter results back into the correct output positions and populate the cache.

Complexity: O(1) per cache hit + O(model forward pass over uncached texts) for misses, in one batched call rather than N individual calls — this is the primary reason to prefer encode_batch over looping encode.

encode_route(self, route: Route)

If route.examples is non-empty, sets route.embeddings = self.encode_batch(route.examples). No-op (leaves route.embeddings unchanged) if route.examples is empty.

encode_group(self, group: RouteGroup)

Same pattern for group.intent_examples → group.embeddings.

clear_cache(self)

Empties self._cache entirely (self._cache.clear()). There is no partial/TTL-based eviction — this is a full flush.

cache_size (property)

Returns len(self._cache).

Examples

1. Basic:

Python · 3 lines
from zerda.router.routing.pipeline import EmbeddingProvider
embedder = EmbeddingProvider("sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")
vec = embedder.encode("hello world")

2. Intermediate — batch encoding for a route's examples:

Python · 4 lines
from zerda.router import Route
route = Route(name="faq", description="...", handler=faq_handler, examples=["q1", "q2", "q3"])
embedder.encode_route(route)
print(len(route.embeddings))  # 3

3. Advanced — cache inspection:

Python · 3 lines
embedder.encode("repeated query")
embedder.encode("repeated query")  # served from cache
print(embedder.cache_size)  # 1, not 2

4. Error handling — missing dependency:

Python · 4 lines
try:
    embedder = EmbeddingProvider("some-model")
except ImportError as e:
    print(e)  # "sentence-transformers is required. Install with: pip install sentence-transformers"

5. Production — memory management via explicit cache clearing:

Python · 4 lines
# If your app encodes a huge number of unique one-off queries and memory
# growth matters, periodically clear the unbounded embedding cache:
if embedder.cache_size > 50_000:
    embedder.clear_cache()

Best practices

  • Prefer encode_batch over calling encode in a loop whenever you have several texts up front (e.g. a route's initial example list) — it batches the actual model call.
  • Be aware the embedding cache is unbounded and never expires on its own — for long-running processes with high query cardinality, monitor cache_size and call clear_cache() if memory becomes a concern.

Related APIs: Route.embeddings, RouteGroup.embeddings

RoutingPipeline

Python · 2 lines
class RoutingPipeline:
    def __init__(self, cfg: RouterConfig, embedder: EmbeddingProvider, scorer: HybridScorer, confidence: ConfidenceEvaluator): ...

Two-level (group → route) routing pipeline. Stateless with respect to routes — each call to run() receives a fresh snapshot of active groups; all persistent state (feedback, cache) lives in components owned by HierarchicalRouter.

Methods

run(self, query: str, groups: List[RouteGroup], request_id: str, embedding_cache: Optional[Dict[str, List[float]]] = None) -> Tuple[List[RouteCandidate], RoutingTrace]

Executes the full pipeline for one query.

Parameters:

NameTypeDescription
querystrThe user's query.
groupsList[RouteGroup]All groups to consider (typically list(self._groups.values()) from the router).
request_idstrCorrelation ID, stamped onto the returned trace.
embedding_cacheOptional[Dict[str, List[float]]]Accepted as a parameter but not observed to be read or written anywhere inside run() in the analyzed source — the method relies entirely on self._embedder's own internal cache instead. Passing a value here has no observed effect.

Returns: (candidates, trace) — candidates is a list of RouteCandidate, best-first, truncated to top_k; trace is a fully populated RoutingTrace.

Algorithm (internal workflow):

  1. Create the trace.
  2. Filter groups to active_groups = [g for g in groups if g.enabled]. If empty, note it and return ([], trace) immediately.
  3. Embed the query once via self._embedder.encode(query), timing it into trace.embedding_time_ms.
  4. Call self._select_groups(...) (see below) to narrow down to the best group(s). If none selected, note it and return ([], trace).
  5. Collect candidate_routes = all enabled_only=True routes from every selected group.
  6. If no candidate routes, note it and return ([], trace).
  7. Score all candidates in one call: self._scorer.score_all(query, candidate_routes, q_emb), timing into trace.scoring_time_ms.
  8. Take the top self._cfg.scoring.top_k scored tuples, evaluate confidence per candidate, and build RouteCandidate objects.
  9. Populate trace.candidates, and if there's a best candidate, stamp trace.selected_route/selected_group/confidence; if that candidate doesn't clear should_route, add an explanatory note citing the score and threshold.
  10. Return (candidates, trace).

Complexity: One embedding call (O(model inference) for the query) + one batched scoring call over all candidate routes in the selected group(s) — group and route selection are each single-pass linear scans, not iterative/recursive.

Thread safety: The pipeline object itself holds no per-call mutable state (each call constructs its own local trace), but the injected embedder and scorer have their own thread-safety characteristics (see EmbeddingProvider above).

_select_groups(self, query, query_emb, groups, trace) -> List[RouteGroup] (private, documented for clarity)

Algorithm:

  • If exactly one group is active, return it directly — no scoring needed (documented in source as avoiding "double-encoding").
  • Otherwise, build a synthetic Route proxy per group: name=group.name, description=group.description, a no-op CallableHandler, examples=group.intent_examples, keywords=group.intent_keywords, priority=group.priority, and — critically — proxy.embeddings = group.embeddings (pre-computed group intent embeddings are reused directly rather than re-encoded).
  • Score all proxies via the same HybridScorer.score_all used for routes.
  • Take the single top-scoring proxy (scored[0]) — the implementation currently only ever selects one group, despite the docstring calling this "top group(s)" and leaving a comment noting it can be "extended for multi-group queries."
  • Map the winning proxy's name back to the real RouteGroup object and return it as a single-element list (or [] if the mapping somehow fails).
  • Add a trace note recording which group was selected, its score, and its confidence level.

[!NOTE] Group selection is currently single-winner only Even though multiple groups may score close to the top group, _select_groups returns at most one group. If you need routes from multiple domains to be compared together, put them in the same RouteGroup rather than relying on multi-group selection, since only the single best-scoring group's routes reach step 5 of run().

Examples

1. Basic — direct pipeline invocation (advanced/testing use):

Python · 11 lines
from zerda.router.routing.pipeline import RoutingPipeline, EmbeddingProvider, ConfidenceEvaluator
from zerda.router.routing.scorers import HybridScorer, SemanticScorer, KeywordScorer
from zerda.router import RouterConfig

cfg = RouterConfig()
embedder = EmbeddingProvider(cfg.embedding.model_name)
scorer = HybridScorer(cfg.scoring, SemanticScorer(cfg.scoring), KeywordScorer(), None)
confidence = ConfidenceEvaluator(cfg.scoring)
pipeline = RoutingPipeline(cfg, embedder, scorer, confidence)

candidates, trace = pipeline.run("What's the weather?", groups=[tools_group], request_id="req-1")

2. Inspecting why a group was chosen:

Python
print(trace.notes)  # e.g. ["Group selected: 'tools' (score=0.734, confidence=medium)"]

Related APIs: HybridScorer, RouteCandidate, RoutingTrace, HierarchicalRouter

Related pages