Skip to content
← How we build

backend/src/router/pattern_matcher.py

"""Phase 1 pattern and entity scoring (zero LLM cost)."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from router.entities import extract_entities


@dataclass
class RouteScore:
    route_id: str
    pattern: str | None
    confidence: float
    entities: dict[str, str]
    matched: bool
    destination_type: str
    destination_id: str


def _required_entities_present(
    route: dict[str, Any],
    entities: dict[str, str],
) -> bool:
    required = route.get("entities", {}).get("required", [])
    return all(entity in entities for entity in required)


def score_routes(message: str, routes: list[dict[str, Any]]) -> list[RouteScore]:
    normalized = message.lower()
    entities = extract_entities(message)
    scores: list[RouteScore] = []

    for route in routes:
        route_id = route["id"]
        base_confidence = float(route.get("base_confidence", 0.5))
        destination_type = "pipeline" if route.get("pipeline") else "skill"
        destination_id = route.get("pipeline") or route.get("skill") or route_id

        best_pattern: str | None = None
        best_confidence = 0.0
        matched = False

        for pattern in route.get("patterns", []):
            if pattern.lower() in normalized:
                matched = True
                confidence = base_confidence
                if best_confidence < confidence:
                    best_confidence = confidence
                    best_pattern = pattern

        if matched and not _required_entities_present(route, entities):
            best_confidence = min(best_confidence, 0.20)
            matched = False

        scores.append(
            RouteScore(
                route_id=route_id,
                pattern=best_pattern,
                confidence=best_confidence,
                entities=entities,
                matched=matched,
                destination_type=destination_type,
                destination_id=destination_id,
            )
        )

    scores.sort(key=lambda item: item.confidence, reverse=True)
    return scores

Follow the work.

Occasional updates on SignalFoundry, MarketCompass, and what we are building at CompassFoundry Labs.

No spam. Unsubscribe anytime.