core.result

Typed result objects produced by the routing pipeline: scored candidates, the full decision trace, and the final result objects returned to callers.

RouteCandidate

@dataclass
class RouteCandidate:
    route_name:      str
    group_name:      Optional[str]
    combined_score:  float
    semantic_score:  float
    keyword_score:   float
    llm_score:       float
    confidence:      ConfidenceLevel
    score_bias:      float = 0.0

A scored route candidate surfaced by the RoutingPipeline.

Fields

Properties

effective_score -> float

Returns min(1.0, self.combined_score + self.score_bias).

[!WARNING] Potential double-application of score_bias Since RoutingPipeline.run already stores HybridScorer's bias-inclusive result into combined_score, and score_bias is also stored alongside it, RouteCandidate.effective_score re-adds score_bias on top of an already bias-inclusive combined_score. Based on static analysis alone, this means effective_score may double-count the bias relative to the value HybridScorer originally computed. This is stated as an observation from reading the source, not a claim about intended behavior — verify against your version's actual scoring output if precise bias accounting matters for your use case.

Methods

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

Returns a dict with rounded (4 decimal places) score fields: route, group, score (= effective_score), semantic, keyword, llm, confidence (.value string).

Related APIs: RoutingTrace, HybridScorer

RoutingTrace

@dataclass
class RoutingTrace:
    request_id:        str
    query:             str
    candidates:        List[RouteCandidate]      = field(default_factory=list)
    selected_group:    Optional[str]              = None
    selected_route:    Optional[str]              = None
    confidence:        Optional[ConfidenceLevel]   = None
    used_llm_fallback: bool                        = False
    used_cache:        bool                        = False
    embedding_time_ms: float                       = 0.0
    scoring_time_ms:   float                       = 0.0
    execution_time_ms: float                       = 0.0
    total_time_ms:     float                       = 0.0
    notes:             List[str]                   = field(default_factory=list)

Full audit trail of a single routing decision — emitted regardless of whether a match was found.

Fields

Methods

add_note(self, note: str)

Appends a string to self.notes. Used throughout RoutingPipeline and ExecutionEngine to record human-readable context (e.g. "Sequential: succeeded on 'docs_qa'.", "No group matched the query.").

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

Returns a JSON-friendly summary: request_id, query (truncated to 120 chars), selected_group, selected_route, confidence, used_llm_fallback, used_cache, a timing_ms sub-dict (each value rounded to 2 decimals), the first 5 candidates (each via RouteCandidate.to_dict()), and notes.

[!NOTE] Trace serialization truncates candidates RoutingTrace.to_dict() includes only self.candidates[:5] regardless of ScoringConfig.top_k — if you need the full candidate list beyond 5, access trace.candidates directly rather than trace.to_dict()["candidates"].

Related APIs: RouteCandidate, RoutingResult

RoutingResult

@dataclass
class RoutingResult:
    response:    Optional[HandlerResponse]
    trace:       RoutingTrace
    matched:     bool  = False

The final result object returned by HierarchicalRouter.route_query() / route_async() / route(). Bundles the handler's response with the full routing trace so callers can inspect why a route was chosen without additional instrumentation.

Fields

Properties

Methods

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

Returns {"matched", "success", "route", "group", "confidence", "content", "error", "time_ms", "trace"} — trace is the nested full RoutingTrace.to_dict() output.

Magic methods

• __bool__ — returns self.success, so if result: means "matched and the handler succeeded."
• __repr__ — if matched, shows route/group/confidence/time/success; otherwise shows RoutingResult(no_match, time=...ms).

Examples

1. Basic — check success before using content:

result = router.route_query("What's the weather in Cairo?")
if result:
    print(result.content)
else:
    print("Routing failed:", result.response.error if result.response else "no response")

2. Intermediate — inspect the winning candidate's scores:

if result.trace.candidates:
    best = result.trace.candidates[0]
    print(f"{best.route_name}: semantic={best.semantic_score:.3f} keyword={best.keyword_score:.3f}")

3. Advanced — full trace serialization for logging:

import json
logger.info("routing_trace", extra={"trace": json.dumps(result.to_dict())})

4. Error handling — distinguishing "no match" from "handler failed":

if not result.matched:
    print("No route matched:", result.trace.notes)
elif not result.success:
    print("Route matched but handler failed:", result.response.error)
else:
    print("Success:", result.content)

Related APIs: HandlerResponse, RoutingTrace, MultiRoutingResult

MultiRoutingResult

@dataclass
class MultiRoutingResult:
    results:          List[RoutingResult]
    aggregated:       Optional[HandlerResponse]  = None
    trace:            Optional[RoutingTrace]     = None

Wraps results from multi-route (parallel) execution — produced by ExecutionEngine._execute_parallel / _execute_parallel_async.

Fields

Properties

• success -> bool — bool(self.aggregated and self.aggregated.success).
• content -> Any — self.aggregated.content if self.aggregated else None.

Methods

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

Returns {"success", "content", "results" (list of each RoutingResult.to_dict()), "trace"}.

Related APIs: ExecutionEngine, BaseAggregator