Router Modular

HierarchicalRouter  ← primary public entry point
    │
    ├── RoutingPipeline         ← orchestrates the decision flow
    │       ├── EmbeddingProvider   ← sentence-transformer wrapper + cache
    │       ├── HybridScorer        ← weighted combination of:
    │       │       ├── SemanticScorer   (embedding cosine similarity)
    │       │       ├── KeywordScorer    (keyword veto/boost)
    │       │       └── LLMScorer        (optional, for low-confidence cases)
    │       └── ConfidenceEvaluator ← maps score → HIGH / MEDIUM / LOW / NONE
    │
    ├── ExecutionEngine         ← runs matched handler(s)
    │       ├── Single mode
    │       ├── Sequential mode (first success wins)
    │       ├── Parallel mode   (concurrent + aggregation)
    │       ├── Streaming
    │       └── Tool chaining
    │
    ├── CacheManager            ← LRU+TTL route decision cache
    ├── FeedbackEngine          ← EMA-based adaptive score biases
    └── MetricsCollector        ← rolling-window latency & hit-rate metrics

RouteGroup  ← named domain namespace (e.g. "rag", "tools", "chat")
    └── Route[]  ← individual routing destination + handler

query
  → cache lookup  (skip pipeline on hit)
  → embed query
  → score groups  (pick best domain)
  → score routes  (pick top-k within domain)
  → evaluate confidence  (HIGH / MEDIUM / LOW / NONE)
  → execute handler(s)
  → record feedback & metrics
  → RoutingResult

from fennec_community.router import (
    HierarchicalRouter, RouteGroup, Route,
    BaseHandler, HandlerRequest, HandlerResponse,
    make_rag_group, make_tools_group,
)

# 1. Define handlers
class DocsQAHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        answer = my_rag_pipeline(request.query)
        return HandlerResponse.ok(answer)

class WeatherHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        city = extract_city(request.query)
        data = fetch_weather(city)
        return HandlerResponse.ok(data)

# 2. Build route groups
rag_group = make_rag_group(priority=10)
rag_group.add_route(Route(
    name        = "docs_qa",
    description = "Answer questions from the knowledge base",
    handler     = DocsQAHandler(),
    examples    = ["What is the refund policy?", "How do I reset my password?"],
))

tools_group = make_tools_group(priority=20)
tools_group.add_route(Route(
    name        = "weather",
    description = "Get current weather for a city",
    handler     = WeatherHandler(),
    examples    = ["What's the weather in Cairo?", "Is it raining in London?"],
))

# 3. Create router and register groups
router = HierarchicalRouter()
router.register_group(rag_group)
router.register_group(tools_group)

# 4. Route a query
result = router.route_query("What are the return conditions?")
if result.success:
    print(result.content)
    print(f"Routed to: {result.route_name} ({result.confidence})")

@dataclass
class RouterConfig:
    embedding:     EmbeddingConfig     = EmbeddingConfig()
    scoring:       ScoringConfig       = ScoringConfig()
    execution:     ExecutionConfig     = ExecutionConfig()
    cache:         CacheConfig         = CacheConfig()
    feedback:      FeedbackConfig      = FeedbackConfig()
    observability: ObservabilityConfig = ObservabilityConfig()
    llm_model:     Optional[str]       = None
    llm_api_key:   Optional[str]       = None
    raise_on_no_match: bool            = False

RouterConfig.fast() -> RouterConfig

router = HierarchicalRouter(config=RouterConfig.fast())

RouterConfig.accurate() -> RouterConfig

router = HierarchicalRouter(config=RouterConfig.accurate())

from fennec_community.router import RouteKeywords

weather_keywords = RouteKeywords(
    any_of   = ["weather", "temperature", "rain", "forecast", "climate"],
    excluded = ["historical", "average", "record"],
    boost    = 0.10,
)

from fennec_community.router import BaseHandler, HandlerRequest, HandlerResponse

class MyHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        # Your logic here
        result = do_something(request.query)
        return HandlerResponse.ok(result)

handler.handle(request: HandlerRequest) -> HandlerResponse

class RAGHandler(BaseHandler):
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def handle(self, request: HandlerRequest) -> HandlerResponse:
        docs    = self.retriever.search(request.query, top_k=5)
        answer  = self.llm.generate(query=request.query, context=docs)
        return HandlerResponse.ok(
            content  = answer,
            sources  = [d.source for d in docs],
            doc_count = len(docs),
        )

async handler.handle_async(request: HandlerRequest) -> HandlerResponse

class AsyncLLMHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        # Sync fallback
        return HandlerResponse.ok(sync_call(request.query))

    async def handle_async(self, request: HandlerRequest) -> HandlerResponse:
        result = await async_llm_call(request.query)
        return HandlerResponse.ok(result)

handler.validate(request: HandlerRequest) -> bool

class SecureHandler(BaseHandler):
    def validate(self, request: HandlerRequest) -> bool:
        return "user_id" in request.context  # Require authentication

    def handle(self, request: HandlerRequest) -> HandlerResponse:
        user_id = request.context["user_id"]
        return HandlerResponse.ok(fetch_user_data(user_id))

handler.stream(request: HandlerRequest) -> Iterator[str]

class StreamingLLMHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        return HandlerResponse.ok("".join(self.stream(request)))

    def stream(self, request: HandlerRequest) -> Iterator[str]:
        for chunk in openai_stream(request.query):
            yield chunk.delta.content or ""

async handler.stream_async(request: HandlerRequest) -> AsyncIterator[str]

request.with_metadata(**kwargs) -> HandlerRequest

enriched = request.with_metadata(route_name="weather", confidence=0.91)
# enriched.metadata == {"route_name": "weather", "confidence": 0.91}
# request.metadata is unchanged

HandlerResponse.ok(content: Any, **meta) -> HandlerResponse

return HandlerResponse.ok(
    answer,
    sources    = ["doc_1", "doc_2"],
    confidence = 0.93,
    tokens     = 142,
)

HandlerResponse.fail(error: str, **meta) -> HandlerResponse

return HandlerResponse.fail(
    "City not found in weather database",
    query   = request.query,
    code    = 404,
)

Route(
    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,
)

from fennec_community.router import Route, RouteKeywords

route = Route(
    name        = "invoice_lookup",
    description = "Look up invoice details by invoice ID or customer",
    handler     = InvoiceHandler(),
    examples    = [
        "Show me invoice #INV-2024-001",
        "Find all invoices for Acme Corp",
        "What's the status of my last invoice?",
    ],
    keywords    = RouteKeywords(
        any_of   = ["invoice", "billing", "receipt", "payment"],
        excluded = ["create", "new invoice"],
        boost    = 0.08,
    ),
    tags        = {"finance", "read-only"},
    priority    = 5,
)

route.add_example(example: str) -> bool

added = route.add_example("Can you pull up invoice number 5082?")
if added:
    router.re_encode_route(route.name)

route.remove_example(example: str) -> bool

route.enable()  -> None
route.disable() -> None

# Disable during maintenance
route.disable()
# Re-enable
route.enable()

route.to_dict() -> Dict[str, Any]

import json
with open("routes.json", "w") as f:
    json.dump([r.to_dict() for r in my_routes], f)

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

with open("routes.json") as f:
    data = json.load(f)
route = Route.from_dict(data[0], handler=MyHandler())

RouteGroup(
    name:             str,
    description:      str,
    intent_examples:  Optional[List[str]]      = None,
    intent_keywords:  Optional[RouteKeywords]  = None,
    priority:         int                      = 0,
    enabled:          bool                     = True,
    metadata:         Optional[Dict[str, Any]] = None,
)

group.add_route(route: Route) -> RouteGroup

group = RouteGroup("rag", "Knowledge base Q&A")
group \
    .add_route(Route(name="docs_qa",   ...)) \
    .add_route(Route(name="faq_lookup", ...))

group.update_route(route: Route) -> RouteGroup

group.remove_route(name: str) -> bool

group.get_route(name: str) -> Optional[Route]

group.get_routes(enabled_only: bool = True) -> List[Route]

group.get_by_tag(tag: str) -> List[Route]

read_only_routes = group.get_by_tag("read-only")

group.route(
    name:        str,
    description: str,
    examples:    Optional[List[str]]     = None,
    keywords:    Optional[RouteKeywords] = None,
    priority:    int                     = 0,
    tags:        Optional[Set[str]]      = None,
    tools:       Optional[List[str]]     = None,
) -> Callable

chat_group = RouteGroup("chat", "General conversation")

@chat_group.route(
    name        = "greeting",
    description = "Greet the user",
    examples    = ["hello", "hi", "good morning"],
    priority    = 10,
)
def greet(query: str):
    return f"Hello! You said: {query}"

@chat_group.route(
    name        = "joke",
    description = "Tell a joke",
    examples    = ["tell me a joke", "say something funny"],
)
class JokeHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        return HandlerResponse.ok("Why did the robot go on a diet? Too many bytes!")

group.enable()  -> None
group.disable() -> None

group.invalidate_embeddings() -> None

group.summary() -> Dict[str, Any]

make_rag_group(priority: int = 10) -> RouteGroup

from fennec_community.router import make_rag_group

rag = make_rag_group(priority=10)
rag.add_route(Route(
    name     = "policy_qa",
    description = "Answer questions from company policy documents",
    handler  = PolicyRAGHandler(),
    examples = ["What is the vacation policy?", "How many sick days do I get?"],
))
router.register_group(rag)

make_tools_group(priority: int = 20) -> RouteGroup

from fennec_community.router import make_tools_group

tools = make_tools_group(priority=20)
tools.add_route(Route(
    name        = "send_email",
    description = "Send an email to a recipient",
    handler     = EmailHandler(),
    examples    = ["Send an email to John", "Email the report to sales@company.com"],
))
router.register_group(tools)

make_chat_group(priority: int = 0) -> RouteGroup

HierarchicalRouter(config: Optional[RouterConfig] = None)

# Default configuration
router = HierarchicalRouter()

# Custom configuration
from fennec_community.router import RouterConfig, ScoringConfig, ExecutionConfig, ExecutionMode

config = RouterConfig()
config.scoring.high_confidence_threshold = 0.90
config.execution.mode = ExecutionMode.SEQUENTIAL
config.cache.ttl_seconds = 600

router = HierarchicalRouter(config=config)

# Preset configurations
router = HierarchicalRouter(config=RouterConfig.fast())
router = HierarchicalRouter(config=RouterConfig.accurate())

router.register_group(group: RouteGroup) -> HierarchicalRouter

router \
    .register_group(make_tools_group()) \
    .register_group(make_rag_group()) \
    .register_group(make_chat_group())

router.update_group(group: RouteGroup) -> HierarchicalRouter

router.register(route: Route, group_name: Optional[str] = None) -> None

router.register(
    Route(name="quick_answer", description="...", handler=QuickHandler(), examples=[...]),
    group_name="chat"
)

router.unregister(route_name: str) -> bool

router.unregister_group(group_name: str) -> bool

router.set_fallback(handler: Union[BaseHandler, Callable]) -> None

def fallback_handler(query: str):
    return f"I couldn't find a specific answer for: {query}"

router.set_fallback(fallback_handler)

router.route_query(
    query:         str,
    context:       Optional[Dict[str, Any]] = None,
    return_result: bool                     = True,
    **handler_kwargs,
) -> RoutingResult

result = router.route_query("What is the refund policy?")
if result.success:
    print(result.content)

result = router.route_query(
    "Show my last 5 invoices",
    context = {
        "user_id":    "usr_abc123",
        "session_id": "sess_xyz789",
        "locale":     "en-US",
    },
)
print(f"Route: {result.route_name}, Confidence: {result.confidence}")
print(f"Latency: {result.total_time_ms:.1f}ms")

result = router.route_query("Compare Q1 vs Q2 revenue")
trace  = result.trace

print(f"Group selected: {trace.selected_group}")
print(f"Route selected: {trace.selected_route}")
print(f"Confidence: {trace.confidence}")
print(f"Embedding time: {trace.embedding_time_ms:.1f}ms")
print(f"Scoring time:   {trace.scoring_time_ms:.1f}ms")
print(f"Execution time: {trace.execution_time_ms:.1f}ms")

for i, candidate in enumerate(trace.candidates, 1):
    print(f"  #{i} {candidate.route_name}: score={candidate.effective_score:.3f} ({candidate.confidence.value})")

async router.route_async(request: HandlerRequest) -> RoutingResult

import asyncio
from fennec_community.router import HandlerRequest

async def handle_request(query: str):
    request = HandlerRequest(query=query, context={"user_id": "usr_001"})
    result  = await router.route_async(request)
    return result.content

answer = asyncio.run(handle_request("What's the weather in Cairo?"))

router.stream(
    query:   str,
    context: Optional[Dict[str, Any]] = None,
) -> Iterator[str]

for chunk in router.stream("Explain quantum computing simply"):
    print(chunk, end="", flush=True)

async router.stream_async(
    query:   str,
    context: Optional[Dict[str, Any]] = None,
) -> AsyncIterator[str]

async def stream_response(query: str):
    async for chunk in router.stream_async(query):
        print(chunk, end="", flush=True)

router.chain(
    query:       str,
    route_names: List[str],
    context:     Optional[Dict[str, Any]] = None,
) -> HandlerResponse

# Pipeline: extract intent → retrieve docs → generate answer
response = router.chain(
    query       = "What happened to the quarterly results?",
    route_names = ["intent_extractor", "doc_retriever", "answer_generator"],
    context     = {"user_id": "usr_001"},
)
print(response.content)

router.feedback(
    route_name: str,
    success:    bool,
    score:      float = 0.0,
    query:      str   = "",
) -> None

ema_new  = alpha × outcome + (1 - alpha) × ema_prev
bias     = clip((ema - 0.5) × 0.6,  -0.3, +0.3)

result = router.route_query("What is our refund policy?")

# After the user rates the response
user_rating = get_user_rating()  # e.g., thumbs up/down
router.feedback(
    route_name = result.route_name,
    success    = (user_rating == "up"),
    score      = result.trace.candidates[0].effective_score if result.trace.candidates else 0.0,
    query      = "What is our refund policy?",
)

router.re_encode_route(route_name: str) -> bool

route = router.get_route("docs_qa")
route.add_example("Where can I find the terms of service?")
router.re_encode_route("docs_qa")

router.re_encode_group(group_name: str) -> bool

router.metrics() -> Dict[str, Any]

import json
m = router.metrics()
print(json.dumps(m, indent=2))
print(f"Match rate: {m['match_rate_pct']}%")
print(f"P95 latency: {m['latency_ms']['p95']}ms")

router.cache_stats() -> Dict[str, Any]

stats = router.cache_stats()
print(f"Cache hit rate: {stats['route_cache']['hit_rate']}%")
print(f"Cache size: {stats['route_cache']['size']}/{stats['route_cache']['max_size']}")

router.feedback_summary() -> Dict[str, Any]

fb = router.feedback_summary()
for route_name, state in fb["states"].items():
    print(f"{route_name}: success_rate={state['success_rate']:.1%}, bias={state['bias']:+.3f}")

router.list_groups() -> List[str]

router.list_routes(group_name: Optional[str] = None) -> List[str]

all_routes   = router.list_routes()
tools_routes = router.list_routes("tools")

router.get_route(name: str) -> Optional[Route]

router.get_group(name: str) -> Optional[RouteGroup]

router.summary() -> Dict[str, Any]

import json
print(json.dumps(router.summary(), indent=2))

result.to_dict()  # → Dict with matched, success, route, group, confidence, content, error, time_ms, trace

candidate.to_dict()  # → Dict with route, group, score, semantic, keyword, llm, confidence

trace.add_note(note: str) -> None

trace.to_dict()  # → Full audit trail dict

cache.get_decision(query: str) -> Optional[Tuple[str, Optional[str]]]

cache.set_decision(
    query:      str,
    route_name: str,
    group_name: Optional[str] = None,
) -> None

cache.invalidate(query: str) -> bool

cache.clear() -> None

cache.stats() -> Dict[str, Any]

feedback_engine.record(
    route_name: str,
    group_name: Optional[str],
    success:    bool,
    score:      float,
    query:      str = "",
) -> None

feedback_engine.apply_biases(route_map: Dict[str, Route]) -> None

feedback_engine.bulk_record(records: List[Dict[str, Any]]) -> None

feedback_engine = router._feedback  # Access internal engine if needed

feedback_engine.bulk_record([
    {"route_name": "docs_qa",  "success": True,  "score": 0.91},
    {"route_name": "weather",  "success": False, "score": 0.78, "group_name": "tools"},
    {"route_name": "docs_qa",  "success": True,  "score": 0.88},
])
feedback_engine.apply_biases(router._route_map)

feedback_engine.save(path: Optional[str] = None) -> None

feedback_engine.save("feedback_state.json")

feedback_engine.get_state(route_name: str) -> Optional[RouteFeedbackState]

feedback_engine.summary() -> Dict[str, Any]

feedback_engine.reset(route_name: Optional[str] = None) -> None

from fennec_community.router import (
    HierarchicalRouter, RouterConfig, ExecutionConfig, ExecutionMode,
    RouteGroup, Route, RouteKeywords,
    BaseHandler, HandlerRequest, HandlerResponse,
    make_rag_group, make_tools_group, make_chat_group,
)

# ── Configuration ───────────────────────────────────────────────────────
config = RouterConfig()
config.scoring.high_confidence_threshold   = 0.88
config.scoring.medium_confidence_threshold = 0.65
config.execution.mode                      = ExecutionMode.SEQUENTIAL
config.feedback.persist_path               = "router_feedback.json"
config.observability.slow_route_ms         = 300.0

# ── Handlers ────────────────────────────────────────────────────────────
class PolicyHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        result = rag_search(request.query, index="policy_docs")
        return HandlerResponse.ok(result["answer"], sources=result["sources"])

class WeatherHandler(BaseHandler):
    async def handle_async(self, request: HandlerRequest) -> HandlerResponse:
        city = extract_city(request.query)
        data = await async_weather_api(city)
        return HandlerResponse.ok(data)

class ChatHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        reply = llm_chat(request.query, history=request.context.get("history", []))
        return HandlerResponse.ok(reply)

# ── Route Groups ─────────────────────────────────────────────────────────
rag = make_rag_group(priority=10)
rag.add_route(Route(
    name        = "policy_qa",
    description = "Answer questions from company policy documents",
    handler     = PolicyHandler(),
    examples    = ["What is the leave policy?", "How do I apply for remote work?"],
    keywords    = RouteKeywords(any_of=["policy", "procedure", "rule", "regulation"], boost=0.08),
))

tools = make_tools_group(priority=20)
tools.add_route(Route(
    name        = "weather",
    description = "Get current weather for a city",
    handler     = WeatherHandler(),
    examples    = ["What's the weather in Cairo?", "Is it cold in London today?"],
    keywords    = RouteKeywords(any_of=["weather", "temperature", "forecast", "rain"], boost=0.10),
))

chat = make_chat_group(priority=0)
chat.add_route(Route(
    name        = "general_chat",
    description = "General conversation and open-ended questions",
    handler     = ChatHandler(),
    examples    = ["Hello!", "Tell me a joke", "How are you?"],
))

# ── Router ────────────────────────────────────────────────────────────────
with HierarchicalRouter(config=config) as router:
    router \
        .register_group(tools) \
        .register_group(rag) \
        .register_group(chat)

    router.set_fallback(lambda q: f"Sorry, I couldn't find an answer for: {q}")

    # Route a query
    result = router.route_query(
        "What's the remote work policy?",
        context={"user_id": "usr_001", "locale": "en-US"},
    )

    if result.success:
        print(result.content)
        print(f"→ {result.route_name} [{result.confidence.value}] {result.total_time_ms:.1f}ms")

    # Record user feedback
    router.feedback(result.route_name, success=True)

import asyncio
from fennec_community.router import HierarchicalRouter, HandlerRequest

router = HierarchicalRouter()
# ... register groups ...

async def handle_query(query: str, user_id: str) -> str:
    request = HandlerRequest(
        query   = query,
        context = {"user_id": user_id},
    )
    result = await router.route_async(request)
    return result.content if result.success else f"Error: {result.response.error}"

async def main():
    answers = await asyncio.gather(
        handle_query("What's the weather?", "usr_001"),
        handle_query("Find the Q3 report",  "usr_002"),
        handle_query("Hello!",              "usr_003"),
    )
    for a in answers:
        print(a)

asyncio.run(main())

import sys

router = HierarchicalRouter()
# ... register groups with streaming handlers ...

print("Response: ", end="")
for chunk in router.stream("Explain the company's AI strategy"):
    print(chunk, end="", flush=True)
print()

router = HierarchicalRouter()

# Register pipeline stages as individual routes
router.register(Route("intent_classifier", "Classify the user intent",   handler=IntentHandler()),  group_name="pipeline")
router.register(Route("doc_retriever",     "Retrieve relevant documents", handler=RetrieverHandler()), group_name="pipeline")
router.register(Route("answer_generator",  "Generate grounded answer",    handler=GeneratorHandler()), group_name="pipeline")

response = router.chain(
    query       = "What caused the 2008 financial crisis?",
    route_names = ["intent_classifier", "doc_retriever", "answer_generator"],
    context     = {"user_id": "usr_001"},
)

print(response.content)
# Each step receives the previous step's output in context["last_result"]

from fennec_community.router import HierarchicalRouter, RouteGroup, RouteKeywords, BaseHandler, HandlerRequest, HandlerResponse

router  = HierarchicalRouter()
support = RouteGroup("support", "Customer support queries", priority=15)

@support.route(
    name        = "order_status",
    description = "Check the status of a customer order",
    examples    = ["Where is my order?", "Track order #12345", "When will my package arrive?"],
    keywords    = RouteKeywords(any_of=["order", "package", "delivery", "track"], boost=0.09),
    tags        = {"read", "orders"},
)
class OrderStatusHandler(BaseHandler):
    def handle(self, request: HandlerRequest) -> HandlerResponse:
        order_id = extract_order_id(request.query)
        status   = db.get_order_status(order_id)
        return HandlerResponse.ok(status)

@support.route(
    name        = "refund_request",
    description = "Process or check the status of a refund",
    examples    = ["I want a refund", "My refund hasn't arrived", "Cancel and refund my order"],
    keywords    = RouteKeywords(any_of=["refund", "money back", "return", "cancel"], boost=0.10),
    tags        = {"write", "refunds"},
)
def handle_refund(query: str):
    return process_refund(extract_order_id(query))

router.register_group(support)

import time, json

router = HierarchicalRouter()

# Periodic health check
def health_check():
    m = router.metrics()
    return {
        "status":       "healthy" if m["match_rate_pct"] > 80 else "degraded",
        "match_rate":   m["match_rate_pct"],
        "p95_ms":       m["latency_ms"]["p95"],
        "cache_hits":   m["cache_hit_rate_pct"],
        "top_routes":   sorted(
            m["routes"].items(),
            key=lambda x: x[1]["calls"],
            reverse=True,
        )[:5],
    }

# Inspect trace after a failure
result = router.route_query("some ambiguous query")
if not result.success:
    trace = result.trace
    print(f"No match. Top candidates:")
    for c in trace.candidates:
        print(f"  {c.route_name}: {c.effective_score:.3f} ({c.confidence.value})")
    print(f"Notes: {trace.notes}")