YOKPilot3 / source_router.py
mfirat007's picture
Upload 32 files
0a6fd56 verified
Raw
History Blame Contribute Delete
1.95 kB
from __future__ import annotations
from typing import Any
from normative_runtime import route_normative_question
def route_question(
question: str,
documents: list[dict[str, Any]] | dict[str, dict[str, Any]] | None = None,
cross_document_edges: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Route through the semantic addresses compiled from the active corpus.
``documents`` and ``cross_document_edges`` remain in the public signature
for compatibility with retrieval callers. The authoritative data is the
immutable package loaded by ``init_normative_runtime``; no law-specific
Python lexicon or article switch is maintained here.
"""
route = route_normative_question(question)
if route.get("document_scores") or not documents:
return route
docs = list(documents.values()) if isinstance(documents, dict) else list(documents or [])
document_ids = [str(item.get("document_id", "")) for item in docs if item.get("document_id")]
return {
"candidate_document_ids": document_ids,
"document_scores": {document_id: 0.0 for document_id in document_ids},
"top_document_id": "",
"top_score": 0.0,
"runner_up_score": 0.0,
"scope_confident": False,
"cross_document": False,
"candidate_edge_ids": [],
"candidate_edges": [],
"target_articles_by_document": {},
"reasons": {document_id: [] for document_id in document_ids},
}
def score_document_domain(question: str, document: dict[str, Any]) -> float:
route = route_question(question, [document], [])
return float(route.get("document_scores", {}).get(document.get("document_id", ""), 0.0))
def _target_articles(question: str) -> dict[str, list[str]]:
"""Compatibility facade for diagnostics; targets are corpus-derived."""
return dict(route_normative_question(question).get("target_articles_by_document", {}) or {})