File size: 5,030 Bytes
465b477
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
"""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


@dataclass
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

    @property
    def valid(self) -> bool:
        return self.exists and self.supported


@dataclass
class PathReport:
    checks: list[TripleCheck]
    connected: bool
    touches_question: bool
    covers_answer: bool

    @property
    def all_valid(self) -> bool:
        return bool(self.checks) and all(c.valid for c in self.checks)

    @property
    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)