Mining a Causal Graph from a Corpus with causalatee.mining¶

causalatee.mining turns a corpus of raw documents into an aggregated causal graph: each document flows through detection/candidate-extraction/identification stages, and the resulting relations are reduced into a causalatee.graph.Graph by (cause, effect) pair. This is the same kind of resource CauseNet or CausalBank represent, built from your own corpus instead of an existing one.

This notebook demonstrates the pipeline mechanics with a small, deterministic stand-in extraction function instead of a real model, so it runs without downloading any weights or needing a GPU. A real pipeline plugs in causalatee.models Protocol implementations (Detection, CandidateExtraction, PairwiseIdentification) at the .filter/.map stages instead -- see the API reference for the full stage-by-stage design.

Setup¶

causalatee.mining needs the mining extra.

In [ ]:

%pip install -q "causalatee[mining]"

%pip install -q "causalatee[mining]"

Define a document source and a toy extraction function¶

Pipeline.documents consumes any AsyncIterable[Document], adapting each document's .text into the pipeline's value and folding its .id/.metadata into a "source" key of the item's provenance metadata (see Pipeline.documents) -- here each document also carries a url, so provenance flows automatically all the way to the mined graph without the extraction function itself ever seeing it. Real corpora would stream from a database, an API, or a file on disk; here, four documents are yielded directly. The extraction function only ever sees plain text -- it stays metadata-blind, matching every causalatee.models Protocol -- and returns causalatee's ExtractedRelation shape ({"e1", "e2", "relationship", "score"} -- see the glossary for why this is "relationship", not "relation": one (e1, e2) pair's classification into a relation, not the relation itself), the same shape causalatee.models.compose_extraction produces from real Detection/CandidateExtraction/ PairwiseIdentification models.

In [ ]:

from causalatee.mining import Document, Pipeline, graph_sink


async def source():
    yield Document(
        id="1",
        text="The storm caused severe flooding across the region.",
        metadata={"url": "https://news.example/1"},
    )
    yield Document(
        id="2",
        text="Analysts say the storm caused flooding downtown.",
        metadata={"url": "https://news.example/2"},
    )
    yield Document(
        id="3",
        text="Sugar does not cause hyperactivity in children.",
        metadata={"url": "https://news.example/3"},
    )
    yield Document(
        id="4",
        text="Rising sea levels are driven by climate change.",
        metadata={"url": "https://news.example/4"},
    )


def toy_extraction(text):
    text = text.lower()
    if "storm" in text and "flooding" in text:
        return [{"e1": "storm", "e2": "flooding", "relationship": "Causal", "score": 0.92}]
    if "sugar" in text and "hyperactivity" in text:
        return [{"e1": "sugar", "e2": "hyperactivity", "relationship": "Countercausal", "score": 0.81}]
    if "sea levels" in text and "climate change" in text:
        return [{"e1": "climate change", "e2": "sea levels", "relationship": "Causal", "score": 0.88}]
    return []

from causalatee.mining import Document, Pipeline, graph_sink async def source(): yield Document( id="1", text="The storm caused severe flooding across the region.", metadata={"url": "https://news.example/1"}, ) yield Document( id="2", text="Analysts say the storm caused flooding downtown.", metadata={"url": "https://news.example/2"}, ) yield Document( id="3", text="Sugar does not cause hyperactivity in children.", metadata={"url": "https://news.example/3"}, ) yield Document( id="4", text="Rising sea levels are driven by climate change.", metadata={"url": "https://news.example/4"}, ) def toy_extraction(text): text = text.lower() if "storm" in text and "flooding" in text: return [{"e1": "storm", "e2": "flooding", "relationship": "Causal", "score": 0.92}] if "sugar" in text and "hyperactivity" in text: return [{"e1": "sugar", "e2": "hyperactivity", "relationship": "Countercausal", "score": 0.81}] if "sea levels" in text and "climate change" in text: return [{"e1": "climate change", "e2": "sea levels", "relationship": "Causal", "score": 0.88}] return []

Run the pipeline¶

graph_sink() accepts a support mapping (named, always-uncapped support DIMENSIONS -- here just "how many distinct documents mention this pair") and an evidence policy (a BOUNDED sample of raw mention metadata, here at most 2 per edge) -- see the API reference for why these are two deliberately separate mechanisms. .map(toy_extraction, concurrency=1) runs the extraction stage; .reduce_items(graph_sink(...)) drains the pipeline into the sink, which spools every relation (paired with its item's metadata) unaggregated and only aggregates lazily on first access. reduce_items -- rather than plain reduce -- is what hands the sink the full item (value AND metadata) it needs to compute provenance; ordinary sinks that don't care about metadata keep using .reduce() instead. Document 1 and 2 both mention storm -> flooding, so that pair's support should end up as 2 (not two separate edges), with two evidence entries tracing back to each document's own URL.

In [ ]:

from causalatee.mining import EvidencePolicy

sink = graph_sink(
    support={"documents": lambda m: m["source"]["id"]},
    evidence=EvidencePolicy(max_items=2),
)
graph = await Pipeline.documents(source()).map(toy_extraction, concurrency=1).reduce_items(sink)

# Not closed yet -- inspect first, close only once fully done with it (below).
print(f"{len(graph.nodes)} nodes, {len(graph.edges)} edges")
for edge in sorted(graph.edges, key=lambda e: e.source.id):
    print(f"{edge.source.id!r} -> {edge.target.id!r}:")
    for key, value in edge.metadata.items():
        print(f"    {key}: {value!r}")

from causalatee.mining import EvidencePolicy sink = graph_sink( support={"documents": lambda m: m["source"]["id"]}, evidence=EvidencePolicy(max_items=2), ) graph = await Pipeline.documents(source()).map(toy_extraction, concurrency=1).reduce_items(sink) # Not closed yet -- inspect first, close only once fully done with it (below). print(f"{len(graph.nodes)} nodes, {len(graph.edges)} edges") for edge in sorted(graph.edges, key=lambda e: e.source.id): print(f"{edge.source.id!r} -> {edge.target.id!r}:") for key, value in edge.metadata.items(): print(f" {key}: {value!r}")

Expected output:

6 nodes, 3 edges
'climate change' -> 'sea levels':
    support: 1
    causal_count: 1
    countercausal_count: 0
    avg_score: 0.88
    support_by: {'documents': 1}
    evidence: [{'source': {'id': '4', 'url': 'https://news.example/4'}}]
'storm' -> 'flooding':
    support: 2
    causal_count: 2
    countercausal_count: 0
    avg_score: 0.92
    support_by: {'documents': 2}
    evidence: [{'source': {'id': '1', 'url': 'https://news.example/1'}}, {'source': {'id': '2', 'url': 'https://news.example/2'}}]
'sugar' -> 'hyperactivity':
    support: 1
    causal_count: 0
    countercausal_count: 1
    avg_score: 0.81
    support_by: {'documents': 1}
    evidence: [{'source': {'id': '3', 'url': 'https://news.example/3'}}]

storm -> flooding aggregated the two mentions into one edge (support=2), and the countercausal mention of sugar -> hyperactivity is a real Relation.Countercausal, not folded into the causal count -- causal_count/countercausal_count are tracked separately, matching how CCNC itself distinguishes the two (the only dataset in causalatee's whole collection with a genuine denial-of- causation signal, not merely negative-outcome causation -- see the glossary/Relation docs). support_by["documents"] always equals support here because each document only mentions each pair once in this toy example -- with real data the two can diverge (e.g. the same document mentioning a pair in two different sentences would raise support without raising a support_by["documents"] dimension counting distinct documents). evidence is capped at 2 items regardless of how high support gets -- try lowering EvidencePolicy(max_items=...) to 1 above and re-running to see the cap take effect on storm -> flooding, the only edge with more than one mention in this example.

Persist the mined graph as CGF¶

GraphSink is a causalatee.graph.Graph, so it can be passed to save_cgf directly, no conversion step needed -- the same as any other Graph implementation in this package.

In [ ]:

from causalatee.graph import load_cgf, save_cgf

with graph:
    save_cgf(graph, "mined.cgf")

with load_cgf("mined.cgf", validate=True) as mapped:
    storm = mapped.get_node("storm")
    (out_edge,) = list(storm.outgoing_edges())
    print(f"{out_edge.source.id!r} -> {out_edge.target.id!r}: {dict(out_edge.metadata)}")

from causalatee.graph import load_cgf, save_cgf with graph: save_cgf(graph, "mined.cgf") with load_cgf("mined.cgf", validate=True) as mapped: storm = mapped.get_node("storm") (out_edge,) = list(storm.outgoing_edges()) print(f"{out_edge.source.id!r} -> {out_edge.target.id!r}: {dict(out_edge.metadata)}")

Expected output:

'storm' -> 'flooding': {'support': 2, 'causal_count': 2, 'countercausal_count': 0, 'avg_score': 0.92, 'support_by': {'documents': 2}, 'evidence': [{'source': {'id': '1', 'url': 'https://news.example/1'}}, {'source': {'id': '2', 'url': 'https://news.example/2'}}]}

support_by/evidence round-trip through save_cgf/load_cgf exactly like the base fields -- GraphSink.edge_metadata_schema infers its Avro schema from the first aggregated edge specifically so a dynamic shape like this (which depends on what support/evidence were configured) doesn't need a schema hand-written in advance (see causalatee.graph.infer_avro_schema). with graph: closes GraphSink's temporary spool database once the block exits -- always close a GraphSink (or use it as a context manager, as above) once you're fully done reading from it, not before -- closing ends its usable lifetime, including for save_cgf.