Hierarchical

HierarchicalRouter

class HierarchicalRouter(BaseRouter):
    def __init__(self, config: Optional[RouterConfig] = None): ...

The public entry point for the entire system. Owns and wires together every other component: the embedding provider, the hybrid scorer, the routing pipeline, the cache, the execution engine, the feedback engine, and observability.

Overview

HierarchicalRouter exists to give applications a single object to construct, register routes on, and query — while internally coordinating caching, two-level scoring, confidence evaluation, execution (with retry/parallel/streaming support), adaptive feedback, and structured logging/metrics. See Architecture for the full request lifecycle diagram.

Constructor

Parameters:

Side effects (in order):

1. Constructs RouterLogger and MetricsCollector from config.observability.
2. Constructs EmbeddingProvider(model_name, batch_size) from config.embedding — this may raise ImportError if sentence-transformers is not installed (see Installation).
3. Constructs an LLMScorer only if config.llm_model is truthy; otherwise the hybrid scorer has no LLM signal.
4. Constructs HybridScorer(scoring_cfg, semantic, keyword, llm).
5. Constructs ConfidenceEvaluator and RoutingPipeline.
6. Constructs CacheManager(config.cache).
7. Constructs FeedbackEngine(config.feedback).
8. Initializes empty _groups: Dict[str, RouteGroup] and _route_map: Dict[str, Route].
9. Selects an aggregator: MergeAggregator() if config.execution.aggregation.value == "merge", else FirstWinsAggregator() (this is the only branch point — AggregationStrategy.VOTING also falls into the FirstWinsAggregator branch; see Configuration).
10. Constructs ExecutionEngine(config.execution, route_map=self._route_map, aggregator) — note route_map is passed by reference, so routes registered after ExecutionEngine construction are automatically visible to it.
11. Sets _fallback_handler = None.

Raises: ImportError (indirectly, via EmbeddingProvider) if sentence-transformers is missing.

Registration API

register_group(self, group: RouteGroup) -> HierarchicalRouter

Registers a whole group at once, encoding embeddings for the group's intent examples (if any and not already computed) and for every contained route.

Raises: ValueError — if a group with the same name is already registered ("Group '{name}' already registered. Use update_group() to replace it.").

Returns: self (fluent chaining).

Side effects: Populates self._groups[group.name], adds every route to self._route_map, triggers embedding encoding via EmbeddingProvider. Does not clear the decision cache (unlike update_group, register, unregister, unregister_group) — since this is a brand-new group, there's nothing stale to invalidate.

update_group(self, group: RouteGroup) -> HierarchicalRouter

Replaces an existing group (removing its old routes from _route_map first), or adds it if not present. Re-encodes embeddings and clears the entire decision cache (self._cache.clear()), since previously cached decisions may reference routes that no longer exist.

Returns: self.

register(self, route: Route, group_name: Optional[str] = None) -> None

Registers a single route. If group_name is omitted, auto-creates (or reuses) a group literally named "default" with description "Default route group".

Side effects: Adds the route to the target group and _route_map; encodes embeddings if needed; clears the decision cache.

Raises: Whatever RouteGroup.add_route raises — ValueError if a route with the same name already exists in the target group.

unregister(self, route_name: str) -> bool

Removes a route by name from both _route_map and its parent group. Clears the decision cache. Returns True if a route was found and removed, False otherwise.

unregister_group(self, group_name: str) -> bool

Removes an entire group and all its routes from _route_map. Clears the decision cache. Returns True/False accordingly.

set_fallback(self, handler: Union[BaseHandler, Callable]) -> None

Sets the global fallback handler invoked by _handle_no_match when no route matches with sufficient confidence. Plain callables are wrapped in CallableHandler.

Primary routing API

route_query(self, query: str, context: Optional[Dict[str, Any]] = None, return_result: bool = True, **handler_kwargs) -> RoutingResult

The main synchronous entry point.

Parameters:

Returns: RoutingResult.

Raises: ValueError — if query is empty or whitespace-only ("Query cannot be empty."). LookupError — if no route matches, no fallback handler is set, and config.raise_on_no_match is True.

Internal workflow:

1. Validate query is non-empty.
2. Build a HandlerRequest with a fresh request_id and merged context.
3. Cache check — CacheManager.get_decision(query). On hit, if the cached route still exists and is enabled, execute it directly (skipping embedding + pipeline entirely), build a minimal RoutingTrace with used_cache=True, and run post-route hooks.
4. Pipeline run — RoutingPipeline.run(query, groups, request_id) → ranked candidates + RoutingTrace.
5. Confidence gate — if there are no candidates, or ConfidenceEvaluator.should_route(best.confidence) is False, delegate to _handle_no_match.
6. Cache the decision — CacheManager.set_decision(...) for the winning route/group.
7. Execute — ExecutionEngine.execute(request, candidates, trace).
8. Post-route hooks — _post_route(...): records metrics, logs the decision, and (if feedback.enabled) auto-records feedback using result.success as the outcome and the winning candidate's effective_score as the feedback score.
9. Return the RoutingResult.

Side effects: Cache reads/writes, metrics recording, structured logging, and (if feedback is enabled) an automatic feedback record + bias application on every call — this happens in addition to any manual router.feedback(...) calls you make yourself.

Complexity: Dominated by embedding encoding (O(model inference)) and scoring (O(number of candidate routes) per group, with SemanticScorer comparing against each route's example embeddings).

Thread safety: Not explicitly documented or synchronized in the analyzed source — CacheManager internally guards its cache with a lock (see CacheManager), but _route_map/_groups mutation (register/unregister) is not lock-protected. Treat concurrent registration and querying from multiple threads as unsafe unless you add your own synchronization.

Examples: See Quickstart for basic usage, Concepts for scoring/confidence details.

route(self, request: HandlerRequest) -> RoutingResult

BaseRouter interface implementation — thin wrapper: return self.route_query(request.query, context=request.context). Note this discards any request_id already set on request (a new one is generated inside route_query), and discards request.metadata.

route_async(self, request: HandlerRequest) -> RoutingResult (async)

Async counterpart to route_query, with the same cache → pipeline → confidence-gate → execute → post-route flow, using await route.execute_async(...) and await self._executor.execute_async(...). Unlike route_query, this method takes a HandlerRequest directly (not a raw query string) and reuses request.request_id if already set, rather than always generating a new one.

[!WARNING] route_async does not call _post_route on the cache-hit path Reading the source: the cache-hit branch inside route_async returns directly (return RoutingResult(...)) without calling self._post_route(...), whereas route_query's cache-hit branch does call _post_route. This means cache hits via route_async do not trigger automatic feedback recording, metrics logging, or the decision-logging call that route_query's cache hits do — an asymmetry between the sync and async paths as observed in the analyzed source.

stream(self, query: str, context: Optional[Dict[str, Any]] = None) -> Iterator[str]

Runs the pipeline (bypassing the decision cache entirely — this method does not consult CacheManager) to get ranked candidates, then delegates token yielding to ExecutionEngine.stream(request, candidates).

stream_async(self, query: str, context: Optional[Dict[str, Any]] = None) -> AsyncIterator[str]

Async counterpart to stream, delegating to ExecutionEngine.stream_async.

chain(self, query: str, route_names: List[str], context: Optional[Dict[str, Any]] = None) -> HandlerResponse

Builds a HandlerRequest and delegates directly to ExecutionEngine.chain(request, route_names) — bypasses routing/scoring entirely, since the caller explicitly names the routes to execute in sequence. See Advanced Usage → Tool chaining.

Feedback

feedback(self, route_name: str, success: bool, score: float = 0.0, query: str = "") -> None

Manually records a feedback signal, looking up group_name from _route_map if the route is known. Calls FeedbackEngine.record(...) then immediately FeedbackEngine.apply_biases(self._route_map), so the bias takes effect on the very next scoring pass.

Re-encoding

re_encode_route(self, route_name: str) -> bool

Looks up the route, calls EmbeddingProvider.encode_route(route), and clears the entire decision cache. Returns False if the route name is unknown.

re_encode_group(self, group_name: str) -> bool

Re-encodes a group's intent examples (if any) and every contained route's examples, then clears the decision cache. Returns False if the group name is unknown.

Observability & inspection

Context manager protocol

def __enter__(self) -> HierarchicalRouter: return self
def __exit__(self, *_):
    if self._cfg.feedback.persist_path:
        self._feedback.save()

Using HierarchicalRouter as a context manager auto-saves learned feedback biases to config.feedback.persist_path on exit (if set) — no explicit save() call needed:

with HierarchicalRouter(config=RouterConfig(feedback=FeedbackConfig(persist_path="fb.json"))) as router:
    ...  # register routes, route queries
# fb.json written here automatically

__call__(self, query: str, **kwargs) -> RoutingResult

Shorthand: router("some query") is equivalent to router.route_query("some query").

Private helpers (documented for behavioral clarity)

_handle_no_match(self, request, trace, start_time) -> RoutingResult

Invoked whenever no candidate clears the confidence threshold. Priority order: (1) if a fallback handler is set, invoke it and return matched=False with its response; (2) else if raise_on_no_match is True, raise LookupError; (3) else return a failed RoutingResult via HandlerResponse.fail(...). The fallback handler takes priority over raise_on_no_match — setting both means the fallback always wins.

_post_route(self, result, request_id, route_name, group_name)

Stamps result.trace.selected_route/selected_group, records metrics, logs the decision, and — if config.feedback.enabled — auto-records a feedback sample using result.success and immediately applies updated biases.

Complete example: production setup with fallback, feedback persistence, and inspection

from zerda.router import (
    HierarchicalRouter, RouterConfig, FeedbackConfig,
    make_rag_group, make_tools_group, make_chat_group, Route,
    BaseHandler, HandlerResponse,
)

class ClarifyHandler(BaseHandler):
    def handle(self, request):
        return HandlerResponse.ok("Could you rephrase that?")

config = RouterConfig(feedback=FeedbackConfig(persist_path="router_feedback.json"))

with HierarchicalRouter(config=config) as router:
    router.set_fallback(ClarifyHandler())

    rag = make_rag_group()
    rag.add_route(Route(name="docs_qa", description="Answer from docs", handler=DocsQAHandler(), examples=[...]))
    router.register_group(rag)

    result = router.route_query("What's in the refund policy?")
    if result:
        print(result.content)

    router.feedback(route_name="docs_qa", success=True, score=0.9)
    print(router.summary())

Related APIs: RoutingPipeline, ExecutionEngine, CacheManager, FeedbackEngine