Spaces:
Running on Zero
Running on Zero
| """Checking a claimed reasoning path against the graph. | |
| Used in two places with the same code: inside GraphTrace's verifier, and in evaluation | |
| (path validity / completeness) for every method, including text-only RAG. | |
| A claimed triple is *valid* when (a) both names link to graph nodes, (b) the graph | |
| contains that (subject, relation, object) fact, and (c) at least one evidence edge for | |
| the fact is source-supported: its NLI support score is at least ``min_support``, or no | |
| score was computed and its verbatim quote anchored at extraction time (true for every | |
| stored edge). | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass, field | |
| from .graph import KG | |
| from .linking import EntityLinker | |
| from .schemas import RELATIONS | |
| RELATION_ALIASES = { | |
| "evaluated_on": "evaluates_on", | |
| "evaluate_on": "evaluates_on", | |
| "evaluates": "evaluates_on", | |
| "propose": "proposes", | |
| "proposed": "proposes", | |
| "use": "uses", | |
| "used": "uses", | |
| "address": "addresses", | |
| "reports": "reports_metric", | |
| "compared_with": "compares_with", | |
| "compares": "compares_with", | |
| } | |
| def canonical_relation(r: str) -> str | None: | |
| r = r.strip().lower().replace(" ", "_").replace("-", "_") | |
| r = RELATION_ALIASES.get(r, r) | |
| return r if r in RELATIONS else None | |
| class TripleCheck: | |
| triple: list[str] | |
| subject_id: str | None | |
| relation: str | None | |
| object_id: str | None | |
| exists: bool | |
| supported: bool | |
| edge_ids: list[str] = field(default_factory=list) | |
| problem: str | None = None | |
| def valid(self) -> bool: | |
| return self.exists and self.supported | |
| class PathReport: | |
| checks: list[TripleCheck] | |
| connected: bool | |
| touches_question: bool | |
| covers_answer: bool | |
| def all_valid(self) -> bool: | |
| return bool(self.checks) and all(c.valid for c in self.checks) | |
| def valid_fraction(self) -> float | None: | |
| return sum(c.valid for c in self.checks) / len(self.checks) if self.checks else None | |
| def problems(self) -> list[str]: | |
| out = [f"{c.triple}: {c.problem}" for c in self.checks if c.problem] | |
| if self.checks and not self.connected: | |
| out.append("the path is not connected") | |
| if self.checks and not self.touches_question: | |
| out.append("the path does not start from an entity mentioned in the question") | |
| if not self.covers_answer: | |
| out.append("the answer entities do not appear on the path") | |
| return out | |
| def to_dict(self) -> dict: | |
| return { | |
| "checks": [c.__dict__ for c in self.checks], | |
| "connected": self.connected, | |
| "touches_question": self.touches_question, | |
| "covers_answer": self.covers_answer, | |
| "all_valid": self.all_valid, | |
| } | |
| def _link(linker: EntityLinker, name: str) -> str | None: | |
| lk = linker.link_name(name) | |
| return lk.node_id if lk else None | |
| def check_triple(kg: KG, linker: EntityLinker, triple: list[str], min_support: float | None) -> TripleCheck: | |
| s_name, r_name, o_name = (list(triple) + ["", "", ""])[:3] | |
| s, o, r = _link(linker, s_name), _link(linker, o_name), canonical_relation(r_name) | |
| if r is None: | |
| return TripleCheck(list(triple), s, None, o, False, False, problem=f"unknown relation '{r_name}'") | |
| if s is None or o is None: | |
| missing = s_name if s is None else o_name | |
| return TripleCheck(list(triple), s, r, o, False, False, problem=f"entity '{missing}' is not in the graph") | |
| edges = kg.fact_edges(s, r, o) | |
| if not edges: | |
| return TripleCheck(list(triple), s, r, o, False, False, problem="no such fact in the graph") | |
| supported = [e for e in edges if min_support is None or e.confidence is None or e.confidence >= min_support] | |
| return TripleCheck( | |
| list(triple), | |
| s, | |
| r, | |
| o, | |
| True, | |
| bool(supported), | |
| edge_ids=[e.edge_id for e in edges], | |
| problem=None if supported else "the fact's evidence does not support it (low support score)", | |
| ) | |
| def check_path( | |
| kg: KG, | |
| linker: EntityLinker, | |
| triples: list[list[str]], | |
| question_nodes: set[str], | |
| answer_nodes: set[str], | |
| min_support: float | None = 0.5, | |
| ) -> PathReport: | |
| checks = [check_triple(kg, linker, t, min_support) for t in triples] | |
| nodes = [n for c in checks for n in (c.subject_id, c.object_id) if n] | |
| parent = {n: n for n in nodes} | |
| def find(x: str) -> str: | |
| while parent[x] != x: | |
| parent[x] = parent[parent[x]] | |
| x = parent[x] | |
| return x | |
| for c in checks: | |
| if c.subject_id and c.object_id: | |
| parent[find(c.subject_id)] = find(c.object_id) | |
| roots = {find(n) for n in nodes} | |
| connected = len(roots) <= 1 | |
| touches = not question_nodes or any(n in question_nodes for n in nodes) | |
| covers = not answer_nodes or answer_nodes <= set(nodes) | |
| return PathReport(checks=checks, connected=connected, touches_question=touches, covers_answer=covers) | |