"""Deterministic and neural baseline adapters used by the evaluator.""" from __future__ import annotations import math from collections.abc import Callable, Mapping from dataclasses import dataclass, replace from itertools import islice import torch from graph.graph import GraphStore from graph.model import GraphModel from graph.types import ( EntityAnswer, GraphPath, GraphQuery, PathStep, ReasoningResult, ReasoningStatus, ) from graph.verification import detect_conflicting_paths, verify_path #: Every baseline is a function from a graph and a query to a typed result. Predictor = Callable[[GraphStore, GraphQuery], ReasoningResult] class BaselineUnavailableError(RuntimeError): """Raised when a named upstream baseline has no supplied checkpoint/adapter.""" #: Paths enumerated from each start entity. A cap on the whole search lets the #: first start's dense neighbourhood starve every later start of paths. PATHS_PER_START = 8192 def _bounded_paths(graph: GraphStore, starts, max_hops: int): for start in starts: yield from islice(graph.iter_paths((start,), max_hops), PATHS_PER_START) def heuristic_graph_search(graph: GraphStore, query: GraphQuery) -> ReasoningResult: paths = tuple( path for path in _bounded_paths( graph, query.start_entity_ids or query.start_entities, query.max_hops ) if query.min_hops <= path.hop_count <= query.max_hops ) paths = tuple(verify_path(path, graph, query) for path in paths) valid = tuple(path for path in paths if path.verified) answers = tuple( EntityAnswer(endpoint, 1.0) for endpoint in dict.fromkeys( path.endpoint_id for path in valid if path.endpoint_id ) ) relation_answers = tuple( dict.fromkeys(path.steps[-1].relation_id for path in valid if path.steps) ) boolean_answer = bool(answers) if query.answer_mode == "boolean" else None aggregate_value = ( len(answers) if query.answer_mode in {"count", "aggregate"} else None ) entity_answers = answers if query.answer_mode == "entity" else () conflicts = tuple( path for group in detect_conflicting_paths(valid, graph) for path in group ) status = ( ReasoningStatus.SUPPORTED if valid else ( ReasoningStatus.UNSUPPORTED if query.target_entity_ids else ReasoningStatus.INSUFFICIENT_EVIDENCE ) ) if conflicts and query.operation in {"verify", "boolean", "is_supported"}: status = ReasoningStatus.CONTRADICTED if ( query.target_entity_ids and valid and not (set(query.target_entity_ids) & {path.endpoint_id for path in valid}) ): status = ReasoningStatus.UNSUPPORTED if query.answer_mode == "boolean" and status != ReasoningStatus.SUPPORTED: boolean_answer = ( False if status in {ReasoningStatus.UNSUPPORTED, ReasoningStatus.CONTRADICTED} else None ) entity_answers = ( answers if query.answer_mode == "entity" and status == ReasoningStatus.SUPPORTED else () ) return ReasoningResult( status=status, answers=entity_answers, supporting_paths=valid if query.return_paths and status == ReasoningStatus.SUPPORTED else (), conflicting_paths=conflicts if query.return_paths else (), confidence=1.0 if valid else 0.0, calibrated=False, model_version="heuristic-graph-search", graph_version=graph.graph_version, reasoning_steps=max((path.hop_count for path in valid), default=0), relation_answers=relation_answers if query.answer_mode == "relation" else (), boolean_answer=boolean_answer, aggregate_value=aggregate_value, ) def adamic_adar_search(graph: GraphStore, query: GraphQuery) -> ReasoningResult: """Use local common-neighbor evidence to rank otherwise valid graph paths.""" starts = query.start_entity_ids or query.start_entities neighbor_sets = { node: {edge.target_id for edge in graph.outgoing(node)} for node in graph.nodes } paths = tuple( path for path in _bounded_paths(graph, starts, query.max_hops) if query.min_hops <= path.hop_count <= query.max_hops ) scored: list[GraphPath] = [] for path in paths: checked = verify_path(path, graph, query) if not checked.verified: continue score = 0.0 if path.start_id and path.endpoint_id: common = neighbor_sets.get(path.start_id, set()) & neighbor_sets.get( path.endpoint_id, set() ) score = sum( 1.0 / max(1.0, math.log1p(len(neighbor_sets.get(node, ())))) for node in common ) scored.append(replace(checked, score=score)) scored.sort(key=lambda path: path.score, reverse=True) answers = tuple( EntityAnswer(endpoint, path.score) for endpoint, path in ( (path.endpoint_id, path) for path in scored if path.endpoint_id ) ) relation_answers = tuple( dict.fromkeys(path.steps[-1].relation_id for path in scored if path.steps) ) boolean_answer = bool(answers) if query.answer_mode == "boolean" else None aggregate_value = ( len(answers) if query.answer_mode in {"count", "aggregate"} else None ) conflicts = tuple( path for group in detect_conflicting_paths(tuple(scored), graph) for path in group ) status = ( ReasoningStatus.SUPPORTED if scored else ( ReasoningStatus.UNSUPPORTED if query.target_entity_ids else ReasoningStatus.INSUFFICIENT_EVIDENCE ) ) if conflicts and query.operation in {"verify", "boolean", "is_supported"}: status = ReasoningStatus.CONTRADICTED if ( query.target_entity_ids and scored and not (set(query.target_entity_ids) & {path.endpoint_id for path in scored}) ): status = ReasoningStatus.UNSUPPORTED if query.answer_mode == "boolean" and status != ReasoningStatus.SUPPORTED: boolean_answer = ( False if status in {ReasoningStatus.UNSUPPORTED, ReasoningStatus.CONTRADICTED} else None ) entity_answers = ( answers if query.answer_mode == "entity" and status == ReasoningStatus.SUPPORTED else () ) return ReasoningResult( status=status, answers=entity_answers, supporting_paths=tuple(scored) if query.return_paths and status == ReasoningStatus.SUPPORTED else (), conflicting_paths=conflicts if query.return_paths else (), confidence=1.0 if scored else 0.0, calibrated=False, model_version="adamic-adar", graph_version=graph.graph_version, reasoning_steps=max((path.hop_count for path in scored), default=0), relation_answers=relation_answers if query.answer_mode == "relation" else (), boolean_answer=boolean_answer, aggregate_value=aggregate_value, ) def neural_link_predictor( graph: GraphStore, query: GraphQuery, model: GraphModel, *, gat: bool = False ) -> ReasoningResult: """One-hop GraphSAGE/GAT link-prediction baseline, separate from GRM search.""" expected_encoder = "gat" if gat else "sage" actual_encoder = ( "gat" if model.config.graph_encoder == "gat" or model.config.use_gat else "sage" ) if actual_encoder != expected_encoder: raise BaselineUnavailableError( f"{expected_encoder} baseline requires a model configured with graph_encoder={expected_encoder!r}; got {actual_encoder!r}" ) example = model.graph_to_example(graph) # ``graph_to_example`` builds CPU tensors; the encoder lives wherever the # model was placed. Without this move the baseline works on CPU and raises a # device mismatch on GPU, which is how it was actually found. device = next(model.parameters()).device encoded = model.graph_encoder(example.graph.to(device)) starts = set(query.start_entity_ids or query.start_entities) candidates: list[GraphPath] = [] for edge in graph.edges: if edge.source_id not in starts: continue source_index = example.node_external_ids.index(edge.source_id) target_index = example.node_external_ids.index(edge.target_id) score = float( (encoded.node_states[source_index] * encoded.node_states[target_index]) .sum() .detach() .cpu() ) candidate = GraphPath( ( PathStep( edge.source_id, edge.relation_id, edge.target_id, edge.source_type, edge.target_type, edge.attributes, ), ), score=score, ) checked = verify_path(candidate, graph, query) if checked.verified: candidates.append(checked) candidates.sort(key=lambda path: path.score, reverse=True) answers = tuple( EntityAnswer(path.endpoint_id, path.score) for path in candidates if path.endpoint_id ) return ReasoningResult( status=ReasoningStatus.SUPPORTED if candidates else ReasoningStatus.INSUFFICIENT_EVIDENCE, answers=answers if query.answer_mode == "entity" else (), supporting_paths=tuple(candidates[: model.config.answer_top_k]) if query.return_paths else (), conflicting_paths=(), confidence=float( torch.sigmoid( torch.tensor(candidates[0].score if candidates else -4.0) ).item() ), calibrated=False, model_version=f"{expected_encoder}-link-predictor", graph_version=graph.graph_version, reasoning_steps=1 if candidates else 0, ) @dataclass(frozen=True) class ReasonOverrides: """Typed keyword overrides for :meth:`GraphModel.reason`. A typed record rather than a ``dict[str, object]``: the untyped mapping could not be splatted into ``reason`` without defeating the type checker, and the core pipeline is meant to avoid ``Any`` (PLAN section 51). """ use_pat: bool | None = None verify: bool | None = None select: bool = True query_conditioned: bool = True use_types: bool = True beam_width: int | None = None def describe(self) -> dict[str, str]: """Human-readable summary of the fields this ablation changes.""" default = ReasonOverrides() return { field: str(getattr(self, field)) for field in ( "use_pat", "verify", "select", "query_conditioned", "use_types", "beam_width", ) if getattr(self, field) != getattr(default, field) } #: Runtime ablations change how one fixed set of weights is used. Ablations that #: require *different weights* (no path supervision, no PAT supervision) are not #: listed here: they are produced by training a variant and are dispatched #: through an explicit predictor instead. Returning an unmodified #: ``model.reason`` under an ablation name, as the previous implementation did #: for ``graph-without-query-conditioning`` and ``graph-without-path-supervision``, #: reports a comparison that never ran. RUNTIME_ABLATIONS: Mapping[str, ReasonOverrides] = { "graph": ReasonOverrides(), "graph-without-pat": ReasonOverrides(use_pat=False), "graph-without-verifier": ReasonOverrides(verify=False), "graph-without-selection": ReasonOverrides(select=False), "graph-without-query-conditioning": ReasonOverrides(query_conditioned=False), "graph-untyped": ReasonOverrides(use_types=False), "graph-greedy": ReasonOverrides(beam_width=1), } #: Ablations that can only be evaluated against separately trained weights. TRAINING_ABLATIONS: frozenset[str] = frozenset( {"graph-without-path-supervision", "graph-without-pat-supervision"} ) #: The model was renamed from "grm" to "graph". Evaluation artifacts produced before #: the rename key their rows by the old names, so those names keep resolving rather #: than becoming "unknown baseline" errors when an old report is re-read or compared. DEPRECATED_BASELINE_NAMES: Mapping[str, str] = { "grm": "graph", "grm-3b": "graph", "grm-full": "graph", **{ f"grm-{suffix}": f"graph-{suffix}" for suffix in ( "without-pat", "without-verifier", "without-selection", "without-query-conditioning", "untyped", "greedy", "without-path-supervision", "without-pat-supervision", ) }, } def apply_overrides( model: GraphModel, graph: GraphStore, query: GraphQuery, overrides: ReasonOverrides ) -> ReasoningResult: """Call ``reason`` with an ablation's overrides, fully typed.""" return model.reason( graph, query, use_pat=overrides.use_pat, verify=overrides.verify, beam_width=overrides.beam_width, use_types=overrides.use_types, query_conditioned=overrides.query_conditioned, select=overrides.select, ) def make_baseline( name: str, model: GraphModel | None = None, upstream: Predictor | None = None, ): """Resolve a baseline name to a predictor, or fail loudly. ``upstream`` supplies the released G-Reasoner-34M predictor. It is injected rather than constructed here because it needs the downloaded checkpoint, a text embedding model and ``torch_geometric``, none of which should become an import dependency of the evaluation package. """ if name == "heuristic": return heuristic_graph_search if name == "adamic_adar": return adamic_adar_search if name == "g-reasoner-34m": if upstream is None: raise BaselineUnavailableError( "g-reasoner-34m requires the released upstream checkpoint and a text " "embedder; build one with graph.evaluation.upstream_baseline and pass " "it as upstream=" ) return upstream if name == "g-reasoner-34m-finetuned": raise BaselineUnavailableError( "g-reasoner-34m-finetuned requires fine-tuning the released checkpoint on " "this task; the released weights alone are the zero-shot baseline" ) if name in TRAINING_ABLATIONS: raise BaselineUnavailableError( f"{name} is a training-time ablation: evaluate it against weights trained " f"with that objective disabled, then pass them as an explicit predictor" ) if name in {"graphsage", "gat"}: if model is None: raise ValueError(f"baseline {name} requires a configured model") return lambda graph, query: neural_link_predictor( graph, query, model, gat=name == "gat" ) if model is None: raise ValueError(f"baseline {name} requires a GRM model adapter") # Pre-rename names stay readable so old artifacts do not become errors. key = DEPRECATED_BASELINE_NAMES.get(name, name) if key in RUNTIME_ABLATIONS: overrides = RUNTIME_ABLATIONS[key] return lambda graph, query: apply_overrides(model, graph, query, overrides) raise ValueError(f"unknown baseline {name!r}; known: {sorted(RUNTIME_ABLATIONS)}")