src/corpus_router.py

"""Decide which corpora a question needs to reach.

Deliberately rule-based rather than model-based. Routing here is a cheap,
verifiable decision on a handful of surface signals, and an LLM call per question
would add latency and non-determinism to a step a substring match answers
exactly.

The policy is asymmetric on purpose:

* Narrowing to a subset requires the question to **name** a corpus —
  "nach dem Rahmenvertrag", "§ 31 SGB V".
* Everything else queries all corpora and lets score fusion decide.

That asymmetry follows from the cost of being wrong. One corpus too many costs a
few hundred milliseconds; one corpus too few means the assistant answers
confidently from the wrong statute, which is the failure mode this pipeline
exists to prevent. A bare "§ 16" is therefore *not* narrowed even though the
Rahmenvertrag is the likelier intent — both documents have a § 16, and they say
entirely different things.
"""

from __future__ import annotations

from typing import Any, Dict, List, Sequence

from corpus_registry import CORPUS_ALIASES


def _normalize(text: str) -> str:
    return " ".join(str(text or "").lower().split())


def mentioned_corpora(question: str, corpus_ids: Sequence[str]) -> List[str]:
    """Corpora named in the question, in registry order.

    A statute reference such as "§ 31 SGB V" is covered by this same check: the
    statute name is itself an alias, so no separate norm-reference parsing is
    needed here. Parsing the § itself is the retriever's job.
    """
    haystack = _normalize(question)
    hits: List[str] = []
    for corpus_id in corpus_ids:
        candidates = (*CORPUS_ALIASES.get(corpus_id.lower(), ()), corpus_id.lower())
        if any(alias and alias in haystack for alias in candidates):
            hits.append(corpus_id)
    return hits


def route_question(question: str, corpus_ids: Sequence[str]) -> List[str]:
    """Return the corpora to query. Falls back to all of them without a signal."""
    available = list(corpus_ids)
    if len(available) <= 1:
        return available
    return mentioned_corpora(question, available) or available


def explain_routing(question: str, corpus_ids: Sequence[str]) -> Dict[str, Any]:
    """Routing decision with its reason, for /debug/routing and tests."""
    available = list(corpus_ids)
    named = mentioned_corpora(question, available)
    selected = route_question(question, available)
    return {
        "question": question,
        "available": available,
        "named_corpora": named,
        "selected": selected,
        "reason": "explicit_mention" if named and len(available) > 1 else "no_signal_query_all",
    }