Tutorial: Streaming Responses

This tutorial covers streaming tokens from the winning handler, using HierarchicalRouter.stream/stream_async directly.

1. Implement a handler with real streaming

By default, BaseHandler.stream() just yields the entire handle() result as one chunk. Override it for genuine token-level streaming:

import time
from zerda.router import BaseHandler, HandlerRequest, HandlerResponse

class StoryHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        return HandlerResponse.ok("Once upon a time, in a small village, there lived a curious fox.")

    def stream(self, request: HandlerRequest):
        text = "Once upon a time, in a small village, there lived a curious fox."
        for word in text.split():
            yield word + " "
            time.sleep(0.05)  # simulate token generation latency

    async def stream_async(self, request: HandlerRequest):
        import asyncio
        text = "Once upon a time, in a small village, there lived a curious fox."
        for word in text.split():
            yield word + " "
            await asyncio.sleep(0.05)

2. Register and stream synchronously

from zerda.router import HierarchicalRouter, RouteGroup, Route

router = HierarchicalRouter()
group = RouteGroup("chat", "General conversation")
group.add_route(Route(
    name="story",
    description="Tell a short story",
    handler=StoryHandler(),
    examples=["tell me a story", "once upon a time"],
))
router.register_group(group)

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

3. Stream asynchronously

import asyncio

async def main():
    async for chunk in router.stream_async("Tell me a story"):
        print(chunk, end="", flush=True)

asyncio.run(main())

[!NOTE] stream()/stream_async() bypass the decision cache HierarchicalRouter.stream/stream_async re-run the routing pipeline on every call — unlike route_query, they do not consult CacheManager. This is a deliberate simplification in the analyzed source; if you need cached routing decisions for streaming calls, route first via route_query/route_async to get the route name, then invoke that handler's stream/stream_async directly yourself.

4. Event-driven streaming for UIs

For a UI that needs to show which route was selected before tokens arrive, wrap things with RouterStreamExecutor — see Advanced Usage → Streaming with routing events for the full pattern and its HierarchicalRouter compatibility caveats.

Next steps

• Tutorial: Closing the Feedback Loop
• Advanced Usage for tool chaining