Route

Defines Route and its keyword-signal configuration RouteKeywords.

RouteKeywords

@dataclass
class RouteKeywords:
    required:   List[str] = field(default_factory=list)
    any_of:     List[str] = field(default_factory=list)
    excluded:   List[str] = field(default_factory=list)
    boost:      float     = 0.05

Keyword signals that improve routing accuracy without an embedding call. Consumed by KeywordScorer.

Fields

Example

from zerda.router import RouteKeywords

kw = RouteKeywords(
    any_of=["weather", "temperature", "forecast"],
    excluded=["history", "historical"],
    boost=0.15,
)

Related APIs: Route, KeywordScorer

Route

class Route:
    def __init__(
        self,
        name: str,
        description: str,
        handler: Union[BaseHandler, Callable],
        examples: Optional[List[str]] = None,
        keywords: Optional[RouteKeywords] = None,
        group: Optional[str] = None,
        tags: Optional[Set[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        enabled: bool = True,
        priority: int = 0,
        tools: Optional[List[str]] = None,
        score_bias: float = 0.0,
    ): ...

A named routing destination with semantic examples, keyword signals, a handler, and runtime metrics.

Overview

Route is the atomic unit the scorer and executor operate on. It accepts either a BaseHandler instance or a plain callable (transparently wrapped via CallableHandler). Beyond the handler, a Route tracks example utterances used for semantic scoring, optional keyword rules, group membership, tags, arbitrary metadata, an enabled/disabled flag, a priority tie-breaker, tool-chaining declarations, and a persistent score bias set by the feedback engine.

Constructor

Parameters:

Raises:

• ValueError — if name is empty or whitespace-only ("Route name cannot be empty.").
• ValueError — if description is empty or whitespace-only ("Route description cannot be empty.").

Side effects: Sets created_at and updated_at to time.time(). Initializes embeddings = [] (populated later by an EmbeddingProvider).

Attributes

Methods

execute(self, request: HandlerRequest, similarity: float = 0.0) -> HandlerResponse

Runs the handler synchronously: checks enabled, calls handler.validate(request), then handler.handle(request), timing the call with time.perf_counter().

Parameters:

Returns: HandlerResponse — the handler's response, with processing_time back-filled if the handler left it at 0.0.

Raises: RuntimeError — if self.enabled is False (f"Route '{self.name}' is disabled."). This propagates to the caller; ExecutionEngine does not appear to catch it in the analyzed source, so disabled routes reaching execute() directly (e.g. via cache hits referencing a since-disabled route — though route_query's cache-hit path explicitly checks route.enabled first) would raise.

Side effects: None beyond timing; does not mutate self.

Exception handling: Any exception raised by handler.validate is not caught here (only handler.handle is wrapped in try/except) — a raising validate() implementation will propagate out of execute(). Exceptions from handler.handle() are caught and converted to HandlerResponse.fail(f"Handler raised {type(exc).__name__}: {exc}").

Complexity: O(handler execution time); no additional algorithmic overhead.

Thread safety: Depends entirely on the wrapped handler's handle() implementation — Route.execute itself has no shared mutable state beyond timestamps it doesn't touch here.

execute_async(self, request: HandlerRequest, similarity: float = 0.0) -> HandlerResponse (async)

Async counterpart to execute — identical control flow, but calls await handler.handle_async(request) instead of handler.handle(request). Same enabled-check and same exception-wrapping behavior.

add_example(self, example: str) -> bool

Adds a new example utterance if non-empty and not already present. Strips whitespace before comparing/storing.

Returns: True if the example was added, False if it was empty or already present.

Side effects: Appends to self.examples; sets self.embeddings = [] (invalidating cached embeddings — you must re-encode); updates self.updated_at.

remove_example(self, example: str) -> bool

Removes an example if present.

Returns: True if removed, False if not found.

Side effects: Same invalidation as add_example — self.embeddings = [], self.updated_at refreshed.

invalidate_embeddings(self)

Explicitly clears self.embeddings and refreshes updated_at, without touching examples. Useful if you've changed the embedding model or want to force re-encoding without editing examples.

enable(self) / disable(self)

Toggle self.enabled and refresh self.updated_at. Disabling a route excludes it from RouteGroup.get_routes(enabled_only=True) (the default) and from RoutingPipeline's candidate collection.

apply_score_bias(self, delta: float)

Adds delta to self.score_bias, clamped to [-0.3, 0.3]. Called by FeedbackEngine.apply_biases (which actually sets score_bias to the computed bias directly rather than calling this method — see FeedbackEngine for the exact mechanism). Refreshes updated_at.

to_dict(self) -> Dict[str, Any]

Serializes the route's configuration (not its handler, embeddings, or timestamps) to a plain dict: name, description, examples, keywords (as a nested dict), group, tags (as a list), metadata, enabled, priority, tools, score_bias.

Route.from_dict(data: Dict[str, Any], handler: Union[BaseHandler, Callable]) -> Route (classmethod)

Reconstructs a Route from a dict produced by to_dict() (or a hand-built equivalent), given a handler instance to attach (handlers are never serialized). Missing optional keys fall back to the same defaults as the constructor.

Example — round-tripping a route's configuration:

data = route.to_dict()
# ... persist `data` as JSON, handler is reconstructed separately ...
restored = Route.from_dict(data, handler=DocsQAHandler())

Magic methods

• __eq__ — two Route instances are equal iff self.name == other.name (handler, examples, etc. are not compared).
• __hash__ — hash(self.name), consistent with __eq__.
• __repr__ — e.g. Route(name='docs_qa', group='rag', examples=3, priority=0, status=enabled, bias=+0.120).

Examples

1. Basic — callable handler:

from zerda.router import Route

route = Route(
    name="echo",
    description="Echo back the query",
    handler=lambda q: q,
    examples=["repeat this", "echo test"],
)

2. Intermediate — with keywords and priority:

from zerda.router import Route, RouteKeywords

route = Route(
    name="urgent_escalation",
    description="Escalate urgent customer complaints to a human agent",
    handler=EscalationHandler(),
    examples=["I want to speak to a manager", "this is unacceptable"],
    keywords=RouteKeywords(any_of=["urgent", "manager", "complaint"], boost=0.2),
    priority=100,  # outrank normal support routes on ties
)

3. Advanced — dynamic example management:

route = Route(name="faq", description="Answer FAQs", handler=FaqHandler())
for q in load_faq_questions():
    route.add_example(q)

# After bulk-adding examples, force the router to re-encode:
router.register(route, group_name="faq")
router.re_encode_route("faq")

4. Error handling — disabled route:

route.disable()
try:
    route.execute(HandlerRequest(query="test"))
except RuntimeError as e:
    print(e)  # "Route 'faq' is disabled."

5. Production-ready — serializable route registry:

import json

routes_config = [r.to_dict() for r in group.get_routes(enabled_only=False)]
with open("routes.json", "w") as f:
    json.dump(routes_config, f, indent=2)

# On restart, reconstruct with the appropriate handler for each name:
handler_registry = {"docs_qa": DocsQAHandler(), "weather": WeatherHandler()}
with open("routes.json") as f:
    for data in json.load(f):
        restored = Route.from_dict(data, handler=handler_registry[data["name"]])
        group.add_route(restored)

Best practices

• Call router.re_encode_route(name) (not just Route.add_example) after adding examples to an already-registered route — add_example only invalidates the local cache, it doesn't trigger re-encoding or clear the router's decision cache.
• Use priority sparingly and only as a genuine tie-breaker; it does not influence the hybrid score itself, only ordering within RouteGroup.get_routes().
• Keep tools accurate if you plan to build tool-chaining logic on top of it — Route itself does not validate or enforce these declarations.

Common mistakes

• Assuming Route.execute's similarity parameter affects handler behavior — it is accepted but not observed to be used inside execute/execute_async in the analyzed source.
• Forgetting that Route.__eq__/__hash__ are name-based — putting two Route objects with the same name but different handlers into a set() will silently collapse them to one.

Related APIs: RouteGroup, BaseHandler, FeedbackEngine