File size: 2,658 Bytes
d80690d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | """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",
}
|