| """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 |
| ) |
|
|
