| """Causal, scientific, and commonsense graph sources. |
| |
| The loaders in this module intentionally keep evidence type in edge attributes |
| and keep it visible in the serialized record metadata. A mined statement, |
| an ontology relationship, a temporal link, and a hand-authored structural |
| causal model are useful for different training tasks and must not be silently |
| treated as interchangeable evidence. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import bz2 |
| import csv |
| import gzip |
| import json |
| import random |
| import re |
| import zipfile |
| from collections import Counter |
| from collections.abc import Iterable, Iterator, Mapping, Sequence |
| from dataclasses import dataclass |
| from pathlib import Path |
| from urllib.parse import unquote |
| from xml.etree import ElementTree |
|
|
| from graph.graph import EdgeRecord, GraphStore |
| from graph.real.corpus import ( |
| CorpusStats, |
| RealGraph, |
| SourceMetadata, |
| _competing_paths, |
| _extract_neighborhood, |
| _matching_paths, |
| _without_support, |
| relation_phrase, |
| serialize_example, |
| ) |
| from graph.synthetic.generator import ReasoningExample |
| from graph.types import GraphPath, GraphQuery, PathStep, ReasoningStatus |
|
|
|
|
| @dataclass(frozen=True) |
| class CausalExample: |
| """Trainer-native example plus causal evidence metadata.""" |
|
|
| example: ReasoningExample |
| metadata: Mapping[str, str] |
|
|
|
|
| def _slug(value: str, *, limit: int = 120) -> str: |
| value = unquote(value).strip().lower().replace("_", " ") |
| value = re.sub(r"\s+", " ", value) |
| return value[:limit] or "unknown" |
|
|
|
|
| def _node_id(prefix: str, label: str) -> str: |
| return f"{prefix}:{_slug(label).replace(' ', '_')}" |
|
|
|
|
| def _make_graph( |
| source: SourceMetadata, |
| nodes: Mapping[str, str], |
| edges: Iterable[EdgeRecord], |
| *, |
| labels: Mapping[str, str], |
| node_attributes: Mapping[str, Mapping[str, str]], |
| relation_ids: Iterable[str], |
| ) -> RealGraph: |
| relation_ids = tuple(sorted(set(relation_ids))) |
| edge_list = tuple(edges) |
| relation_attributes = { |
| relation: { |
| "functional": "false", |
| "description": relation_phrase(relation), |
| } |
| for relation in relation_ids |
| } |
| return RealGraph( |
| graph=GraphStore( |
| nodes, |
| edge_list, |
| graph_version=f"{source.source_id}@{source.version}", |
| node_attributes=node_attributes, |
| relation_attributes=relation_attributes, |
| ), |
| node_labels=labels, |
| relation_phrases={ |
| relation: relation_phrase(relation) for relation in relation_ids |
| }, |
| source=source, |
| ) |
|
|
|
|
| _CONCEPTNET_RELATIONS = { |
| "/r/Causes": "causes", |
| "/r/HasPrerequisite": "has_prerequisite", |
| "/r/UsedFor": "used_for", |
| "/r/CapableOf": "capable_of", |
| "/r/PartOf": "part_of", |
| "/r/IsA": "is_a", |
| "/r/RelatedTo": "associated_with", |
| "/r/Desires": "desires", |
| "/r/NotDesires": "not_desires", |
| } |
|
|
|
|
| def _conceptnet_label(uri: str) -> str | None: |
| parts = uri.split("/") |
| if len(parts) < 4 or parts[1] != "c" or parts[2] != "en": |
| return None |
| return unquote(parts[3]).replace("_", " ").strip() or None |
|
|
|
|
| def load_conceptnet( |
| path: str | Path, |
| source: SourceMetadata, |
| *, |
| max_edges: int = 350_000, |
| ) -> RealGraph: |
| """Load a bounded English ConceptNet assertion slice.""" |
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| seen: set[tuple[str, str, str]] = set() |
| with gzip.open(path, "rt", encoding="utf-8", newline="") as handle: |
| reader = csv.reader(handle, delimiter="\t") |
| for row in reader: |
| if not row or row[0].startswith("#") or len(row) < 5: |
| continue |
| relation = _CONCEPTNET_RELATIONS.get(row[1]) |
| if relation is None: |
| continue |
| start = _conceptnet_label(row[2]) |
| end = _conceptnet_label(row[3]) |
| if start is None or end is None or start == end: |
| continue |
| source_id = _node_id("conceptnet", start) |
| target_id = _node_id("conceptnet", end) |
| key = (source_id, relation, target_id) |
| if key in seen: |
| continue |
| seen.add(key) |
| nodes[source_id] = "concept" |
| nodes[target_id] = "concept" |
| labels[source_id] = start |
| labels[target_id] = end |
| if len(row) > 4: |
| try: |
| data = json.loads(row[4]) |
| except json.JSONDecodeError: |
| data = {} |
| else: |
| data = {} |
| attributes.setdefault(source_id, {}) |
| attributes.setdefault(target_id, {}) |
| edge_attributes = { |
| "evidence_kind": "commonsense_assertion", |
| "dataset": str(data.get("dataset", "ConceptNet")), |
| "weight": str(data.get("weight", "")), |
| } |
| edges.append( |
| EdgeRecord( |
| source_id, |
| relation, |
| target_id, |
| "concept", |
| "concept", |
| relation_phrase(relation), |
| edge_attributes, |
| ) |
| ) |
| if len(edges) >= max_edges: |
| break |
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids={edge.relation_id for edge in edges}, |
| ) |
|
|
|
|
| def load_causenet( |
| path: str | Path, |
| source: SourceMetadata, |
| *, |
| max_edges: int = 100_000, |
| ) -> RealGraph: |
| """Load CauseNet causal assertions without copying source sentences.""" |
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| seen: set[tuple[str, str, str]] = set() |
| with bz2.open(path, "rt", encoding="utf-8") as handle: |
| for line in handle: |
| if not line.strip(): |
| continue |
| record = json.loads(line) |
| relation = record.get("causal_relation", {}) |
| cause = str(relation.get("cause", {}).get("concept", "")).strip() |
| effect = str(relation.get("effect", {}).get("concept", "")).strip() |
| if not cause or not effect or cause == effect: |
| continue |
| source_id = _node_id("causenet", cause) |
| target_id = _node_id("causenet", effect) |
| key = (source_id, "causes", target_id) |
| if key in seen: |
| continue |
| seen.add(key) |
| nodes[source_id] = "causal_concept" |
| nodes[target_id] = "causal_concept" |
| labels[source_id] = cause.replace("_", " ") |
| labels[target_id] = effect.replace("_", " ") |
| attributes.setdefault(source_id, {}) |
| attributes.setdefault(target_id, {}) |
| sources = record.get("sources", ()) |
| source_types = sorted( |
| { |
| str(item.get("type", "unknown")) |
| for item in sources |
| if isinstance(item, Mapping) |
| } |
| ) |
| edges.append( |
| EdgeRecord( |
| source_id, |
| "causes", |
| target_id, |
| "causal_concept", |
| "causal_concept", |
| relation_phrase("causes"), |
| { |
| "evidence_kind": "mined_causal_assertion", |
| "source_types": ",".join(source_types), |
| "confidence": "0.83_estimated_precision", |
| }, |
| ) |
| ) |
| if len(edges) >= max_edges: |
| break |
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids={"causes"}, |
| ) |
|
|
|
|
| _OBO_RELATIONS = { |
| "is_a": "is_a", |
| "part_of": "part_of", |
| "regulates": "regulates", |
| "positively_regulates": "regulates_positive", |
| "negatively_regulates": "regulates_negative", |
| "has_role": "has_role", |
| } |
|
|
|
|
| def _open_text(path: str | Path): |
| path = Path(path) |
| if path.suffix == ".gz": |
| return gzip.open(path, "rt", encoding="utf-8") |
| return path.open(encoding="utf-8") |
|
|
|
|
| def _obo_terms(path: str | Path) -> Iterator[dict[str, object]]: |
| current: dict[str, object] | None = None |
| with _open_text(path) as handle: |
| for raw_line in handle: |
| line = raw_line.rstrip("\n") |
| if line == "[Term]": |
| if current is not None: |
| yield current |
| current = {"relations": []} |
| continue |
| if current is None or not line or line.startswith("!"): |
| continue |
| if line.startswith("id: "): |
| current["id"] = line[4:].strip() |
| elif line.startswith("name: "): |
| current["name"] = line[6:].strip() |
| elif line.startswith("namespace: "): |
| current["namespace"] = line[11:].strip() |
| elif line.startswith("def: "): |
| current["definition"] = line[5:].split('"', 2)[1] |
| elif line.startswith("is_a: "): |
| target = line[6:].split(" !", 1)[0].strip() |
| cast_relations = current["relations"] |
| assert isinstance(cast_relations, list) |
| cast_relations.append(("is_a", target)) |
| elif line.startswith("relationship: "): |
| bits = line[14:].split() |
| if len(bits) >= 2: |
| cast_relations = current["relations"] |
| assert isinstance(cast_relations, list) |
| cast_relations.append((bits[0], bits[1])) |
| elif line.startswith("is_obsolete: true"): |
| current["obsolete"] = True |
| if current is not None: |
| yield current |
|
|
|
|
| def load_obo_graph( |
| path: str | Path, |
| source: SourceMetadata, |
| *, |
| max_terms: int | None = None, |
| max_edges: int = 500_000, |
| ) -> RealGraph: |
| """Load selected typed relationships from an OBO ontology.""" |
| records = [record for record in _obo_terms(path) if not record.get("obsolete")] |
| if max_terms is not None: |
| records = records[:max_terms] |
| selected = { |
| str(record["id"]) |
| for record in records |
| if record.get("id") and record.get("name") |
| } |
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| for record in records: |
| term_id = str(record["id"]) |
| node_id = f"obo:{term_id}" |
| name = str(record.get("name", term_id)) |
| namespace = str(record.get("namespace", "ontology_term")) |
| nodes[node_id] = namespace |
| labels[node_id] = name |
| attributes[node_id] = { |
| "definition": str(record.get("definition", ""))[:4000], |
| "namespace": namespace, |
| "ontology_id": term_id, |
| } |
| relation_ids: set[str] = set() |
| for record in records: |
| source_term = str(record["id"]) |
| source_id = f"obo:{source_term}" |
| raw_relations = record.get("relations", []) |
| assert isinstance(raw_relations, list) |
| for raw_relation, target_term in raw_relations: |
| relation = _OBO_RELATIONS.get(str(raw_relation), str(raw_relation)) |
| if target_term not in selected: |
| continue |
| target_id = f"obo:{target_term}" |
| if source_id not in nodes or target_id not in nodes: |
| continue |
| relation_ids.add(relation) |
| edges.append( |
| EdgeRecord( |
| source_id, |
| relation, |
| target_id, |
| nodes[source_id], |
| nodes[target_id], |
| relation_phrase(relation), |
| { |
| "evidence_kind": "ontology_assertion", |
| "ontology_id": source.source_id, |
| }, |
| ) |
| ) |
| if len(edges) >= max_edges: |
| break |
| if len(edges) >= max_edges: |
| break |
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids=relation_ids, |
| ) |
|
|
|
|
| def _local_tag(tag: str) -> str: |
| return tag.rsplit("}", 1)[-1] |
|
|
|
|
| def load_causal_timebank( |
| archive: str | Path, |
| source: SourceMetadata, |
| *, |
| max_documents: int | None = None, |
| ) -> RealGraph: |
| """Load event causality and temporal links from Causal-TimeBank XML.""" |
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| relation_ids: set[str] = set() |
| with zipfile.ZipFile(archive) as handle: |
| names = [ |
| name |
| for name in handle.namelist() |
| if name.lower().endswith((".xml", ".tml")) |
| ] |
| for index, name in enumerate(names): |
| if max_documents is not None and index >= max_documents: |
| break |
| try: |
| root = ElementTree.fromstring(handle.read(name)) |
| except ElementTree.ParseError: |
| continue |
| prefix = _slug(name.rsplit("/", 1)[-1].rsplit(".", 1)[0], limit=80) |
| event_labels: dict[str, str] = {} |
| instance_to_event: dict[str, str] = {} |
| links: list[tuple[str, str, str]] = [] |
| for element in root.iter(): |
| tag = _local_tag(element.tag) |
| if tag == "EVENT" and element.attrib.get("eid"): |
| event_id = f"timebank:{prefix}:{element.attrib['eid']}" |
| text = " ".join("".join(element.itertext()).split()) |
| event_labels[event_id] = text or element.attrib["eid"] |
| elif tag == "MAKEINSTANCE": |
| event = element.attrib.get("eventID") |
| instance = element.attrib.get("eiid") |
| if event and instance: |
| instance_to_event[instance] = f"timebank:{prefix}:{event}" |
| elif tag == "CLINK": |
| links.append( |
| ( |
| "causes", |
| element.attrib.get( |
| "causeID", element.attrib.get("eventInstanceID", "") |
| ), |
| element.attrib.get( |
| "effectID", |
| element.attrib.get("relatedToEventInstance", ""), |
| ), |
| ) |
| ) |
| elif tag == "TLINK": |
| relation = element.attrib.get("relType", "").upper() |
| if relation not in {"BEFORE", "AFTER"}: |
| continue |
| left = element.attrib.get("eventInstanceID", "") |
| right = element.attrib.get("relatedToEventInstance", "") |
| if relation == "AFTER": |
| left, right = right, left |
| links.append(("before", left, right)) |
| nodes.update({event: "event" for event in event_labels}) |
| labels.update(event_labels) |
| attributes.update( |
| { |
| event: { |
| "document": name, |
| "evidence_kind": "human_annotated_event", |
| } |
| for event in event_labels |
| } |
| ) |
| for relation, left, right in links: |
| left_id = instance_to_event.get(left, f"timebank:{prefix}:{left}") |
| right_id = instance_to_event.get(right, f"timebank:{prefix}:{right}") |
| if left_id not in nodes or right_id not in nodes or left_id == right_id: |
| continue |
| relation_ids.add(relation) |
| edges.append( |
| EdgeRecord( |
| left_id, |
| relation, |
| right_id, |
| "event", |
| "event", |
| relation_phrase(relation), |
| { |
| "evidence_kind": ( |
| "human_annotated_causal_link" |
| if relation == "causes" |
| else "human_annotated_temporal_link" |
| ), |
| "document": name, |
| }, |
| ) |
| ) |
| return _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids=relation_ids, |
| ) |
|
|
|
|
| def _sample_path( |
| graph: GraphStore, |
| start: str, |
| rng: random.Random, |
| relation_ids: set[str], |
| min_hops: int, |
| max_hops: int, |
| ) -> GraphPath | None: |
| target_hops = rng.randint(min_hops, max_hops) |
| node = start |
| visited = {start} |
| steps: list[PathStep] = [] |
| for _ in range(target_hops): |
| choices = [ |
| edge |
| for edge in graph.outgoing(node) |
| if edge.relation_id in relation_ids and edge.target_id not in visited |
| ] |
| if not choices: |
| return None |
| edge = rng.choice(choices) |
| steps.append( |
| PathStep( |
| edge.source_id, |
| edge.relation_id, |
| edge.target_id, |
| edge.source_type, |
| edge.target_type, |
| edge.attributes, |
| ) |
| ) |
| node = edge.target_id |
| visited.add(node) |
| return GraphPath(tuple(steps)) |
|
|
|
|
| def _causal_query( |
| label: str, |
| target_type: str, |
| phrases: Sequence[str], |
| hops: int, |
| task_type: str, |
| ) -> str: |
| phrase_text = " and ".join(dict.fromkeys(phrases)) |
| if task_type == "intervention": |
| return ( |
| f"Under an intervention that changes {label}, which {target_type.lower()} " |
| f"is predicted to change through the mechanism involving {phrase_text}?" |
| ) |
| if task_type == "counterfactual": |
| return ( |
| f"If {label} had not occurred, which {target_type.lower()} would be less " |
| f"expected under the mechanism involving {phrase_text}?" |
| ) |
| if task_type == "temporal_precedence": |
| return ( |
| f"Which {target_type.lower()} occurs after {label} according to the " |
| "annotated event sequence?" |
| ) |
| if task_type == "ontology_path": |
| return ( |
| f"Which {target_type.lower()} can be reached from {label} through " |
| f"{phrase_text} across {hops} graph step(s)?" |
| ) |
| if task_type == "mechanism_chain": |
| return ( |
| f"Which {target_type.lower()} is the downstream result of a mechanism " |
| f"starting with {label} and involving {phrase_text}?" |
| ) |
| return ( |
| f"Which {target_type.lower()} can be causally traced from {label} through " |
| f"{phrase_text}?" |
| ) |
|
|
|
|
| def _safe_query_label(label: str, composition: Sequence[str]) -> str: |
| """Avoid echoing a raw relation token when it is also an entity label.""" |
| safe = label |
| for relation in composition: |
| safe = re.sub( |
| rf"(?<![A-Za-z0-9_]){re.escape(relation)}(?![A-Za-z0-9_])", |
| "the starting condition", |
| safe, |
| flags=re.IGNORECASE, |
| ) |
| return safe |
|
|
|
|
| def generate_causal_graph_examples( |
| real_graph: RealGraph, |
| *, |
| count: int, |
| seed: int, |
| relation_ids: Iterable[str], |
| min_hops: int = 1, |
| max_hops: int = 3, |
| unsupported_rate: float = 0.2, |
| ) -> Iterator[CausalExample]: |
| """Generate causal/temporal/ontology examples from a typed source graph.""" |
| allowed = set(relation_ids) |
| if not allowed: |
| raise ValueError("relation_ids cannot be empty") |
| rng = random.Random(seed) |
| starts = [ |
| node |
| for node in real_graph.graph.nodes |
| if any(edge.relation_id in allowed for edge in real_graph.graph.outgoing(node)) |
| ] |
| if not starts and count: |
| raise ValueError("graph has no outgoing edges in the requested relation set") |
| produced = 0 |
| attempts = 0 |
| max_attempts = max(100, count * 100) |
| while produced < count and attempts < max_attempts: |
| attempts += 1 |
| start = rng.choice(starts) |
| sampled = _sample_path( |
| real_graph.graph, start, rng, allowed, min_hops, max_hops |
| ) |
| if sampled is None: |
| continue |
| composition = sampled.relation_sequence |
| neighbourhood = _extract_neighborhood( |
| real_graph.graph, |
| start, |
| sampled, |
| rng, |
| max_hops=max_hops, |
| max_branch=12, |
| max_nodes=80, |
| ) |
| supporting = _matching_paths(neighbourhood, start, composition) |
| if not supporting: |
| continue |
| unsupported = rng.random() < unsupported_rate |
| active_graph = ( |
| _without_support( |
| neighbourhood, |
| supporting, |
| blocked_prefix=(start, composition[0]), |
| ) |
| if unsupported |
| else neighbourhood |
| ) |
| first_relation = composition[0] |
| if first_relation == "before": |
| task_type = "temporal_precedence" |
| elif "causes" in composition or first_relation.startswith("regulates"): |
| task_type = rng.choice( |
| ( |
| "cause_identification", |
| "effect_prediction", |
| "intervention", |
| "counterfactual", |
| "mechanism_chain" if len(composition) > 1 else "effect_prediction", |
| ) |
| ) |
| else: |
| task_type = "ontology_path" |
| phrases = [ |
| real_graph.relation_phrases.get(relation, relation_phrase(relation)) |
| for relation in composition |
| ] |
| label = _safe_query_label(real_graph.node_labels.get(start, start), composition) |
| target_type = sampled.steps[-1].target_type or "entity" |
| query = GraphQuery( |
| text=_causal_query( |
| label, target_type, phrases, len(composition), task_type |
| ), |
| 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", |
| ) |
| active_supporting = () if unsupported else supporting |
| answers = ( |
| () |
| if unsupported |
| else tuple( |
| endpoint |
| for endpoint in dict.fromkeys(path.endpoint_id for path in supporting) |
| if endpoint is not None |
| ) |
| ) |
| competing = _competing_paths( |
| active_graph, start, len(composition), composition, limit=32 |
| ) |
| evidence_kind = str( |
| sampled.steps[0].edge_attributes.get("evidence_kind", "graph_assertion") |
| ) |
| metadata = { |
| "task_type": task_type, |
| "evidence_kind": evidence_kind, |
| "evidence_class": real_graph.source.evidence_class, |
| "causal_truth_status": real_graph.source.causal_truth_status, |
| "causal_relation": first_relation, |
| "support": "absent" if unsupported else "present", |
| "negative_kind": "insufficient_evidence" if unsupported else "none", |
| "confidence": str( |
| sampled.steps[0].edge_attributes.get("confidence", "source_defined") |
| ), |
| } |
| example = ReasoningExample( |
| example_id=f"{real_graph.source.source_id}-{seed}-{produced}", |
| graph=active_graph.to_example(), |
| query=query, |
| answers=answers, |
| negative_answers=tuple( |
| endpoint |
| for endpoint in dict.fromkeys(path.endpoint_id for path in competing) |
| if endpoint is not None |
| ), |
| supporting_paths=active_supporting, |
| conflicting_paths=(), |
| status=( |
| ReasoningStatus.INSUFFICIENT_EVIDENCE |
| if unsupported |
| else ReasoningStatus.SUPPORTED |
| ), |
| difficulty=len(composition), |
| seed=seed + produced, |
| generator_version="causal-corpus-v1", |
| task_type=task_type, |
| intent_id=f"{real_graph.source.domain}:{task_type}", |
| composition=composition, |
| competing_paths=competing, |
| stage_name=f"causal_{real_graph.source.domain}", |
| ) |
| yield CausalExample(example, metadata) |
| produced += 1 |
| if produced < count: |
| raise RuntimeError( |
| f"generated only {produced}/{count} examples from {real_graph.source.source_id}" |
| ) |
|
|
|
|
| @dataclass(frozen=True) |
| class SCMScenario: |
| domain: str |
| cause: str |
| mediator: str |
| effect: str |
| confounder: str |
|
|
|
|
| SCENARIOS: tuple[SCMScenario, ...] = ( |
| SCMScenario("health", "smoking", "airway inflammation", "lung cancer risk", "age"), |
| SCMScenario( |
| "health", |
| "sleep deprivation", |
| "slower reaction time", |
| "driving errors", |
| "work schedule", |
| ), |
| SCMScenario( |
| "health", |
| "pathogen exposure", |
| "viral replication", |
| "infection symptoms", |
| "crowding", |
| ), |
| SCMScenario( |
| "health", |
| "vaccination", |
| "immune memory", |
| "severe infection risk", |
| "health access", |
| ), |
| SCMScenario( |
| "economics", |
| "higher interest rates", |
| "higher borrowing cost", |
| "lower business investment", |
| "inflation expectations", |
| ), |
| SCMScenario( |
| "economics", |
| "job loss", |
| "lower household income", |
| "lower discretionary spending", |
| "regional industry", |
| ), |
| SCMScenario( |
| "economics", |
| "skills training", |
| "worker productivity", |
| "higher wages", |
| "prior education", |
| ), |
| SCMScenario( |
| "economics", |
| "supply disruption", |
| "lower available inventory", |
| "higher prices", |
| "demand growth", |
| ), |
| SCMScenario( |
| "climate", |
| "greenhouse gas emissions", |
| "radiative forcing", |
| "higher global temperature", |
| "industrialization", |
| ), |
| SCMScenario( |
| "climate", |
| "deforestation", |
| "less root stabilization", |
| "greater soil erosion", |
| "heavy rainfall", |
| ), |
| SCMScenario( |
| "climate", |
| "drought", |
| "lower soil moisture", |
| "lower crop yield", |
| "irrigation access", |
| ), |
| SCMScenario( |
| "climate", |
| "urban tree cover", |
| "more shade and evapotranspiration", |
| "lower local heat", |
| "building density", |
| ), |
| SCMScenario( |
| "education", |
| "deliberate practice", |
| "retrieval fluency", |
| "better test performance", |
| "prior preparation", |
| ), |
| SCMScenario( |
| "education", |
| "timely feedback", |
| "error correction", |
| "improved revision quality", |
| "teacher attention", |
| ), |
| SCMScenario( |
| "education", |
| "smaller class size", |
| "more individual interaction", |
| "higher engagement", |
| "school funding", |
| ), |
| SCMScenario( |
| "education", |
| "background knowledge", |
| "faster comprehension", |
| "better problem solving", |
| "reading exposure", |
| ), |
| SCMScenario( |
| "software", |
| "request caching", |
| "fewer database reads", |
| "lower response latency", |
| "traffic pattern", |
| ), |
| SCMScenario( |
| "software", |
| "automated tests", |
| "earlier defect detection", |
| "fewer regressions", |
| "engineering time", |
| ), |
| SCMScenario( |
| "software", |
| "dependency upgrade", |
| "changed transitive API", |
| "build failure", |
| "release timing", |
| ), |
| SCMScenario( |
| "software", |
| "traffic increase", |
| "queue growth", |
| "higher service latency", |
| "capacity planning", |
| ), |
| SCMScenario( |
| "engineering", |
| "lubrication", |
| "lower contact friction", |
| "less component wear", |
| "operating temperature", |
| ), |
| SCMScenario( |
| "engineering", |
| "excessive load", |
| "higher material stress", |
| "structural failure risk", |
| "material defect", |
| ), |
| SCMScenario( |
| "engineering", |
| "battery heating", |
| "higher reaction rate", |
| "shorter battery life", |
| "charge cycle", |
| ), |
| SCMScenario( |
| "engineering", |
| "thermal expansion", |
| "increased joint displacement", |
| "seal leakage", |
| "manufacturing tolerance", |
| ), |
| SCMScenario( |
| "biology", |
| "insulin signaling", |
| "greater glucose uptake", |
| "lower blood glucose", |
| "body composition", |
| ), |
| SCMScenario( |
| "biology", |
| "predator abundance", |
| "greater prey mortality", |
| "smaller prey population", |
| "habitat quality", |
| ), |
| SCMScenario( |
| "biology", |
| "gene transcription", |
| "more messenger RNA", |
| "higher protein production", |
| "cell state", |
| ), |
| SCMScenario( |
| "biology", |
| "antibiotic exposure", |
| "bacterial growth suppression", |
| "lower bacterial load", |
| "initial dose", |
| ), |
| SCMScenario( |
| "everyday", |
| "heavy rainfall", |
| "rapid water accumulation", |
| "street flooding", |
| "drainage capacity", |
| ), |
| SCMScenario( |
| "everyday", |
| "regular exercise", |
| "improved cardiovascular capacity", |
| "greater endurance", |
| "baseline health", |
| ), |
| SCMScenario( |
| "everyday", |
| "traffic congestion", |
| "slower vehicle flow", |
| "longer travel time", |
| "road capacity", |
| ), |
| SCMScenario( |
| "social", |
| "misinformation exposure", |
| "false belief adoption", |
| "poor decision quality", |
| "prior trust", |
| ), |
| SCMScenario( |
| "physics", |
| "greater net force", |
| "higher acceleration", |
| "greater velocity change", |
| "object mass", |
| ), |
| SCMScenario( |
| "physics", |
| "higher temperature", |
| "greater particle motion", |
| "higher pressure in a fixed volume", |
| "container volume", |
| ), |
| SCMScenario( |
| "physics", |
| "increased resistance", |
| "greater electrical energy loss", |
| "lower circuit power delivery", |
| "source voltage", |
| ), |
| SCMScenario( |
| "astronomy", |
| "stellar core contraction", |
| "higher core temperature", |
| "heavier element fusion", |
| "stellar mass", |
| ), |
| SCMScenario( |
| "astronomy", |
| "planetary albedo", |
| "reflected stellar radiation", |
| "lower surface heating", |
| "atmospheric composition", |
| ), |
| SCMScenario( |
| "earth_science", |
| "tectonic plate motion", |
| "stress accumulation", |
| "earthquake rupture", |
| "fault geometry", |
| ), |
| SCMScenario( |
| "earth_science", |
| "volcanic ash emission", |
| "reduced incoming sunlight", |
| "short-term surface cooling", |
| "eruption magnitude", |
| ), |
| SCMScenario( |
| "agriculture", |
| "nitrogen fertilizer", |
| "plant nutrient availability", |
| "higher crop biomass", |
| "soil organic matter", |
| ), |
| SCMScenario( |
| "agriculture", |
| "pest infestation", |
| "leaf area loss", |
| "lower crop yield", |
| "natural predator abundance", |
| ), |
| SCMScenario( |
| "energy", |
| "wind speed", |
| "turbine rotor torque", |
| "higher wind-power output", |
| "air density", |
| ), |
| SCMScenario( |
| "energy", |
| "insulation quality", |
| "lower heat transfer", |
| "lower building energy demand", |
| "outdoor temperature", |
| ), |
| SCMScenario( |
| "transport", |
| "higher road capacity", |
| "less queue accumulation", |
| "shorter travel time", |
| "commuter demand", |
| ), |
| SCMScenario( |
| "transport", |
| "rail service disruption", |
| "passenger diversion to roads", |
| "greater highway congestion", |
| "alternative route capacity", |
| ), |
| SCMScenario( |
| "law_policy", |
| "tax rate increase", |
| "higher marginal cost", |
| "lower targeted consumption", |
| "household income", |
| ), |
| SCMScenario( |
| "law_policy", |
| "housing subsidy", |
| "lower effective housing cost", |
| "greater housing demand", |
| "local supply elasticity", |
| ), |
| SCMScenario( |
| "finance", |
| "higher default probability", |
| "higher expected credit loss", |
| "higher loan interest rate", |
| "borrower income volatility", |
| ), |
| SCMScenario( |
| "finance", |
| "diversified holdings", |
| "lower idiosyncratic exposure", |
| "lower portfolio volatility", |
| "asset correlation", |
| ), |
| SCMScenario( |
| "cybersecurity", |
| "multi-factor authentication", |
| "harder credential misuse", |
| "lower account takeover risk", |
| "password reuse", |
| ), |
| SCMScenario( |
| "cybersecurity", |
| "unpatched software", |
| "known exploitable vulnerability", |
| "higher breach risk", |
| "attacker access", |
| ), |
| SCMScenario( |
| "manufacturing", |
| "process temperature drift", |
| "dimensional variation", |
| "higher defect rate", |
| "incoming material variance", |
| ), |
| SCMScenario( |
| "manufacturing", |
| "preventive maintenance", |
| "lower machine downtime", |
| "higher production availability", |
| "equipment age", |
| ), |
| SCMScenario( |
| "telecommunications", |
| "network congestion", |
| "packet queueing", |
| "higher communication latency", |
| "available bandwidth", |
| ), |
| SCMScenario( |
| "telecommunications", |
| "stronger signal power", |
| "higher signal-to-noise ratio", |
| "lower decoding error rate", |
| "channel interference", |
| ), |
| SCMScenario( |
| "psychology", |
| "chronic stress", |
| "sleep disruption", |
| "lower working-memory performance", |
| "social support", |
| ), |
| SCMScenario( |
| "psychology", |
| "retrieval practice", |
| "stronger memory consolidation", |
| "better delayed recall", |
| "initial familiarity", |
| ), |
| SCMScenario( |
| "public_health", |
| "clean-water access", |
| "lower pathogen ingestion", |
| "lower diarrheal disease incidence", |
| "sanitation coverage", |
| ), |
| SCMScenario( |
| "public_health", |
| "air pollution exposure", |
| "airway inflammation", |
| "higher respiratory symptoms", |
| "baseline lung health", |
| ), |
| SCMScenario( |
| "materials", |
| "alloy strengthening", |
| "higher dislocation resistance", |
| "greater yield strength", |
| "heat-treatment history", |
| ), |
| SCMScenario( |
| "materials", |
| "repeated cyclic loading", |
| "fatigue crack growth", |
| "lower component lifetime", |
| "surface finish", |
| ), |
| SCMScenario( |
| "product_reliability", |
| "battery overcharge", |
| "electrolyte decomposition", |
| "shorter battery service life", |
| "ambient temperature", |
| ), |
| SCMScenario( |
| "product_reliability", |
| "insufficient cooling", |
| "higher processor temperature", |
| "thermal throttling", |
| "workload intensity", |
| ), |
| SCMScenario( |
| "supply_chain", |
| "port congestion", |
| "longer inbound lead time", |
| "retail stockout risk", |
| "seasonal demand", |
| ), |
| SCMScenario( |
| "supply_chain", |
| "supplier quality drift", |
| "higher incoming defect rate", |
| "lower production yield", |
| "inspection coverage", |
| ), |
| SCMScenario( |
| "software_product", |
| "rate-limit configuration", |
| "fewer burst requests", |
| "lower service overload risk", |
| "traffic distribution", |
| ), |
| SCMScenario( |
| "software_product", |
| "stale dependency", |
| "known vulnerable code path", |
| "higher exploitation risk", |
| "attacker capability", |
| ), |
| SCMScenario( |
| "consumer_product", |
| "clearer setup instructions", |
| "fewer configuration errors", |
| "higher successful first-use rate", |
| "user technical experience", |
| ), |
| SCMScenario( |
| "consumer_product", |
| "higher shipping damage", |
| "more returned units", |
| "lower customer satisfaction", |
| "packaging quality", |
| ), |
| ) |
|
|
|
|
| def _scm_graph( |
| scenario: SCMScenario, |
| variant: str, |
| *, |
| source_id: str = "manual_scm_v1", |
| source_version: str = "1", |
| scm_version: str = "manual_v1", |
| ) -> tuple[RealGraph, str, str]: |
| prefix = f"scm:{scenario.domain}" |
| cause_id = _node_id(prefix, scenario.cause) |
| mediator_id = _node_id(prefix, scenario.mediator) |
| effect_id = _node_id(prefix, scenario.effect) |
| confounder_id = _node_id(prefix, scenario.confounder) |
| nodes = { |
| cause_id: "variable", |
| mediator_id: "mediator", |
| effect_id: "outcome", |
| confounder_id: "confounder", |
| } |
| labels = { |
| cause_id: scenario.cause, |
| mediator_id: scenario.mediator, |
| effect_id: scenario.effect, |
| confounder_id: scenario.confounder, |
| } |
| attributes = { |
| node: {"domain": scenario.domain, "synthetic_scm": scm_version} |
| for node in nodes |
| } |
| edge_attributes = { |
| "evidence_kind": "synthetic_structural_causal_model", |
| "confidence": "ground_truth_for_synthetic_example", |
| } |
| edges = [ |
| EdgeRecord( |
| cause_id, |
| "causes", |
| mediator_id, |
| "variable", |
| "mediator", |
| relation_phrase("causes"), |
| edge_attributes, |
| ), |
| EdgeRecord( |
| mediator_id, |
| "causes", |
| effect_id, |
| "mediator", |
| "outcome", |
| relation_phrase("causes"), |
| edge_attributes, |
| ), |
| EdgeRecord( |
| confounder_id, |
| "causes", |
| cause_id, |
| "confounder", |
| "variable", |
| relation_phrase("causes"), |
| edge_attributes, |
| ), |
| EdgeRecord( |
| confounder_id, |
| "causes", |
| effect_id, |
| "confounder", |
| "outcome", |
| relation_phrase("causes"), |
| edge_attributes, |
| ), |
| ] |
| if variant == "unsupported": |
| edges = [ |
| EdgeRecord( |
| cause_id, |
| "associated_with", |
| effect_id, |
| "variable", |
| "outcome", |
| relation_phrase("associated_with"), |
| { |
| "evidence_kind": "observational_association_only", |
| "confidence": "not_a_causal_claim", |
| }, |
| ) |
| ] |
| elif variant == "reverse": |
| edges = [ |
| EdgeRecord( |
| effect_id, |
| "before", |
| cause_id, |
| "outcome", |
| "variable", |
| relation_phrase("before"), |
| { |
| "evidence_kind": "temporal_or_reverse_hard_negative", |
| "confidence": "hard_negative", |
| }, |
| ) |
| ] |
| source = SourceMetadata( |
| source_id=source_id, |
| domain=scenario.domain, |
| version=source_version, |
| license="project-authored", |
| url="https://example.invalid/project-authored-scm", |
| source_kind="synthetic_structural_model", |
| ) |
| return ( |
| _make_graph( |
| source, |
| nodes, |
| edges, |
| labels=labels, |
| node_attributes=attributes, |
| relation_ids={edge.relation_id for edge in edges}, |
| ), |
| cause_id, |
| effect_id, |
| ) |
|
|
|
|
| def generate_manual_scm_examples( |
| *, |
| count: int, |
| seed: int = 0, |
| source_id: str = "manual_scm_v1", |
| source_version: str = "1", |
| scm_version: str = "manual_v1", |
| stage_name: str = "causal_manual_scm", |
| generator_version: str = "causal-corpus-v1", |
| ) -> Iterator[CausalExample]: |
| """Generate manually authored, domain-balanced SCM examples.""" |
| rng = random.Random(seed) |
| for index in range(count): |
| scenario = SCENARIOS[index % len(SCENARIOS)] |
| composition: tuple[str, ...] |
| answers: tuple[str, ...] |
| paths: tuple[GraphPath, ...] |
| variant = rng.choices( |
| ("supported", "unsupported", "reverse"), weights=(0.64, 0.24, 0.12) |
| )[0] |
| real_graph, cause_id, effect_id = _scm_graph( |
| scenario, |
| variant, |
| source_id=source_id, |
| source_version=source_version, |
| scm_version=scm_version, |
| ) |
| if variant == "reverse": |
| start_id = effect_id |
| composition = ("causes",) |
| task_type = "cause_identification" |
| status = ReasoningStatus.INSUFFICIENT_EVIDENCE |
| answers = () |
| paths = () |
| query_text = ( |
| f"Which outcome can be caused by intervening on {scenario.effect}?" |
| ) |
| elif variant == "unsupported": |
| start_id = cause_id |
| composition = ("causes",) |
| task_type = "intervention" |
| status = ReasoningStatus.INSUFFICIENT_EVIDENCE |
| answers = () |
| paths = () |
| query_text = ( |
| f"If an intervention changes {scenario.cause}, which outcome is " |
| f"predicted to change through a causal mechanism?" |
| ) |
| else: |
| start_id = cause_id |
| composition = ("causes", "causes") |
| task_type = rng.choice( |
| ( |
| "mechanism_chain", |
| "intervention", |
| "counterfactual", |
| "effect_prediction", |
| ) |
| ) |
| target_type = "outcome" |
| query_text = _causal_query( |
| scenario.cause, |
| target_type, |
| [relation_phrase("causes"), relation_phrase("causes")], |
| 2, |
| task_type, |
| ) |
| paths = ( |
| GraphPath( |
| ( |
| PathStep( |
| cause_id, |
| "causes", |
| _node_id(f"scm:{scenario.domain}", scenario.mediator), |
| "variable", |
| "mediator", |
| {}, |
| ), |
| PathStep( |
| _node_id(f"scm:{scenario.domain}", scenario.mediator), |
| "causes", |
| effect_id, |
| "mediator", |
| "outcome", |
| {}, |
| ), |
| ) |
| ), |
| ) |
| status = ReasoningStatus.SUPPORTED |
| answers = (effect_id,) |
| query = GraphQuery( |
| text=query_text, |
| start_entities=(start_id,), |
| start_entity_ids=(start_id,), |
| max_hops=3, |
| min_hops=1, |
| return_paths=True, |
| verify=True, |
| operation="find_entities", |
| answer_mode="entity", |
| ) |
| example = ReasoningExample( |
| example_id=f"{source_id}-{seed}-{index}", |
| graph=real_graph.graph.to_example(), |
| query=query, |
| answers=answers, |
| negative_answers=(), |
| supporting_paths=paths, |
| conflicting_paths=(), |
| status=status, |
| difficulty=len(composition), |
| seed=seed + index, |
| generator_version=generator_version, |
| task_type=task_type, |
| intent_id=f"{scenario.domain}:{task_type}", |
| composition=composition, |
| stage_name=stage_name, |
| ) |
| yield CausalExample( |
| example, |
| { |
| "task_type": task_type, |
| "evidence_kind": ( |
| "synthetic_structural_causal_model" |
| if variant == "supported" |
| else "observational_association_only" |
| if variant == "unsupported" |
| else "temporal_or_reverse_hard_negative" |
| ), |
| "evidence_class": "structural_causal_model", |
| "causal_truth_status": "explicit_ground_truth", |
| "causal_relation": "causes", |
| "support": "present" if variant == "supported" else "absent", |
| "negative_kind": "none" if variant == "supported" else variant, |
| "domain": scenario.domain, |
| "mechanism": ( |
| f"{scenario.cause} -> {scenario.mediator} -> {scenario.effect}" |
| ), |
| "confounder": scenario.confounder, |
| "truth_status": ( |
| "ground_truth" |
| if variant == "supported" |
| else "ground_truth_abstention" |
| if variant == "unsupported" |
| else "ground_truth_directional_negative" |
| ), |
| "intervention": f"do({scenario.cause})", |
| "backdoor_confounder": scenario.confounder, |
| "counterfactual_queryable": "true", |
| }, |
| ) |
|
|
|
|
| def write_causal_jsonl_gz( |
| examples: Iterable[CausalExample], |
| destination: str | Path, |
| source: SourceMetadata, |
| ) -> CorpusStats: |
| """Write causal records using the existing trainer-compatible JSONL format.""" |
| path = Path(destination) |
| path.parent.mkdir(parents=True, exist_ok=True) |
| counts: Counter[str] = Counter() |
| with gzip.open(path, "wt", encoding="utf-8") as handle: |
| for item in examples: |
| truth_status = str( |
| item.metadata.get("causal_truth_status", source.causal_truth_status) |
| ) |
| if ( |
| truth_status == "explicit_ground_truth" |
| and not source.source_id.startswith("manual_scm") |
| ): |
| raise ValueError( |
| "only manual_scm sources may emit explicit_ground_truth records" |
| ) |
| if source.source_id.startswith("manual_scm") and truth_status != ( |
| "explicit_ground_truth" |
| ): |
| raise ValueError( |
| "manual_scm sources must emit explicit_ground_truth records" |
| ) |
| payload = serialize_example(item.example, source) |
| metadata = dict(item.metadata) |
| metadata["causal_truth_status"] = truth_status |
| payload["causal_metadata"] = metadata |
| handle.write(json.dumps(payload, sort_keys=True)) |
| handle.write("\n") |
| counts["examples"] += 1 |
| counts[item.example.status.value] += 1 |
| counts[f"source:{source.source_id}"] += 1 |
| counts[f"hops:{len(item.example.composition)}"] += 1 |
| counts[ |
| f"task:{item.metadata.get('task_type', item.example.task_type)}" |
| ] += 1 |
| counts[f"evidence:{item.metadata.get('evidence_kind', 'unknown')}"] += 1 |
| return CorpusStats( |
| examples=counts["examples"], |
| supported=counts[ReasoningStatus.SUPPORTED.value], |
| insufficient_evidence=counts[ReasoningStatus.INSUFFICIENT_EVIDENCE.value], |
| sources={source.source_id: counts[f"source:{source.source_id}"]}, |
| hops={ |
| key.removeprefix("hops:"): value |
| for key, value in counts.items() |
| if key.startswith("hops:") |
| }, |
| ) |
|
|
