"""
Query routing heuristics for hybrid legal retrieval.
"""
import logging
import re
from dataclasses import dataclass
from typing import Optional
from lalandre_core.config import get_config
from .query_parser import LLMQueryParserClient, ParsedQueryIntent
logger = logging.getLogger(__name__)
[docs]
@dataclass(frozen=True)
class RetrievalPlan:
"""Resolved retrieval strategy for a user question."""
profile: str
granularity: Optional[str]
top_k: int
include_relations_hint: bool
rationale: str
use_graph: bool = False
execution_mode: str = "hybrid"
routing_source: str = "heuristic"
search_query: Optional[str] = None
intent_label: Optional[str] = None
parser_confidence: Optional[float] = None
[docs]
class QueryRouter:
"""
Lightweight router for selecting retrieval settings by query intent.
By default, routing is heuristic-only (fast and deterministic). An optional
An LLM parser can be injected to classify intent and normalize user queries.
"""
_REFERENCE_CUE_RE = re.compile(
r"\b("
r"article\s+\d+|art\.\s*\d+|"
r"3\d{4}[A-Z]\d{4}|"
r"\d{2,4}/\d{2,4}/(?:UE|EU|CE|EC|CEE)|"
r"directive|regulation|r[èe]glement|decision|d[ée]cision"
r")\b",
re.IGNORECASE,
)
_RELATION_CUE_RE = re.compile(
r"\b("
r"amend|amendment|modifi|abrog|repeal|replace|transpose|implement|"
r"transposition|relation|lien|d[ée]roge|compl[eé]ment|corrig"
r")",
re.IGNORECASE,
)
_BROAD_CUE_RE = re.compile(
r"\b("
r"overview|global|synth[èe]se|panorama|r[ée]sum[ée]|compare|comparaison|"
r"diff[ée]rence|impact global|vision d['’]ensemble|vue d['’]ensemble"
r")\b",
re.IGNORECASE,
)
def __init__(self, *, intent_parser: Optional[LLMQueryParserClient] = None) -> None:
self.intent_parser = intent_parser
[docs]
def route(
self,
*,
question: str,
top_k: int,
requested_granularity: Optional[str],
) -> RetrievalPlan:
"""Route a question to the appropriate retrieval profile and defaults."""
bounded_top_k = max(int(top_k), 1)
parsed_intent = self._parse_with_intent_parser(
question=question,
top_k=bounded_top_k,
requested_granularity=requested_granularity,
)
# Respect explicit user granularity except "all", which is treated as auto.
if requested_granularity in {"chunks", "subdivisions"}:
has_relation_cue = bool(self._RELATION_CUE_RE.search(question))
return RetrievalPlan(
profile="manual_override",
granularity=requested_granularity,
top_k=bounded_top_k,
include_relations_hint=(
parsed_intent.include_relations_hint if parsed_intent is not None else has_relation_cue
),
use_graph=(
parsed_intent.use_graph or parsed_intent.include_relations_hint
if parsed_intent is not None
else has_relation_cue
),
rationale=(
f"Granularity forced by request: {requested_granularity}."
if parsed_intent is None
else (f"LLM intent parse with manual granularity override. {parsed_intent.rationale}")
),
execution_mode="hybrid",
routing_source="llm" if parsed_intent is not None else "heuristic",
search_query=(parsed_intent.normalized_query if parsed_intent is not None else None),
intent_label=(parsed_intent.intent_label if parsed_intent is not None else None),
parser_confidence=(parsed_intent.confidence if parsed_intent is not None else None),
)
if parsed_intent is not None:
parsed_plan = self._plan_from_parsed(
parsed_intent=parsed_intent,
default_top_k=bounded_top_k,
)
if parsed_plan is not None:
return parsed_plan
config = get_config()
search_cfg = config.search
graph_cfg = config.graph
has_reference = bool(self._REFERENCE_CUE_RE.search(question))
has_relation_intent = bool(self._RELATION_CUE_RE.search(question))
has_broad_intent = (
bool(self._BROAD_CUE_RE.search(question)) or len(question) >= search_cfg.query_router_broad_query_min_chars
)
if has_broad_intent:
return RetrievalPlan(
profile="global_overview",
granularity="all",
top_k=max(bounded_top_k, search_cfg.query_router_global_overview_min_top_k),
include_relations_hint=True,
use_graph=True,
rationale=("Broad question detected: enable community-aware global retrieval with relation context."),
execution_mode="global",
routing_source="heuristic",
)
if has_reference:
return RetrievalPlan(
profile="citation_precision",
granularity="chunks",
top_k=max(bounded_top_k, search_cfg.query_router_citation_precision_min_top_k),
include_relations_hint=has_relation_intent,
use_graph=has_relation_intent or graph_cfg.use_graph_in_rag,
rationale="Explicit legal references detected: prioritize chunk precision.",
execution_mode="hybrid",
routing_source="heuristic",
)
if has_relation_intent:
return RetrievalPlan(
profile="relationship_focus",
granularity="chunks",
top_k=max(bounded_top_k, search_cfg.query_router_relationship_focus_min_top_k),
include_relations_hint=True,
use_graph=True,
rationale="Relationship intent detected: prioritize chunks and relation context.",
execution_mode="hybrid",
routing_source="heuristic",
)
return RetrievalPlan(
profile="contextual_default",
granularity="chunks",
top_k=max(bounded_top_k, search_cfg.query_router_contextual_default_min_top_k),
include_relations_hint=False,
use_graph=graph_cfg.use_graph_in_rag,
rationale=(
"No strong legal citation cues: use chunk context for semantic coverage"
+ ("; graph enrichment enabled by config." if graph_cfg.use_graph_in_rag else ".")
),
execution_mode="hybrid",
routing_source="heuristic",
)
def _parse_with_intent_parser(
self,
*,
question: str,
top_k: int,
requested_granularity: Optional[str],
) -> Optional[ParsedQueryIntent]:
if self.intent_parser is None:
return None
try:
return self.intent_parser.parse(
question=question,
top_k=top_k,
requested_granularity=requested_granularity,
)
except Exception as exc:
logger.warning("LLM query parser failed unexpectedly: %s", exc)
return None
@staticmethod
def _plan_from_parsed(
*,
parsed_intent: ParsedQueryIntent,
default_top_k: int,
) -> Optional[RetrievalPlan]:
search_cfg = get_config().search
profile = parsed_intent.profile
if profile == "global_overview":
granularity = parsed_intent.granularity or "all"
return RetrievalPlan(
profile=profile,
granularity=granularity,
top_k=max(parsed_intent.top_k or default_top_k, search_cfg.query_router_global_overview_min_top_k),
include_relations_hint=True,
use_graph=True,
rationale=parsed_intent.rationale,
execution_mode="global",
routing_source="llm",
search_query=parsed_intent.normalized_query,
intent_label=parsed_intent.intent_label,
parser_confidence=parsed_intent.confidence,
)
if profile == "citation_precision":
granularity = parsed_intent.granularity or "chunks"
return RetrievalPlan(
profile=profile,
granularity=granularity,
top_k=max(parsed_intent.top_k or default_top_k, search_cfg.query_router_citation_precision_min_top_k),
include_relations_hint=parsed_intent.include_relations_hint,
use_graph=parsed_intent.use_graph or parsed_intent.include_relations_hint,
rationale=parsed_intent.rationale,
execution_mode="hybrid",
routing_source="llm",
search_query=parsed_intent.normalized_query,
intent_label=parsed_intent.intent_label,
parser_confidence=parsed_intent.confidence,
)
if profile == "relationship_focus":
granularity = parsed_intent.granularity or "chunks"
return RetrievalPlan(
profile=profile,
granularity=granularity,
top_k=max(parsed_intent.top_k or default_top_k, search_cfg.query_router_relationship_focus_min_top_k),
include_relations_hint=True,
use_graph=True,
rationale=parsed_intent.rationale,
execution_mode="hybrid",
routing_source="llm",
search_query=parsed_intent.normalized_query,
intent_label=parsed_intent.intent_label,
parser_confidence=parsed_intent.confidence,
)
if profile == "contextual_default":
granularity = parsed_intent.granularity or "chunks"
return RetrievalPlan(
profile=profile,
granularity=granularity,
top_k=max(parsed_intent.top_k or default_top_k, search_cfg.query_router_contextual_default_min_top_k),
include_relations_hint=parsed_intent.include_relations_hint,
use_graph=parsed_intent.use_graph,
rationale=parsed_intent.rationale,
execution_mode="hybrid",
routing_source="llm",
search_query=parsed_intent.normalized_query,
intent_label=parsed_intent.intent_label,
parser_confidence=parsed_intent.confidence,
)
return None