Advanced Usage

Async routing

Every synchronous entry point has an async counterpart. route_async mirrors route_query's cache-check → pipeline → execute → post-route flow, but awaits route.execute_async and ExecutionEngine.execute_async:

import asyncio

async def main():
    result = await router.route_async_query_equivalent  # see note below

[!NOTE] Method name for async routing The async entry point on HierarchicalRouter is route_async(request: HandlerRequest) — it takes a HandlerRequest, not a raw query string (unlike route_query). Build the request yourself:

from zerda.router import HandlerRequest

async def main():
    request = HandlerRequest(query="What's the weather in Cairo?")
    result = await router.route_async(request)
    print(result.content)

asyncio.run(main())

Streaming

Token streaming from the router directly

HierarchicalRouter.stream / stream_async run the pipeline to select the best candidate, then delegate to the winning handler's stream / stream_async method:

for chunk in router.stream("Tell me about RAG"):
    print(chunk, end="", flush=True)

async for chunk in router.stream_async("Tell me about RAG"):
    print(chunk, end="", flush=True)

By default, BaseHandler.stream() just yields the entire handle() result as one chunk — override stream() / stream_async() on your handler for true token-level streaming.

Streaming with routing events

For UIs that want to show which route was selected before tokens start arriving, use RouterStreamExecutor from router.execution.streaming_executor:

from zerda.router import RouterStreamExecutor

executor = RouterStreamExecutor(router)

async for event in executor.astream_events("Explain RAG"):
    if event.is_token:
        print(event.data, end="", flush=True)
    elif event.type == "route_selected":
        print(f"\n[routed to: {event.data['route']}]")

[!WARNING] Router compatibility RouterStreamExecutor._get_handler expects router.route(query) to return either a dict with a "handler" key, an object with a .handler attribute, or an object with a .best attribute exposing .handler. HierarchicalRouter.route() returns a RoutingResult, which does not expose .handler or .best in the analyzed source of core/result. Verify compatibility before relying on RouterStreamExecutor directly against HierarchicalRouter.route() — it appears designed to sit in front of a broader family of "Router-like" objects (see its own docstring, which references a generic Router type), not necessarily HierarchicalRouter specifically.

Wrapping an arbitrary RAG system

StreamingRouterAdapter normalizes any object exposing astream, agenerate, stream, or generate into a uniform async token generator, trying them in that priority order:

from zerda.router import StreamingRouterAdapter

adapter = StreamingRouterAdapter(my_rag_system, route_name="docs_qa")
async for token in adapter.astream("What is attention?"):
    print(token, end="")

Tool chaining

ExecutionEngine.chain (exposed on HierarchicalRouter.chain) runs a fixed sequence of routes, passing each step's response content into the next step's context under step_<name> and last_result:

response = router.chain(
    query="Summarize the Q3 report and email it to finance",
    route_names=["extract_intent", "retrieve_docs", "generate_answer", "send_email"],
)

If any step's handler returns success=False, the chain stops immediately and returns a failed HandlerResponse describing where it broke.

Tuning the feedback loop

from zerda.router import FeedbackConfig

config = RouterConfig(
    feedback=FeedbackConfig(
        min_samples_to_adapt=25,   # require more data before trusting a route's bias
        persist_path="feedback_state.json",
    ),
)
router = HierarchicalRouter(config=config)

# ... after each routed interaction, once ground truth is known ...
router.feedback(route_name="docs_qa", success=True, score=0.91)

# Persist learned biases on shutdown (also happens automatically via __exit__
# if persist_path is set and you use the router as a context manager):
with HierarchicalRouter(config=config) as r:
    ...  # r._feedback.save() is called automatically on exit

Inspect the current learned state:

print(router.feedback_summary())

Re-encoding after route changes

If you add examples to a route after it's already registered, its cached embeddings become stale — Route.add_example already invalidates route.embeddings, but you still need to force re-encoding through the router (which also clears the decision cache, since a cached routing decision may no longer be valid):

router.get_route("docs_qa").add_example("What's covered under warranty?")
router.re_encode_route("docs_qa")

Or re-encode an entire group at once:

router.re_encode_group("rag")

Custom fallback handling

from zerda.router import BaseHandler, HandlerResponse

class ClarifyHandler(BaseHandler):
    def handle(self, request):
        return HandlerResponse.ok(
            "I'm not sure what you mean — could you rephrase that?"
        )

router.set_fallback(ClarifyHandler())

With a fallback handler set, _handle_no_match invokes it instead of returning a failed response or raising LookupError (even if raise_on_no_match=True — the fallback-handler branch is checked first in the analyzed source).

Inspecting router state

router.summary()        # groups, route counts, effective config, current metrics
router.list_groups()    # ["rag", "tools", "chat"]
router.list_routes()    # all registered route names
router.cache_stats()    # hit/miss counts, hit rate, size
router.metrics()        # rolling-window latency/match/confidence stats