from __future__ import annotations from collections import defaultdict from pathlib import Path from typing import Any from urllib.parse import unquote from utils import normalize_for_search MCKF_BASE = "https://mitranlil.ai/mckf/" class MCKFGraphRuntime: """Queryable normative graph with an RDF/SPARQL backend and deterministic fallback.""" def __init__( self, evidence_spans: list[dict[str, Any]], cross_document_edges: list[dict[str, Any]], triples_path: Path | None = None, ) -> None: self.evidence_by_id = { str(item.get("evidence_id", "")): item for item in evidence_spans if item.get("evidence_id") } self.edges = cross_document_edges or [] self.by_document_article: dict[str, list[str]] = defaultdict(list) self.by_concept: dict[str, list[str]] = defaultdict(list) self._build_direct_indexes(evidence_spans) self.graph = None self.backend = "semantic_graph_index" self.load_error = "" self._sparql_cache: dict[tuple[str, str], list[str]] = {} if triples_path and triples_path.exists(): self._load_rdf_graph(triples_path) def status(self) -> dict[str, Any]: return { "backend": self.backend, "rdf_loaded": self.graph is not None, "triples": len(self.graph) if self.graph is not None else 0, "evidence_nodes": len(self.evidence_by_id), "load_error": self.load_error, } def retrieve( self, question_frame: dict[str, Any], requested_roles: set[str] | None = None, allowed_ids: set[str] | None = None, limit: int = 120, ) -> list[tuple[str, float, list[str]]]: requested_roles = requested_roles or set() scores: dict[str, float] = defaultdict(float) reasons: dict[str, list[str]] = defaultdict(list) values_by_predicate = { "hasActor": question_frame.get("actor", []) or [], "hasAction": question_frame.get("action", []) or [], "hasObject": question_frame.get("object", []) or [], "hasCompetentAuthority": question_frame.get("competent_authority", []) or [], "hasBeneficiary": question_frame.get("beneficiary", []) or [], } direct_weights = { "hasActor": 3.0, "hasAction": 3.2, "hasObject": 2.2, "hasCompetentAuthority": 3.4, "hasBeneficiary": 2.4, } for predicate, values in values_by_predicate.items(): for value in values: key = (predicate, normalize_for_search(str(value))) for evidence_id in self.literal_index.get(key, []): if allowed_ids is not None and evidence_id not in allowed_ids: continue scores[evidence_id] += direct_weights[predicate] reasons[evidence_id].append(f"graph:{predicate}={value}") for role in requested_roles: for evidence_id in self.role_index.get(normalize_for_search(role), []): if allowed_ids is not None and evidence_id not in allowed_ids: continue scores[evidence_id] += 2.5 reasons[evidence_id].append(f"graph:role={role}") if self.graph is not None: for evidence_id, predicate, value in self._sparql_matches(values_by_predicate): if allowed_ids is not None and evidence_id not in allowed_ids: continue scores[evidence_id] += 1.0 reasons[evidence_id].append(f"sparql:{predicate}={value}") ranked = sorted(scores, key=lambda evidence_id: scores[evidence_id], reverse=True) return [(evidence_id, scores[evidence_id], reasons[evidence_id]) for evidence_id in ranked[:limit]] def expand(self, seed_ids: list[str], limit: int = 40) -> list[tuple[str, float, list[str]]]: scores: dict[str, float] = defaultdict(float) reasons: dict[str, list[str]] = defaultdict(list) for seed_id in seed_ids: seed = self.evidence_by_id.get(seed_id, {}) if not seed: continue source_key = f"{seed.get('document_id')}::{seed.get('article_id')}" for edge in self.edges: edge_source = f"{edge.get('source_document_id')}::{edge.get('source_article_id')}" edge_target = f"{edge.get('target_document_id')}::{edge.get('target_article_id')}" target_key = edge_target if source_key == edge_source else edge_source if source_key == edge_target else "" if not target_key: continue for evidence_id in self.by_document_article.get(target_key, []): scores[evidence_id] += 1.5 reasons[evidence_id].append(f"graph_edge:{edge.get('relation_type', 'related')}") concept_id = str(seed.get("_concept_id", "") or "") for evidence_id in self.by_concept.get(concept_id, []): if evidence_id == seed_id: continue evidence = self.evidence_by_id[evidence_id] roles = set(evidence.get("semantic_roles", []) or []) if roles & {"istisna", "sart", "sure", "tanim"}: scores[evidence_id] += 0.8 reasons[evidence_id].append("graph:same_concept_qualifier") ranked = sorted(scores, key=lambda evidence_id: scores[evidence_id], reverse=True) return [(evidence_id, scores[evidence_id], reasons[evidence_id]) for evidence_id in ranked[:limit]] def _build_direct_indexes(self, evidence_spans: list[dict[str, Any]]) -> None: self.literal_index: dict[tuple[str, str], list[str]] = defaultdict(list) self.role_index: dict[str, list[str]] = defaultdict(list) for evidence in evidence_spans: evidence_id = str(evidence.get("evidence_id", "") or "") if not evidence_id: continue self.by_document_article[ f"{evidence.get('document_id')}::{evidence.get('article_id')}" ].append(evidence_id) concept_id = str(evidence.get("_concept_id", "") or "") if concept_id: self.by_concept[concept_id].append(evidence_id) frame = evidence.get("semantic_frame", {}) or {} for predicate, field in ( ("hasActor", "actor"), ("hasAction", "action"), ("hasObject", "object"), ("hasCompetentAuthority", "competent_authority"), ("hasBeneficiary", "beneficiary"), ): for value in frame.get(field, []) or []: self.literal_index[(predicate, normalize_for_search(str(value)))].append(evidence_id) for role in evidence.get("semantic_roles", []) or []: self.role_index[normalize_for_search(str(role))].append(evidence_id) def _load_rdf_graph(self, triples_path: Path) -> None: try: from rdflib import Graph # type: ignore graph = Graph() graph.parse(triples_path, format="nt") self.graph = graph self.backend = "rdflib_sparql" except Exception as exc: # noqa: BLE001 self.load_error = str(exc) def _sparql_matches(self, values_by_predicate: dict[str, list[Any]]) -> list[tuple[str, str, str]]: try: from rdflib import Literal # type: ignore except ImportError: return [] rows = [] for predicate, values in values_by_predicate.items(): for value in values: cache_key = (predicate, str(value)) if cache_key in self._sparql_cache: rows.extend((evidence_id, predicate, str(value)) for evidence_id in self._sparql_cache[cache_key]) continue literal = Literal(str(value)).n3() query = f""" PREFIX mckf: <{MCKF_BASE}> SELECT ?unit WHERE {{ ?unit mckf:{predicate} {literal} . }} """ try: matched_ids = [] for result in self.graph.query(query): unit_uri = str(result[0]) evidence_id = unquote(unit_uri.rsplit("/", 1)[-1]) if evidence_id in self.evidence_by_id: rows.append((evidence_id, predicate, str(value))) matched_ids.append(evidence_id) self._sparql_cache[cache_key] = matched_ids except Exception as exc: # noqa: BLE001 self.load_error = str(exc) return rows