"""Connected, typed evidence checks shared by GraphRAG evaluation versions. Facts are structured records emitted by fixed Cypher queries. They are expanded into the very paths those query contracts establish; query text, resolved aliases and context-only documents never create edges. Ordered paths must have compatible endpoint types. Bindings are satisfied by one connected witness, not a bag of independently occurring values. """ from __future__ import annotations import itertools import re import unicodedata from collections import defaultdict from typing import Any RELATION_ENDPOINTS = { "AUTHORED": ("Person", "Thesis"), "HAS_WORK": ("Person", "Work"), "HAS_DOCUMENT": ("Person", "Document"), "EVIDENCED_BY": ("Academic", "Document"), "HAS_KEYWORD": ("Document", "Keyword"), "IN_CATEGORY": ("Document", "Category"), "AUTHORED_DOCUMENT": ("DocumentAuthor", "Document"), "DESCRIBES": ("Document", "Topic"), } def canonical(value: Any) -> str: raw = unicodedata.normalize("NFKD", str(value if value is not None else "")) return re.sub( r"\s+", " ", re.sub( r"[^a-z0-9]+", " ", "".join( character for character in raw.casefold() if not unicodedata.combining(character) ), ), ).strip() def kind_matches(actual: str, expected: str) -> bool: return ( actual in {"Thesis", "Work"} if expected in {"Academic", "Thesis|Work"} else actual in expected.split("|") ) class EvidenceGraph: def __init__(self, result: dict[str, Any]): self.nodes: dict[tuple[str, str], dict] = {} self.conflicts: set[tuple[str, str]] = set() self.adjacency: dict[tuple[str, str], set[tuple[str, tuple[str, str]]]] = defaultdict(set) for path in result.get("paths", []): if not isinstance(path, dict): continue nodes, relationships = path.get("nodes", []), path.get("relationships", []) if not isinstance(nodes, list) or len(relationships) != max(0, len(nodes) - 1): continue keys = [self.node(row.get("kind", ""), row) for row in nodes if isinstance(row, dict)] if len(keys) != len(nodes): continue for left, relation, right in zip(keys, relationships, keys[1:]): self.edge(left, relation, right) for fact in result.get("facts", []): if isinstance(fact, dict): self.add_fact(fact) def node(self, kind: str, properties: dict) -> tuple[str, str]: properties = {key: value for key, value in properties.items() if value is not None} if kind == "Document": identity = properties.get("path") or properties.get("document_path") else: identity = properties.get("element_id") or properties.get("id") if not identity: if kind in {"Thesis", "Work"}: identity = repr( ( properties.get("person"), properties.get("title"), properties.get("year"), properties.get("document_path"), ) ) else: identity = canonical(properties.get("name")) key = (kind, str(identity)) current = self.nodes.setdefault(key, {"kind": kind}) for field in { "name", "title", "year", "start_year", "thesis_type", "work_type", "path", "source_url", }: if ( field in current and field in properties and canonical(current[field]) != canonical(properties[field]) ): self.conflicts.add(key) for field, value in properties.items(): current.setdefault(field, value) return key def edge(self, left: tuple[str, str], relation: str, right: tuple[str, str]) -> None: expected = RELATION_ENDPOINTS.get(relation) if expected is None: return if kind_matches(left[0], expected[0]) and kind_matches(right[0], expected[1]): pass elif kind_matches(right[0], expected[0]) and kind_matches(left[0], expected[1]): left, right = right, left else: return self.adjacency[left].add((relation, right)) self.adjacency[right].add((relation, left)) def neighbors(self, key: tuple[str, str], relation: str, kind: str) -> list[tuple[str, str]]: return [ target for edge, target in self.adjacency.get(key, ()) if key not in self.conflicts and target not in self.conflicts and edge in relation.split("|") and kind_matches(target[0], kind) ] def add_fact(self, fact: dict) -> None: kind = fact.get("fact_type") if kind in {"shared_keyword", "shared_category", "shared_document_author"}: singular = { "shared_keyword": "person_keyword", "shared_category": "person_category", "shared_document_author": "person_document_author", }[kind] for side in ("left", "right"): self.add_fact( { **fact, "fact_type": singular, "person": fact.get(side + "_person"), "person_element_id": fact.get(side + "_person_element_id"), "document_path": fact.get(side + "_document_path"), "source_url": fact.get(side + "_source_url"), } ) return if kind not in { "person_thesis", "person_work", "person_start_year", "person_keyword", "person_category", "person_document_author", } or not fact.get("person"): return person = self.node( "Person", { "name": fact["person"], "element_id": fact.get("person_element_id"), "start_year": fact.get("start_year") if kind == "person_start_year" else None, }, ) path = fact.get("document_path") document = ( self.node("Document", {"path": path, "source_url": fact.get("source_url")}) if path else None ) if kind in {"person_thesis", "person_work"}: academic_kind = "Thesis" if kind == "person_thesis" else "Work" prefix = "thesis" if academic_kind == "Thesis" else "work" academic = self.node( academic_kind, { "element_id": fact.get(prefix + "_element_id"), "id": fact.get(prefix + "_id"), "person": fact["person"], "document_path": path, "title": fact.get("title"), "year": fact.get("year"), prefix + "_type": fact.get(prefix + "_type"), }, ) self.edge(person, "AUTHORED" if academic_kind == "Thesis" else "HAS_WORK", academic) if document: self.edge(academic, "EVIDENCED_BY", document) elif document: self.edge(person, "HAS_DOCUMENT", document) if kind == "person_keyword" and fact.get("keyword"): target = self.node( "Keyword", { "name": fact["keyword"], "id": fact.get("keyword_id"), "element_id": fact.get("keyword_element_id"), }, ) self.edge(document, "HAS_KEYWORD", target) elif kind == "person_category" and fact.get("category"): target = self.node( "Category", {"name": fact["category"], "element_id": fact.get("category_element_id")}, ) self.edge(document, "IN_CATEGORY", target) elif kind == "person_document_author" and fact.get("document_author"): target = self.node( "DocumentAuthor", { "name": fact["document_author"], "id": fact.get("document_author_id"), "element_id": fact.get("document_author_element_id"), }, ) self.edge(target, "AUTHORED_DOCUMENT", document) if fact.get("category"): category = self.node( "Category", {"name": fact["category"], "element_id": fact.get("category_element_id")}, ) self.edge(document, "IN_CATEGORY", category) def metadata_choices( self, person: tuple[str, str], binding: dict, field: str, relation: str, kind: str ) -> list[tuple[Any, Any]]: if field not in binding: return [(None, None)] return [ (document, node) for document in self.neighbors(person, "HAS_DOCUMENT", "Document") for node in self.neighbors(document, relation, kind) if canonical(self.nodes[node].get("name")) == canonical(binding[field]) ] def witnesses(self, binding: dict, *, require_provenance: bool) -> list[dict]: people = [ key for key, row in self.nodes.items() if key[0] == "Person" and key not in self.conflicts and ( "person" not in binding or canonical(row.get("name")) == canonical(binding["person"]) ) ] academic_required = any( field in binding for field in ["academic_kind", "academic_type", "title", "year"] ) witnesses = [] for person in people: if "start_year" in binding and canonical( self.nodes[person].get("start_year") ) != canonical(binding["start_year"]): continue academics = [] if academic_required: for relation, kind in [("AUTHORED", "Thesis"), ("HAS_WORK", "Work")]: for academic in self.neighbors(person, relation, kind): row = self.nodes[academic] if "academic_kind" in binding and not kind_matches( kind, binding["academic_kind"] ): continue if any( canonical(row.get(field)) != canonical(binding[field]) for field in ["title", "year"] if field in binding ): continue type_field = "thesis_type" if kind == "Thesis" else "work_type" if "academic_type" in binding and canonical( row.get(type_field) ) != canonical(binding["academic_type"]): continue docs = self.neighbors(academic, "EVIDENCED_BY", "Document") academics.extend((academic, doc) for doc in docs) if not docs and not require_provenance: academics.append((academic, None)) else: academics = [(None, None)] choices = [ self.metadata_choices(person, binding, *arguments) for arguments in [ ("keyword", "HAS_KEYWORD", "Keyword"), ("category", "IN_CATEGORY", "Category"), ("document_author", "AUTHORED_DOCUMENT", "DocumentAuthor"), ] ] person_docs = ( self.neighbors(person, "HAS_DOCUMENT", "Document") if ( require_provenance and ( "start_year" in binding or not academic_required and not any( field in binding for field in ["keyword", "category", "document_author"] ) ) ) else [None] ) for ( (academic, academic_document), keyword, category, author, person_document, ) in itertools.product(academics, *choices, person_docs): docs = [ doc for doc in [ academic_document, keyword[0], category[0], author[0], person_document, ] if doc ] if binding.get("same_document") and len(set(docs)) > 1: continue if category[0] and author[0] and category[0] != author[0]: continue if "document_path" in binding and any( self.nodes[doc].get("path") != binding["document_path"] for doc in docs ): continue witnesses.append( { "person": person, "academic": academic, "academic_document": academic_document, "person_document": person_document, "keyword": keyword[1], "keyword_document": keyword[0], "category": category[1], "category_document": category[0], "document_author": author[1], "author_document": author[0], } ) return witnesses def pattern_path(self, witness: dict, pattern: dict) -> list[dict] | None: kinds, relations = pattern.get("node_kinds", []), pattern.get("relationships", []) if not kinds or len(relations) != len(kinds) - 1 or not kind_matches("Person", kinds[0]): return None target_roles = { "Keyword": "keyword", "Category": "category", "DocumentAuthor": "document_author", "Thesis": "academic", "Work": "academic", "Academic": "academic", "Thesis|Work": "academic", } allowed = {key for key in witness.values() if key is not None} required_target = witness.get(target_roles.get(kinds[-1], "")) def visit(key, index, path): if index == len(relations): return path if required_target is None or key == required_target else None for next_key in self.neighbors(key, relations[index], kinds[index + 1]): if next_key not in allowed: continue result = visit(next_key, index + 1, path + [next_key]) if result: return result return None found = visit(witness["person"], 0, [witness["person"]]) return [self.nodes[key] for key in found] if found else None def implicit_patterns(binding: dict, index: int) -> list[dict]: patterns = [] if "start_year" in binding or not any( field in binding for field in [ "academic_kind", "academic_type", "title", "year", "keyword", "category", "document_author", ] ): patterns.append( { "binding_index": index, "node_kinds": ["Person", "Document"], "relationships": ["HAS_DOCUMENT"], } ) if any(field in binding for field in ["academic_kind", "academic_type", "title", "year"]): kind = binding.get("academic_kind", "Academic") relationships = ( ["AUTHORED"] if kind == "Thesis" else ["HAS_WORK"] if kind == "Work" else ["AUTHORED|HAS_WORK"] ) patterns.append( { "binding_index": index, "node_kinds": ["Person", kind, "Document"], "relationships": relationships + ["EVIDENCED_BY"], } ) for field, kind, relation in [ ("keyword", "Keyword", "HAS_KEYWORD"), ("category", "Category", "IN_CATEGORY"), ("document_author", "DocumentAuthor", "AUTHORED_DOCUMENT"), ]: if field in binding: patterns.append( { "binding_index": index, "node_kinds": ["Person", "Document", kind], "relationships": ["HAS_DOCUMENT", relation], } ) return patterns def legacy_bindings(question: dict) -> list[dict]: """Best-effort semantic conversion of schema2 annotations, never result data.""" graph = question.get("graph", {}) entities = defaultdict(list) for row in graph.get("expected_entities", []): entities[row.get("type")].append(row.get("value")) people = entities["person"] or [ row["value"] for row in graph.get("start_entities", []) if row["type"] == "person" ] people = people or [None] years = entities["year"] properties = graph.get("expected_properties", []) start_year = any(prop == "Person.start_year" for prop in properties) if not years and any(".year" in prop for prop in properties) and not start_year: years = re.findall(r"\b20\d{2}\b", question.get("question", "")) bindings = [] for person in people: common = {"person": person} if person else {} if len(years) == 1: common["start_year" if start_year else "year"] = years[0] for field, entity_type in [ ("keyword", "keyword"), ("category", "category"), ("document_author", "document_author"), ]: if len(entities[entity_type]) == 1: common[field] = entities[entity_type][0] titles = [("Thesis", value) for value in entities["thesis_title"]] + [ ("Work", value) for value in entities["work_title"] ] if titles: bindings.extend( {**common, "academic_kind": kind, "title": title} for kind, title in titles ) elif len(entities["category"]) > 1: bindings.extend({**common, "category": value} for value in entities["category"]) elif any(key != "person" for key in common): bindings.append(common) return bindings def positive_answer_facts(result: dict) -> list[dict]: facts = [] for row in result.get("facts", []): if not isinstance(row, dict): facts.append(row) continue kind = row.get("fact_type", "") if ( kind == "academic_year_comparison" and row.get("same_academic_year") is False and not row.get("shared_years") ): continue if ( kind == "start_year_comparison" and row.get("same_start_year") is False and row.get("shared_start_year") is None ): continue facts.append(row) return facts def evaluate_contract(question: dict, result: dict) -> dict: graph = question.get("graph", {}) schema = graph.get("schema_version", 2) negative = bool(graph.get("expected_no_answer") or question.get("should_answer") is False) expected_status = graph.get("expected_status") if negative: status = result.get("status") no_positive = not positive_answer_facts(result) and not result.get("paths") aggregation = result.get("aggregation") or {} if schema >= 3 and expected_status is None: expected_status = "no_answer" status_ok = ( status == expected_status if expected_status else status in {"no_answer", "ambiguous"} ) reason = result.get("reason") ok = bool( status_ok and isinstance(reason, str) and reason.strip() and no_positive and not aggregation.get("count") ) return { "answerability_ok": ok, "bindings_ok": True, "binding_score": 1.0, "paths_ok": True, "path_score": 1.0, "binding_results": [], "path_results": [], "expected_no_answer": True, "actual_status": status, } status_ok = ( result.get("status") == "ok" if schema >= 3 else result.get("status") not in {"no_answer", "ambiguous", "error"} ) bindings = graph.get("expected_bindings") if bindings is None: bindings = legacy_bindings(question) empty_count_answer = ( graph.get("expected_count") == 0 and graph.get("expected_entity_set") == [] and (result.get("aggregation") or {}).get("count") == 0 and (result.get("aggregation") or {}).get("complete") is True ) if schema >= 3 and not bindings and not empty_count_answer: status_ok = False evidence = EvidenceGraph(result) explicit_patterns = graph.get("expected_path_patterns", []) source_choices = graph.get("expected_evidence_documents", []) binding_results, path_results = [], [] for index, binding in enumerate(bindings): candidates = evidence.witnesses(binding, require_provenance=schema >= 3) patterns = [p for p in explicit_patterns if p.get("binding_index") == index] if not patterns and schema >= 3: patterns = implicit_patterns(binding, index) winner = None for candidate in candidates: checks = [evidence.pattern_path(candidate, pattern) for pattern in patterns] source_ok = True for choice in source_choices: if choice.get("binding_index") != index: continue document = evidence.nodes.get(candidate.get(choice.get("role")), {}) if not any( document.get("path") == option.get("document_path") and document.get("source_url") == option.get("source_url") for option in choice.get("any_of", []) ): source_ok = False break if source_ok and all(check is not None for check in checks): winner = checks break binding_results.append( {"binding": binding, "matched": bool(candidates) and winner is not None} ) for pattern_index, pattern in enumerate(patterns): path_results.append( { "pattern": pattern, "matched": winner is not None, "witness": winner[pattern_index] if winner is not None else [], } ) bindings_ok = all(row["matched"] for row in binding_results) paths_ok = all(row["matched"] for row in path_results) # A malformed pattern referencing a nonexistent binding must not vanish. if any( not isinstance(p.get("binding_index"), int) or p["binding_index"] not in range(len(bindings)) for p in explicit_patterns ): paths_ok = False if any( not isinstance(p.get("binding_index"), int) or p["binding_index"] not in range(len(bindings)) or p.get("role") not in { "academic_document", "keyword_document", "category_document", "author_document", "person_document", } or not p.get("any_of") for p in source_choices ): bindings_ok = False return { "answerability_ok": status_ok, "bindings_ok": bindings_ok, "binding_score": ( sum(row["matched"] for row in binding_results) / len(binding_results) if binding_results else 1.0 ), "paths_ok": paths_ok, "path_score": ( sum(row["matched"] for row in path_results) / len(path_results) if path_results else float(paths_ok) ), "binding_results": binding_results, "path_results": path_results, "expected_no_answer": False, "actual_status": result.get("status", "ok"), }