"""Adversarial and long-hop example generation over real graph snapshots.""" from __future__ import annotations import random from collections.abc import Iterator, Sequence from graph.graph import EdgeRecord, GraphStore from graph.real.corpus import ( RealGraph, _competing_paths, _extract_neighborhood, _matching_paths, _query_text, _sample_path, _without_support, relation_phrase, ) from graph.synthetic.generator import ReasoningExample from graph.types import GraphPath, GraphQuery, PathStep, ReasoningStatus ROBUSTNESS_MODES = ( "standard", "hard_negative", "missing_edge", "adversarial_contradiction", "long_hop", ) def _step(edge: EdgeRecord) -> PathStep: return PathStep( edge.source_id, edge.relation_id, edge.target_id, edge.source_type, edge.target_type, edge.attributes, ) def _single_step_paths( graph: GraphStore, start: str, relation: str ) -> tuple[GraphPath, ...]: return tuple( GraphPath((_step(edge),)) for edge in graph.outgoing(start) if edge.relation_id == relation ) def _inject_contradiction( neighbourhood: GraphStore, supporting: Sequence[GraphPath], *, seed: int, ) -> tuple[GraphStore, tuple[GraphPath, ...]] | None: if not supporting or len(supporting[0].steps) != 1: return None gold_step = supporting[0].steps[0] candidates = [ node for node, node_type in neighbourhood.nodes.items() if node != gold_step.target_id and node != gold_step.source_id and node_type == gold_step.target_type ] if not candidates: candidates = [f"adv:{seed}:alternative"] target = random.Random(seed).choice(candidates) nodes = dict(neighbourhood.nodes) nodes.setdefault(target, gold_step.target_type or "entity") node_attributes = dict(neighbourhood.node_attributes) node_attributes.setdefault(target, {"label": "adversarial alternative"}) relation_attributes = { key: dict(value) for key, value in neighbourhood.relation_attributes.items() } relation_attributes.setdefault(gold_step.relation_id, {})["functional"] = "true" injected = EdgeRecord( gold_step.source_id, gold_step.relation_id, target, gold_step.source_type or nodes[gold_step.source_id], gold_step.target_type or nodes[target], gold_step.relation_id, { "evidence_kind": "adversarial_perturbation", "perturbation": "functional_relation_alternative_value", }, ) edges = (*neighbourhood.edges, injected) graph = GraphStore( nodes, edges, graph_version=neighbourhood.graph_version, node_attributes=node_attributes, relation_attributes=relation_attributes, ) conflicts = _single_step_paths(graph, gold_step.source_id, gold_step.relation_id) if len({path.endpoint_id for path in conflicts}) < 2: return None return graph, conflicts def generate_robust_examples( real_graph: RealGraph, *, count: int, seed: int = 0, mode: str, min_hops: int = 1, max_hops: int = 4, max_branch: int = 12, max_nodes: int = 80, ) -> Iterator[ReasoningExample]: """Yield deterministic examples for one explicit robustness mode. Contradictions are derived perturbations: they are not claims that the raw source contains two incompatible facts. The generated graph records that distinction on the injected edge and the example's ``robustness_mode``. """ if mode not in ROBUSTNESS_MODES: raise ValueError(f"unknown robustness mode {mode!r}") if count < 0: raise ValueError("count must be non-negative") if min_hops < 1 or min_hops > max_hops: raise ValueError("hop bounds must satisfy 1 <= min_hops <= max_hops") if mode == "long_hop" and min_hops < 5: min_hops = 5 max_hops = max(max_hops, 8) if mode == "adversarial_contradiction": min_hops = max_hops = 1 graph = real_graph.graph starts = [node for node in graph.nodes if graph.outgoing(node)] if not starts and count: raise ValueError("real graph contains no nodes with outgoing edges") rng = random.Random(seed) produced = attempts = 0 max_attempts = max(100, count * 150) while produced < count and attempts < max_attempts: attempts += 1 start = rng.choice(starts) sampled = _sample_path(graph, start, rng, min_hops, max_hops) if sampled is None: continue composition = sampled.relation_sequence neighbourhood = _extract_neighborhood( graph, start, sampled, rng, max_hops=max_hops, max_branch=max_branch, max_nodes=max_nodes, ) supporting = _matching_paths(neighbourhood, start, composition) if not supporting: continue active_graph = neighbourhood active_supporting = supporting answers = tuple( dict.fromkeys(path.endpoint_id for path in supporting if path.endpoint_id) ) conflicting: tuple[GraphPath, ...] = () status = ReasoningStatus.SUPPORTED if mode == "missing_edge": active_graph = _without_support( neighbourhood, supporting, blocked_prefix=(start, composition[0]), ) active_supporting = () answers = () status = ReasoningStatus.INSUFFICIENT_EVIDENCE elif mode == "adversarial_contradiction": injected = _inject_contradiction( neighbourhood, supporting, seed=seed + produced ) if injected is None: continue active_graph, conflicting = injected active_supporting = () answers = () status = ReasoningStatus.CONTRADICTED competing = _competing_paths( active_graph, start, len(composition), composition, limit=64, ) negative_answers = tuple( endpoint for endpoint in dict.fromkeys(path.endpoint_id for path in competing) if endpoint is not None and endpoint not in answers ) if mode == "hard_negative" and not negative_answers: continue phrases = [ real_graph.relation_phrases.get(relation, relation_phrase(relation)) for relation in composition ] label = real_graph.node_labels.get(start, start) target_type = sampled.steps[-1].target_type or "entity" query = GraphQuery( text=_query_text(label, target_type, phrases, len(composition), rng), start_entities=(start,), start_entity_ids=(start,), max_hops=max_hops, min_hops=1, return_paths=True, verify=True, operation="find_entities", answer_mode="entity", ) yield ReasoningExample( example_id=f"{real_graph.source.source_id}-{mode}-{seed}-{produced}", graph=active_graph.to_example(), query=query, answers=answers, negative_answers=negative_answers, supporting_paths=active_supporting, conflicting_paths=conflicting, status=status, difficulty=len(composition), seed=seed + produced, generator_version="real-corpus-robustness-v1", task_type=query.operation, intent_id=f"{real_graph.source.domain}:{mode}", composition=composition, competing_paths=competing, stage_name=f"real_{mode}", robustness_mode=mode, ) produced += 1 if produced < count: raise RuntimeError( f"generated only {produced}/{count} {mode} examples from " f"{real_graph.source.source_id} after {attempts} attempts" )