File size: 4,125 Bytes
3e363e2 | 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 | """Deterministic evidence verification; model scores never override graph truth."""
from __future__ import annotations
from dataclasses import replace
from graph.constraints import path_satisfies_constraints
from graph.graph import GraphStore
from graph.types import ConstraintState, GraphPath, GraphQuery, ReasoningStatus
def verify_path(
path: GraphPath, graph: GraphStore, query: GraphQuery | ConstraintState
) -> GraphPath:
constraints = (
query
if isinstance(query, ConstraintState)
else ConstraintState.from_query(query)
)
valid = bool(path.steps)
for index, step in enumerate(path.steps):
valid = (
valid and step.source_id in graph.nodes and step.target_id in graph.nodes
)
valid = valid and graph.has_edge(
step.source_id, step.relation_id, step.target_id
)
if step.source_type is not None:
valid = valid and graph.nodes.get(step.source_id) == step.source_type
if step.target_type is not None:
valid = valid and graph.nodes.get(step.target_id) == step.target_type
if index and path.steps[index - 1].target_id != step.source_id:
valid = False
valid = valid and path_satisfies_constraints(path, constraints, graph)
return replace(path, verified=valid)
def classify_evidence(
*,
candidate_paths: tuple[GraphPath, ...],
graph: GraphStore,
query: GraphQuery,
) -> ReasoningStatus:
verified = tuple(verify_path(path, graph, query) for path in candidate_paths)
valid = tuple(path for path in verified if path.verified)
if not valid:
return ReasoningStatus.INSUFFICIENT_EVIDENCE
return ReasoningStatus.SUPPORTED
def detect_conflicting_paths(
paths: tuple[GraphPath, ...], graph: GraphStore | None = None
) -> tuple[tuple[GraphPath, ...], ...]:
"""Return groups of paths that assert different values for the same claim.
Three conditions must hold for a contradiction (PLAN section 18):
1. the paths assert a *single* fact, so only one-step claims are grouped --
a multi-hop path is an explanation chain, and two chains reaching
different endpoints are competing answers, not a conflict;
2. they share a subject and relation;
3. that relation is **functional**. ``owns`` and ``uses`` are set-valued, so
two values are two facts; ``works_at`` is single-valued, so two values
cannot both hold. Cardinality is read from the graph, which is the source
of truth. When the graph declares nothing, no contradiction is asserted:
silence must not be read as a conflict.
"""
groups: dict[tuple[str, str], list[GraphPath]] = {}
for path in paths:
if len(path.steps) != 1:
continue
first = path.steps[0]
if not _is_functional(graph, first.relation_id):
continue
groups.setdefault((first.source_id, first.relation_id), []).append(path)
return tuple(
tuple(group)
for group in groups.values()
if len({path.endpoint_id for path in group}) > 1
)
def _is_functional(graph: GraphStore | None, relation_id: str) -> bool:
if graph is None:
return False
attributes = graph.relation_attributes.get(relation_id, {})
return str(attributes.get("functional", "false")).lower() == "true"
def find_contradictions(
graph: GraphStore,
) -> tuple[tuple[str, str, tuple[str, ...]], ...]:
"""Enumerate functional relations in ``graph`` that assert multiple values.
Used by the evaluator and the API to report conflicts that exist in the graph
itself, independently of which paths a search happened to return.
"""
claims: dict[tuple[str, str], set[str]] = {}
for edge in graph.edges:
if not _is_functional(graph, edge.relation_id):
continue
claims.setdefault((edge.source_id, edge.relation_id), set()).add(edge.target_id)
return tuple(
(source, relation, tuple(sorted(targets)))
for (source, relation), targets in sorted(claims.items())
if len(targets) > 1
)
|