"""Intent-driven task construction: the query states *what*, never *how*. Why this module exists ---------------------- The previous generator wrote the gold relation sequence into ``GraphQuery.required_relation_sequence`` and into the question text, then computed the gold answer as "every path matching that sequence". The reasoner filtered its own candidate expansion by the same field. The task was therefore "walk the sequence you were given", which exhaustive traversal solves perfectly and which no amount of training can improve. Every metric pinned at 1.0. Here a task is built from a :class:`~graph.synthetic.ontology.QueryIntent`: * the query carries a natural-language paraphrase and the start entity; * the relation composition that answers it is **never** placed in the query; * ``max_hops`` is a stage-level bound shared by every intent at that stage, so path length is not a giveaway either; * gold answers are the endpoints of composition-matching paths, while endpoints of *other* same-length paths from the same start become hard negatives. A search that does not understand the query therefore cannot choose among the competing endpoints, which is exactly the discrimination PLAN sections 12.5 and 16 ask the model to learn. """ from __future__ import annotations import random from collections.abc import Mapping, Sequence from dataclasses import dataclass from graph.graph import EdgeRecord, GraphStore from graph.synthetic.ontology import ( RELATION_BY_NAME, QueryIntent, ) from graph.types import GraphPath, GraphQuery, PathStep, ReasoningStatus #: Stage-level hop budgets. Every intent at a stage shares the same bound so #: ``max_hops`` never reveals the composition length. STAGE_HOP_BUDGET: Mapping[int, int] = { 0: 2, 1: 2, 2: 3, 3: 4, 4: 4, 5: 5, 6: 5, 7: 5, 8: 5, 9: 5, 10: 5, 11: 5, 12: 5, 13: 6, 14: 6, } @dataclass(frozen=True) class TaskBuild: """The outcome of constructing one task over a (possibly mutated) graph.""" graph: GraphStore query: GraphQuery answers: tuple[str, ...] negative_answers: tuple[str, ...] supporting_paths: tuple[GraphPath, ...] competing_paths: tuple[GraphPath, ...] conflicting_paths: tuple[GraphPath, ...] status: ReasoningStatus intent_id: str composition: tuple[str, ...] start_id: str def paths_matching_composition( graph: GraphStore, start_id: str, composition: Sequence[str] ) -> tuple[GraphPath, ...]: """Enumerate simple paths from ``start_id`` that walk ``composition`` exactly.""" results: list[GraphPath] = [] def walk( node: str, index: int, steps: tuple[PathStep, ...], visited: frozenset[str] ) -> None: if index == len(composition): results.append(GraphPath(steps=steps)) return for edge in graph.outgoing(node): if edge.relation_id != composition[index] or edge.target_id in visited: continue walk( edge.target_id, index + 1, ( *steps, PathStep( source_id=edge.source_id, relation_id=edge.relation_id, target_id=edge.target_id, source_type=edge.source_type, target_type=edge.target_type, edge_attributes=edge.attributes, ), ), visited | {edge.target_id}, ) walk(start_id, 0, (), frozenset({start_id})) return tuple(results) def competing_paths( graph: GraphStore, start_id: str, length: int, composition: Sequence[str], *, limit: int = 64, ) -> tuple[GraphPath, ...]: """Same-length paths from the same start that do **not** walk ``composition``. These are the hard negatives: identical hop count, identical origin, and often an endpoint of the same entity type. Only the relation sequence differs, so discriminating them requires reading the query. """ found: list[GraphPath] = [] for path in graph.iter_paths((start_id,), length): if path.hop_count != length: continue if tuple(path.relation_sequence) == tuple(composition): continue found.append(path) if len(found) >= limit: break return tuple(found) def plant_composition( graph: GraphStore, start_id: str, composition: Sequence[str], type_lookup: Mapping[str, str], relation_lookup: Mapping[str, str], rng: random.Random, ) -> GraphStore: """Add the minimum edges needed for ``composition`` to be walkable from ``start_id``. Planting keeps every generated task answerable without weakening it: the planted path is one of many same-length routes out of ``start_id``, so the model still has to pick the right one. """ nodes_by_type: dict[str, list[str]] = {} for node_id, node_type in graph.nodes.items(): nodes_by_type.setdefault(node_type, []).append(node_id) edges = list(graph.edges) current = start_id visited = {start_id} for canonical in composition: spec = RELATION_BY_NAME[canonical] # Use the builder's own canonical -> graph-local mapping. Inferring it # from edge type signatures is ambiguous whenever two relations share a # domain/range pair (``member_of`` and ``manages`` are both # Person -> Team), and on a renamed graph that ambiguity silently planted # edges under the canonical name that no query could then match. relation_id = relation_lookup[canonical] target_type = type_lookup[spec.range] existing = [ edge.target_id for edge in graph.outgoing(current) if edge.relation_id == relation_id and edge.target_id not in visited ] if existing: current = existing[0] visited.add(current) continue candidates = [ node for node in nodes_by_type.get(target_type, ()) if node not in visited ] if not candidates: raise RuntimeError( f"cannot plant {canonical!r}: no unused node of type {target_type!r} exists" ) target = candidates[rng.randrange(len(candidates))] edges.append( EdgeRecord( source_id=current, relation_id=relation_id, target_id=target, source_type=graph.nodes[current], target_type=graph.nodes[target], relation_name=relation_id, attributes={"weight": "1"}, ) ) current = target visited.add(current) return GraphStore( graph.nodes, edges, graph_version=graph.graph_version, node_attributes=graph.node_attributes, relation_attributes=graph.relation_attributes, ) def candidate_starts( graph: GraphStore, intent: QueryIntent, type_lookup: Mapping[str, str] ) -> tuple[str, ...]: """Nodes typed as the composition's domain, ordered for determinism.""" domain = type_lookup[RELATION_BY_NAME[intent.composition[0]].domain] return tuple(node for node, node_type in graph.nodes.items() if node_type == domain) def build_task( graph: GraphStore, intent: QueryIntent, *, stage: int, rng: random.Random, type_lookup: Mapping[str, str], relation_lookup: Mapping[str, str], answer_mode: str = "entity", held_out_templates: bool = False, variant: str = "standard", ) -> TaskBuild: """Construct one well-posed task for ``intent`` over ``graph``. ``variant`` selects the reasoning scenario: ``standard`` answers the intent, ``missing_edge`` removes the supporting evidence so the correct response is ``INSUFFICIENT_EVIDENCE``, and ``contradiction`` introduces a second conflicting value for the same subject/relation claim. """ composition = tuple(relation_lookup[name] for name in intent.composition) starts = candidate_starts(graph, intent, type_lookup) if not starts: raise RuntimeError( f"graph has no node of the domain type required by intent {intent.intent_id!r}" ) start_id = _select_start(graph, starts, composition, rng) supporting = paths_matching_composition(graph, start_id, composition) if not supporting: graph = plant_composition( graph, start_id, intent.composition, type_lookup, relation_lookup, rng ) supporting = paths_matching_composition(graph, start_id, composition) if not supporting: raise RuntimeError( f"planting failed for intent {intent.intent_id!r} from {start_id!r}" ) conflicts: tuple[GraphPath, ...] = () status = ReasoningStatus.SUPPORTED if variant == "missing_edge": graph = _remove_final_edges(graph, supporting) supporting = paths_matching_composition(graph, start_id, composition) if not supporting: status = ReasoningStatus.INSUFFICIENT_EVIDENCE elif variant == "contradiction": graph, conflicts = _introduce_contradiction(graph, supporting, rng) supporting = paths_matching_composition(graph, start_id, composition) if conflicts: status = ReasoningStatus.CONTRADICTED competing = competing_paths(graph, start_id, len(composition), composition) answers = ( tuple( dict.fromkeys(path.endpoint_id for path in supporting if path.endpoint_id) ) if status in {ReasoningStatus.SUPPORTED, ReasoningStatus.CONTRADICTED} else () ) negatives = tuple( dict.fromkeys( path.endpoint_id for path in competing if path.endpoint_id and path.endpoint_id not in set(answers) ) ) templates = intent.held_out_templates if held_out_templates else intent.templates if not templates: templates = intent.templates text = templates[rng.randrange(len(templates))].format(start=start_id) budget = STAGE_HOP_BUDGET.get(stage, 5) max_hops = max(budget, len(composition)) query = GraphQuery( text=text, start_entities=(start_id,), start_entity_ids=(start_id,), # Deliberately empty: the composition, its length, and the answer type # are all withheld. They are the answer, not the question. allowed_relations=(), required_relations=(), required_relation_sequence=(), required_types=(), min_hops=1, max_hops=max_hops, exact_hops=None, return_paths=True, verify=True, operation="find_entities" if answer_mode == "entity" else answer_mode, answer_mode=answer_mode, target_entity_ids=_target_ids(answer_mode, answers, negatives, status, rng), ) if answer_mode == "boolean": status = _boolean_status(status, query.target_entity_ids, answers) return TaskBuild( graph=graph, query=query, answers=answers, negative_answers=negatives[:16], supporting_paths=supporting, competing_paths=competing[:16], conflicting_paths=conflicts, status=status, intent_id=intent.intent_id, composition=tuple(intent.composition), start_id=start_id, ) def _select_start( graph: GraphStore, starts: Sequence[str], composition: Sequence[str], rng: random.Random, ) -> str: """Prefer a start that already walks the composition and has competitors. A start with competing same-length routes makes the task discriminative; one without them would be answerable by hop-count alone. """ walkable = [ node for node in starts if paths_matching_composition(graph, node, composition) ] contested = [ node for node in walkable if competing_paths(graph, node, len(composition), composition, limit=2) ] pool = contested or walkable or list(starts) return pool[rng.randrange(len(pool))] def _remove_final_edges( graph: GraphStore, supporting: Sequence[GraphPath] ) -> GraphStore: """Delete the last edge of every supporting path, removing the evidence.""" doomed = { (path.steps[-1].source_id, path.steps[-1].relation_id, path.steps[-1].target_id) for path in supporting if path.steps } edges = [ edge for edge in graph.edges if (edge.source_id, edge.relation_id, edge.target_id) not in doomed ] return GraphStore( graph.nodes, edges, graph_version=graph.graph_version, node_attributes=graph.node_attributes, relation_attributes=graph.relation_attributes, ) def _introduce_contradiction( graph: GraphStore, supporting: Sequence[GraphPath], rng: random.Random ) -> tuple[GraphStore, tuple[GraphPath, ...]]: """Add a second, conflicting target for the first claim on a supporting path. A contradiction is a *single* subject/relation asserting two different values. Two different multi-hop routes to different endpoints are competing explanations, not a contradiction, and are handled separately. """ if not supporting or not supporting[0].steps: return graph, () first = supporting[0].steps[0] same_type = [ node for node, node_type in graph.nodes.items() if node_type == first.target_type and node not in {first.source_id, first.target_id} ] if not same_type: return graph, () rival = same_type[rng.randrange(len(same_type))] if graph.has_edge(first.source_id, first.relation_id, rival): return graph, () edges = [ *graph.edges, EdgeRecord( source_id=first.source_id, relation_id=first.relation_id, target_id=rival, source_type=graph.nodes[first.source_id], target_type=graph.nodes[rival], relation_name=first.relation_id, attributes={"contradictory": "true"}, ), ] mutated = GraphStore( graph.nodes, edges, graph_version=graph.graph_version, node_attributes=graph.node_attributes, relation_attributes=graph.relation_attributes, ) claims = tuple( GraphPath( steps=( PathStep( source_id=first.source_id, relation_id=first.relation_id, target_id=target, source_type=graph.nodes[first.source_id], target_type=graph.nodes[target], ), ) ) for target in (first.target_id, rival) ) return mutated, claims def _target_ids( answer_mode: str, answers: Sequence[str], negatives: Sequence[str], status: ReasoningStatus, rng: random.Random, ) -> tuple[str, ...]: """Boolean tasks need a concrete claim to judge; other modes need none.""" if answer_mode != "boolean": return () truthful = bool(answers) and rng.random() < 0.5 if truthful: return (answers[rng.randrange(len(answers))],) if negatives: return (negatives[rng.randrange(len(negatives))],) return (answers[0],) if answers else () def _boolean_status( status: ReasoningStatus, targets: Sequence[str], answers: Sequence[str] ) -> ReasoningStatus: if status is not ReasoningStatus.SUPPORTED: return status if not targets: return status return ( ReasoningStatus.SUPPORTED if set(targets) & set(answers) else ReasoningStatus.UNSUPPORTED )