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