Streaming Executor

Bridges arbitrary streaming RAG systems / handlers with a router pipeline. Provides StreamingRouterAdapter (normalizes any streaming-capable object into a uniform async token generator) and RouterStreamExecutor (a drop-in streaming extension for "Router-like" objects, with an event-based streaming API on top of plain token streaming).

[!NOTE] Designed for a broader family of routers, not exclusively HierarchicalRouter This module's docstrings and duck-typed _get_handler logic (checking for .route() or .process(), then unwrapping .handler / .best.handler / dict "handler" keys) suggest it was designed to sit in front of a general "Router-like" interface, not specifically HierarchicalRouter. See the compatibility warning under RouterStreamExecutor._get_handler below before wiring this directly to HierarchicalRouter.route().

StreamEventType

class StreamEventType:
    ROUTE_SELECTED = "route_selected"
    RETRIEVAL_DONE = "retrieval_done"
    TOKEN          = "token"
    DONE           = "done"
    ERROR          = "error"

Plain string constants (not an Enum) used as RouterStreamEvent.type values.

[!NOTE] RETRIEVAL_DONE is declared but never emitted StreamEventType.RETRIEVAL_DONE is defined but not observed to be used anywhere else in the analyzed source — RouterStreamExecutor only ever yields ROUTE_SELECTED, TOKEN, DONE, and ERROR events. It appears reserved for future use or for RAG systems that report retrieval progress through some other integration point not present in this module.

RouterStreamEvent

@dataclass
class RouterStreamEvent:
    type:    str
    data:    Any
    latency: float = 0.0

Typed event yielded by RouterStreamExecutor.astream_events / stream_events.

Fields

Properties

is_token -> bool

Returns self.type == StreamEventType.TOKEN.

Magic methods

__str__(self) -> str

Returns self.data if is_token, else "" — lets you do str(event) to get just the text content, silently discarding metadata events. Note this only works cleanly for TOKEN events; for non-token events it returns an empty string rather than a string representation of the metadata.

Related APIs: RouterStreamExecutor.astream_events

StreamingRouterAdapter

class StreamingRouterAdapter:
    def __init__(self, rag_system: Any, route_name: str = "default", **llm_kwargs: Any): ...

Wraps an arbitrary RAG system or handler object so its output can be consumed as a uniform async generator of string tokens, regardless of which streaming/generation method the underlying object actually exposes.

Constructor

Methods

astream(self, query: str, **kw: Any) -> AsyncIterator[str] (async generator)

Tries the wrapped object's methods in this priority order, using the first one that exists and produces output:

1. astream(query, **merged) — if it returns a native async generator, yield each token via str(token). If it returns a coroutine instead, await it; if the awaited result is itself an async generator, iterate that; otherwise yield the awaited result as a single str(...) chunk.
2. agenerate(query, **merged) — awaited to get a full string, then split on whitespace and yielded word-by-word (each with a trailing space appended), with await asyncio.sleep(0) between words to yield control back to the event loop. This simulates streaming over a non-streaming async method.
3. stream(query, **merged) — a synchronous generator, iterated directly (blocking the event loop per-iteration, mitigated only by an asyncio.sleep(0) yield-point after each token — this is not offloaded to a thread, unlike the generate fallback below).
4. generate(query, **merged) — a synchronous, non-streaming method, run via await asyncio.to_thread(rag.generate, query, **merged) (so it doesn't block the event loop), then split into words and yielded the same way as agenerate.

Raises: RuntimeError — if none of the four methods exist on rag_system (f"StreamingRouterAdapter: RAG system '{type(rag).__name__}' has no stream/generate method.").

[!WARNING] Priority-3 stream() is not thread-offloaded Unlike the priority-4 generate() fallback (which uses asyncio.to_thread), the priority-3 synchronous stream() path iterates the generator directly on the event loop thread, with only a bare asyncio.sleep(0) between tokens. If the wrapped object's stream() implementation does blocking I/O per-token (e.g. a blocking HTTP streaming client), this can stall the event loop despite the cooperative yield points.

Word-splitting fallback loses original whitespace/punctuation spacing: the agenerate/generate fallback paths reconstruct "streaming" by str(text).split() + " " — multi-space runs, newlines, and other whitespace in the original text are collapsed to single spaces in the yielded output.

Examples

1. Basic — wrapping a system with native async streaming:

class MyRAG:
    async def astream(self, query, **kw):
        for word in ["Hello", "from", "RAG"]:
            yield word + " "

adapter = StreamingRouterAdapter(MyRAG())
async for token in adapter.astream("hi"):
    print(token, end="")

2. Wrapping a synchronous, non-streaming system:

class SimpleRAG:
    def generate(self, query, **kw):
        return "The answer is 42."

adapter = StreamingRouterAdapter(SimpleRAG())
async for token in adapter.astream("what is the answer?"):
    print(token, end="")  # "The " "answer " "is " "42. "

3. Error handling — unsupported object:

adapter = StreamingRouterAdapter(object())
try:
    async for _ in adapter.astream("test"):
        pass
except RuntimeError as e:
    print(e)

Related APIs: RouterStreamExecutor

RouterStreamExecutor

class RouterStreamExecutor:
    def __init__(self, router: Any, fallback_rag: Optional[Any] = None, default_top_k: int = 1): ...

Drop-in streaming extension for any "Router-like" object — calls the router to select a handler, then streams tokens from it, optionally emitting structured progress events along the way.

Constructor

Internal handler resolution

_get_handler(self, query: str) -> Optional[Any]

Algorithm:

1. If self.router has a route attribute, call self.router.route(query); elif it has process, call self.router.process(query); else return self.fallback_rag directly.
2. If the result is None, return self.fallback_rag.
3. If the result is a dict, return result.get("handler") or self.fallback_rag.
4. Elif the result has a .handler attribute, return result.handler.
5. Elif the result has a .best attribute, return getattr(result.best, "handler", self.fallback_rag).
6. Else return self.fallback_rag.
7. Any exception during the above is caught, logged as a warning ("RouterStreamExecutor: routing failed (%s) — using fallback."), and self.fallback_rag is returned.

[!WARNING] HierarchicalRouter.route() does not expose .handler or .best HierarchicalRouter.route(request) returns a RoutingResult, whose attributes are response, trace, matched (plus properties content, success, route_name, group_name, confidence, total_time_ms) — it has neither a .handler attribute nor a .best attribute in the analyzed source of core/result.py. Additionally, HierarchicalRouter.route(request) expects a HandlerRequest, not a raw query string, while _get_handler calls self.router.route(query) with a raw string. Based on static analysis alone, wiring RouterStreamExecutor directly to a HierarchicalRouter instance would call route(query) with the wrong argument type and, even if that were fixed, would not find a .handler/.best attribute on the returned RoutingResult — it would silently fall through to self.fallback_rag (or None if no fallback was set) on every call, calling _get_handler's except Exception branch to catch the TypeError from the argument mismatch. If you want to drive RouterStreamExecutor from a HierarchicalRouter, consider wrapping it in a thin adapter object exposing a route(query) -> object with .handler interface, or pass a pre-built StreamingRouterAdapter as fallback_rag and rely on that instead. This is stated as an observation from static analysis, not a claim about the module authors' intent.

Async streaming methods

astream(self, query: str, **kw: Any) -> AsyncIterator[str]

Resolves a handler via _get_handler. If None, logs a warning and yields the single string "⚠️ No handler available for this query." then returns. Otherwise delegates token-by-token to _astream_from_handler.

_astream_from_handler(self, handler, query, **kw) -> AsyncIterator[str] (internal)

If handler is already a StreamingRouterAdapter, stream from it directly. Otherwise, wrap it in a new StreamingRouterAdapter(handler) on the fly and stream from that — meaning any object satisfying StreamingRouterAdapter's duck-typed method requirements (astream/agenerate/stream/generate) can be returned by your router without pre-wrapping.

astream_events(self, query: str, **kw: Any) -> AsyncIterator[RouterStreamEvent]

Richer streaming that emits a full event sequence:

1. Resolve the handler; immediately yield a ROUTE_SELECTED event with data={"route": route_name, "query_preview": query[:60]}, where route_name is getattr(handler, "route_name", type(handler).__name__ if handler else "none").
2. If handler is None, yield an ERROR event ("No handler found") and return — no further events.
3. Otherwise, stream tokens via _astream_from_handler, yielding one TOKEN event per token. If streaming raises, log it, yield one ERROR event with the exception message, and return (no DONE event follows an error).
4. On successful completion of the token stream, yield a final DONE event with data={"route": route_name, "total_latency": ...}.

Every event's latency field is round(time.perf_counter() - t0, 4) measured from the start of the call.

Synchronous convenience methods

stream(self, query: str, **kw: Any) -> Iterator[str]

Runs astream to completion via asyncio.run(...), collecting all tokens into a list first, then yields them from that list — this is not truly incremental for the caller; all tokens are generated before the first one is yielded back to synchronous code.

[!WARNING] asyncio.run cannot be called from inside a running event loop Per the class docstring: calling stream() (or stream_events()/generate()) from code already running inside an active asyncio event loop will raise a RuntimeError (asyncio.run() cannot be called from a running event loop). Use astream/astream_events/agenerate directly in async contexts instead.

stream_events(self, query: str, **kw: Any) -> Iterator[RouterStreamEvent]

Synchronous variant of astream_events, with the same "collect everything via asyncio.run, then yield from a list" pattern and the same running-event-loop caveat.

Convenience: full-answer generation

agenerate(self, query: str, **kw: Any) -> str (async)

Streams via astream and concatenates all yielded tokens ("".join(parts)) into a single string.

generate(self, query: str, **kw: Any) -> str

Synchronous variant: asyncio.run(self.agenerate(query, **kw)).

Examples

1. Basic — using RouterStreamExecutor with a pre-built StreamingRouterAdapter as the router surface:

from zerda.router.execution.streaming_executor import RouterStreamExecutor, StreamingRouterAdapter

class SimpleRouter:
    """Minimal object satisfying RouterStreamExecutor's duck-typed interface."""
    def __init__(self, adapter):
        self._adapter = adapter
    def route(self, query):
        class _Result:
            handler = self._adapter
        return _Result()

adapter = StreamingRouterAdapter(my_rag_system)
executor = RouterStreamExecutor(SimpleRouter(adapter))

for token in executor.stream("Explain RAG"):
    print(token, end="", flush=True)

2. Async token streaming:

async for token in executor.astream("Explain RAG"):
    print(token, end="", flush=True)

3. Event-based streaming for a UI:

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

4. Error handling — no handler resolvable:

executor = RouterStreamExecutor(router=object())  # no route()/process()
for token in executor.stream("anything"):
    print(token)  # "⚠️ No handler available for this query."

5. Full-answer, non-streaming consumption:

answer = executor.generate("Summarize the document")
print(answer)

Best practices

• Confirm your router object's route()/process() return shape matches one of _get_handler's unwrapping cases (dict["handler"], .handler, .best.handler) before relying on RouterStreamExecutor — see the HierarchicalRouter compatibility warning above.
• Prefer astream_events over astream whenever the UI needs to show routing state (which route was picked) before or during token output.
• Always use the astream*/agenerate async methods when already inside an event loop (e.g. inside a FastAPI async endpoint); reserve stream/stream_events/generate for purely synchronous call sites.

Related APIs: StreamingRouterAdapter, HierarchicalRouter.stream/stream_async