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