causalatee.mining

A streaming, concurrency-aware pipeline for turning a corpus of documents into an aggregated causal graph: source -> flat_map -> filter -> map -> map -> reduce.

Nothing is materialized between stages — an entire corpus, however large, streams through a bounded number of items in flight at once. Sync and async stages compose uniformly; a stage backed by a single GPU model (any causalatee.models protocol instance) should run with concurrency=1 (serialized calls, batched internally via batch_size), while I/O-bound stages (fetching documents) can run with much higher concurrency.

Quick start

from causalatee.mining import Document, Pipeline, causal_predicate, graph_sink
from causalatee.graph import save_cgf

async def source():
    async for post in fetch_posts():                       # user-supplied, I/O-bound
        yield Document(id=post.id, text=post.title)

graph = await (
    Pipeline(source())
    .flat_map(split_into_sentences, concurrency=8)          # CPU-bound; fan out
    .filter(detection_model, batch_size=64, concurrency=1,   # 1 GPU: internal batching
            predicate=causal_predicate)
    .map(candidate_extraction_model, batch_size=64, concurrency=1)  # same GPU
    .map(identification_model, batch_size=64, concurrency=1)        # same GPU
    .reduce(graph_sink())
)
save_cgf(graph, "mined.cgf")

identification_model here is a PairwiseIdentification; each mapped text must already carry candidate-span markers (<e0>, <e1>, ...) for this to work directly — see causalatee.models.identify_candidates for the marking/enumeration/remapping glue, shared with compose_extraction, that a real identification stage function should wrap around the raw model.

Why not just materialize the corpus and loop?

Because a real corpus doesn't fit in memory, and a naive for document in corpus: run_everything(document) loop runs every stage sequentially per item — the GPU sits idle while the next document is fetched, and the next document isn't fetched until the current one finishes identification. Pipeline overlaps stages (document N+1 is being fetched while document N is mid-identification) and lets each stage set its own concurrency independently, while guaranteeing a concurrency=1 GPU stage never receives more than one call at a time regardless of how concurrent its neighboring stages are.

Reducing into a graph

graph_sink is both the Pipeline.reduce sink and (after reduce completes) a causalatee.graph.Graph — await pipeline.reduce(graph_sink()) returns something directly usable with save_cgf. Every relation the pipeline emits is spooled to a temp SQLite database as it arrives (bounded memory regardless of corpus size); aggregation by (cause, effect) pair happens once, lazily, the first time .nodes/ .edges is accessed, and populates a SQLGraph with one add_edge call per aggregated pair — GraphSink only owns the spooling/aggregation, SQLGraph is the generic, reusable storage engine. Close it (or use it as a context manager) once done with the resulting graph, the same way load_cgf is used.

Example

See the Mining a Causal Graph notebook for a runnable walkthrough of the full pipeline -> graph_sink -> save_cgf flow, using a deterministic stand-in extraction function so it runs without a real model.

API Reference

mining

Classes

EvidencePolicy dataclass

EvidencePolicy(max_items: int = 3)

How many raw mentions' metadata to keep as an illustrative evidence sample per aggregated edge.

support/support_by counts are ALWAYS computed from the complete mention set regardless of this policy -- this only bounds the separate, illustrative sample embedded in edge metadata, never the counts themselves.

Methods:

select

select(mentions: list[Mapping[str, object]]) -> list[Mapping[str, object]]

First max_items mentions, in spool order -- the simplest deterministic policy. Subclass (or swap in a different callable-based policy) for a different selection strategy, e.g. highest-score or a random sample.

Source code in causalatee/mining/_graph_sink.py

def select(self, mentions: list[Mapping[str, object]]) -> list[Mapping[str, object]]:
    """First ``max_items`` mentions, in spool order -- the simplest deterministic policy. Subclass (or swap
    in a different callable-based policy) for a different selection strategy, e.g. highest-score or a random
    sample."""

    return mentions[: self.max_items]

GraphSink

GraphSink(*, support: Mapping[str, Callable[[Mapping[str, object]], Hashable]] | None = None, evidence: EvidencePolicy | None = None)

Bases: SQLGraph

Reduce sink + resulting Graph, see module docstring.

support declares named support dimensions beyond the raw mention count -- e.g. {"sentences": lambda m: (m["source"]["id"], m["sentence_index"]), "documents": lambda m: m["source"]["id"]} for a WARC corpus. Each function maps one mention's metadata to a hashable grouping key; the resulting edge metadata's support_by[name] is the number of DISTINCT keys seen for that edge, always over the complete mention set. evidence, if given, bounds a separate sample of raw mention metadata embedded per edge as evidence -- omit it (the default) for no evidence at all.

Source code in causalatee/mining/_graph_sink.py

def __init__(
    self,
    *,
    support: Mapping[str, Callable[[Mapping[str, object]], Hashable]] | None = None,
    evidence: EvidencePolicy | None = None,
) -> None:
    """``support`` declares named support dimensions beyond the raw mention count -- e.g.
    ``{"sentences": lambda m: (m["source"]["id"], m["sentence_index"]), "documents": lambda m:
    m["source"]["id"]}`` for a WARC corpus. Each function maps one mention's metadata to a hashable grouping
    key; the resulting edge metadata's ``support_by[name]`` is the number of DISTINCT keys seen for that edge,
    always over the complete mention set. ``evidence``, if given, bounds a separate sample of raw mention
    metadata embedded per edge as ``evidence`` -- omit it (the default) for no evidence at all."""

    self._support = dict(support) if support else {}
    self._evidence_policy = evidence

    self._tempdir = tempfile.TemporaryDirectory(prefix="causalatee-mining-")
    db_path = Path(self._tempdir.name) / "mentions.sqlite3"
    connection = sqlite3.connect(db_path)
    connection.execute(
        "CREATE TABLE mentions (cause TEXT NOT NULL, effect TEXT NOT NULL, "
        "relationship TEXT NOT NULL, score REAL NOT NULL, metadata TEXT NOT NULL)"
    )
    connection.commit()
    super().__init__(connection)
    self._aggregated = False
    self._aggregation_error: Exception | None = None
    self._edge_metadata_schema: Mapping[str, object] | None = None

Attributes

edge_metadata_schema property

edge_metadata_schema: Mapping[str, object]

Inferred (via causalatee.graph.infer_avro_schema) from the first aggregated edge, since support_by/evidence's shape depends entirely on this instance's own support/evidence configuration -- there is no static schema that could anticipate it. Falls back to the base support/causal_count/countercausal_count/avg_score-only schema if aggregation produced no edges at all (an empty corpus).

Methods:

close

close() -> None

Close the spool database and remove its temp directory.

Source code in causalatee/mining/_graph_sink.py

def close(self) -> None:
    """Close the spool database and remove its temp directory."""

    self._connection.close()
    self._tempdir.cleanup()

GraphSinkError

Bases: ValueError

Raised when GraphSink can't safely aggregate its spooled mentions into a graph -- currently, only when two edges' metadata have genuinely different shapes (see GraphSink._ensure_aggregated's own comment).

Pipeline

Pipeline(source: AsyncIterable[T], _executors: list[ThreadPoolExecutor] | None = None)

Bases: Generic[T]

A chainable, lazily-evaluated stream of items of type T.

Nothing runs until :meth:reduce/:meth:reduce_items (or manual async for) is awaited -- every stage is composed by wrapping async generators, never materialized in between.

Wrap a plain AsyncIterable[T] of bare values -- each is paired with empty metadata. Use :meth:documents instead to build a pipeline whose items already carry source provenance.

Source code in causalatee/mining/_pipeline.py

def __init__(self, source: AsyncIterable[T], _executors: list[ThreadPoolExecutor] | None = None) -> None:
    """Wrap a plain ``AsyncIterable[T]`` of bare values -- each is paired with empty metadata. Use
    :meth:`documents` instead to build a pipeline whose items already carry source provenance."""

    async def wrapped() -> AsyncIterator[PipelineItem[T]]:
        async for value in source:
            yield PipelineItem(value)

    self._source: AsyncIterable[PipelineItem[T]] = wrapped()
    self._executors: list[ThreadPoolExecutor] = _executors if _executors is not None else []

Methods:

documents classmethod

documents(source: DocumentSource) -> Pipeline[str]

Adapt a DocumentSource into a Pipeline[str]: each Document's .text becomes the item's value, and its .id/.metadata are folded into a "source" sub-key of the item's metadata -- so extraction/detection models keep operating on plain strings (matching every causalatee.models Protocol) while provenance flows automatically from the very first stage.

Source code in causalatee/mining/_pipeline.py

@classmethod
def documents(cls, source: DocumentSource) -> Pipeline[str]:
    """Adapt a ``DocumentSource`` into a ``Pipeline[str]``: each ``Document``'s ``.text`` becomes the item's
    ``value``, and its ``.id``/``.metadata`` are folded into a ``"source"`` sub-key of the item's metadata --
    so extraction/detection models keep operating on plain strings (matching every ``causalatee.models``
    Protocol) while provenance flows automatically from the very first stage."""

    async def wrapped() -> AsyncIterator[PipelineItem[str]]:
        async for document in source:
            yield PipelineItem(document.text, {"source": {"id": document.id, **document.metadata}})

    return cls._from_items(wrapped(), [])

map

map(fn: AsyncOrSyncFn, *, concurrency: int = 1, batch_size: int = 1) -> Pipeline[R]

Apply fn to every item's value, preserving its metadata unchanged.

At batch_size=1 (default), fn is called once per item, naturally (fn(value) -> result) -- the ordinary single-item mapping convention.

At batch_size > 1, fn is called once per group of up to batch_size values, and must accept and return a list of the SAME length (fn(values: list[T]) -> list[R]) -- enforced via zip(..., strict=True) when results are paired back with their original items, so a model that silently drops or duplicates a result raises immediately instead of misattributing provenance. causalatee.models protocol instances are already batch-native this way, so plugging one in as fn with batch_size > 1 needs no extra glue.

concurrency bounds how many calls to fn may be in flight at once. Use concurrency=1 for a single GPU model (its own internal batching, via batch_size, is what gives it throughput -- concurrent calls would just contend for the same device). Use a higher concurrency for I/O-bound work.

Source code in causalatee/mining/_pipeline.py

def map(
    self,
    fn: AsyncOrSyncFn,
    *,
    concurrency: int = 1,
    batch_size: int = 1,
) -> Pipeline[R]:
    """Apply ``fn`` to every item's ``value``, preserving its metadata unchanged.

    At ``batch_size=1`` (default), ``fn`` is called once per item, naturally (``fn(value) -> result``) -- the
    ordinary single-item mapping convention.

    At ``batch_size > 1``, ``fn`` is called once per group of up to ``batch_size`` values, and must accept
    and return a list of the SAME length (``fn(values: list[T]) -> list[R]``) -- enforced via ``zip(...,
    strict=True)`` when results are paired back with their original items, so a model that silently drops or
    duplicates a result raises immediately instead of misattributing provenance. ``causalatee.models`` protocol
    instances are already batch-native this way, so plugging one in as ``fn`` with ``batch_size > 1`` needs
    no extra glue.

    ``concurrency`` bounds how many calls to ``fn`` may be in flight at once. Use ``concurrency=1`` for a
    single GPU model (its own internal batching, via ``batch_size``, is what gives it throughput --
    concurrent calls would just contend for the same device). Use a higher ``concurrency`` for I/O-bound
    work.
    """

    async_fn = self._as_async(fn, concurrency)

    if batch_size == 1:

        async def item_fn(item: PipelineItem[T]) -> PipelineItem[R]:
            result = await async_fn(item.value)
            return PipelineItem(result, item.metadata)

        # aiostream's stubs model map's fn as accepting *args (for multi-source zipping we don't use);
        # verified correct at runtime, see module probe.
        mapped: Any = stream.map(self._source, item_fn, task_limit=concurrency)  # type: ignore[arg-type]

        async def one_at_a_time() -> AsyncIterator[PipelineItem[R]]:
            async with mapped.stream() as streamer:
                async for result in streamer:
                    yield result

        return Pipeline._from_items(one_at_a_time(), self._executors)

    batches = _in_batches(self._source, batch_size)

    async def batch_fn(batch: list[PipelineItem[T]]) -> list[PipelineItem[R]]:
        results = await async_fn([item.value for item in batch])
        return [PipelineItem(result, item.metadata) for item, result in zip(batch, results, strict=True)]

    # aiostream's stubs model map's fn as accepting *args (for multi-source zipping we don't use);
    # verified correct at runtime, see module probe.
    mapped_batches: Any = stream.map(batches, batch_fn, task_limit=concurrency)  # type: ignore[arg-type]

    async def unbatched() -> AsyncIterator[PipelineItem[R]]:
        async with mapped_batches.stream() as streamer:
            async for result_batch in streamer:
                for item in result_batch:
                    yield item

    return Pipeline._from_items(unbatched(), self._executors)

filter

filter(fn: AsyncOrSyncFn, *, predicate: Callable[[Any], bool] = bool, concurrency: int = 1, batch_size: int = 1) -> Pipeline[T]

Keep only items for which predicate(fn(value)) is true (or, batched, predicate(fn(values)[i]) for each value in the batch), preserving metadata on every kept item unchanged.

fn need not itself return a boolean -- e.g. plug a causalatee.models.Detection model in directly as fn with predicate=causal_predicate (see this package's causal_predicate) rather than writing that translation yourself each time.

Source code in causalatee/mining/_pipeline.py

def filter(
    self,
    fn: AsyncOrSyncFn,
    *,
    predicate: Callable[[Any], bool] = bool,
    concurrency: int = 1,
    batch_size: int = 1,
) -> Pipeline[T]:
    """Keep only items for which ``predicate(fn(value))`` is true (or, batched, ``predicate(fn(values)[i])``
    for each value in the batch), preserving metadata on every kept item unchanged.

    ``fn`` need not itself return a boolean -- e.g. plug a ``causalatee.models.Detection`` model in directly
    as ``fn`` with ``predicate=causal_predicate`` (see this package's ``causal_predicate``) rather than
    writing that translation yourself each time.
    """

    async_fn = self._as_async(fn, concurrency)

    if batch_size == 1:

        async def process_one(item: PipelineItem[T]) -> tuple[PipelineItem[T], Any]:
            return item, await async_fn(item.value)

        # aiostream's stubs model map's fn as accepting *args (for multi-source zipping we don't use);
        # verified correct at runtime, see module probe.
        mapped: Any = stream.map(self._source, process_one, task_limit=concurrency)  # type: ignore[arg-type]

        async def filtered_one() -> AsyncIterator[PipelineItem[T]]:
            async with mapped.stream() as streamer:
                async for item, result in streamer:
                    if predicate(result):
                        yield item

        return Pipeline._from_items(filtered_one(), self._executors)

    batches = _in_batches(self._source, batch_size)

    async def process_batch(batch: list[PipelineItem[T]]) -> list[tuple[PipelineItem[T], Any]]:
        results = await async_fn([item.value for item in batch])
        return list(zip(batch, results, strict=True))

    # aiostream's stubs model map's fn as accepting *args (for multi-source zipping we don't use);
    # verified correct at runtime, see module probe.
    mapped_batches: Any = stream.map(batches, process_batch, task_limit=concurrency)  # type: ignore[arg-type]

    async def filtered_batched() -> AsyncIterator[PipelineItem[T]]:
        async with mapped_batches.stream() as streamer:
            async for pairs in streamer:
                for item, result in pairs:
                    if predicate(result):
                        yield item

    return Pipeline._from_items(filtered_batched(), self._executors)

flat_map

flat_map(fn: AsyncOrSyncFn, *, concurrency: int = 1, annotate: Callable[[R, int], Mapping[str, object]] | None = None) -> Pipeline[R]

Apply fn to every item's value, where fn(value) -> Iterable[R] returns zero or more output values per input (e.g. splitting a document into sentences), and flatten the results into a single stream.

Each resulting item inherits the parent's metadata. annotate, if given, is called as annotate(child_value, index) for each of the (up to) N values fn produced (index counting from 0 within that one expansion) and its return value is MERGED on top of the inherited metadata -- the mechanism for attaching metadata the expansion itself introduces (e.g. a sentence's own text/index within its source document), without a separate .annotate() call per new field.

Source code in causalatee/mining/_pipeline.py

def flat_map(
    self,
    fn: AsyncOrSyncFn,
    *,
    concurrency: int = 1,
    annotate: Callable[[R, int], Mapping[str, object]] | None = None,
) -> Pipeline[R]:
    """Apply ``fn`` to every item's ``value``, where ``fn(value) -> Iterable[R]`` returns zero or more output
    values per input (e.g. splitting a document into sentences), and flatten the results into a single stream.

    Each resulting item inherits the parent's metadata. ``annotate``, if given, is called as
    ``annotate(child_value, index)`` for each of the (up to) N values ``fn`` produced (``index`` counting from
    0 within that one expansion) and its return value is MERGED on top of the inherited metadata -- the
    mechanism for attaching metadata the expansion itself introduces (e.g. a sentence's own text/index within
    its source document), without a separate ``.annotate()`` call per new field.
    """

    async_fn = self._as_async(fn, concurrency)

    async def item_fn(item: PipelineItem[T]) -> list[PipelineItem[R]]:
        results = await async_fn(item.value)
        children = []
        for index, value in enumerate(results):
            metadata = {**item.metadata, **annotate(value, index)} if annotate is not None else item.metadata
            children.append(PipelineItem(value, metadata))
        return children

    # aiostream's stubs model map's fn as accepting *args (for multi-source zipping we don't use);
    # verified correct at runtime, see module probe.
    mapped: Any = stream.map(self._source, item_fn, task_limit=concurrency)  # type: ignore[arg-type]

    async def flattened() -> AsyncIterator[PipelineItem[R]]:
        async with mapped.stream() as streamer:
            async for results in streamer:
                for item in results:
                    yield item

    return Pipeline._from_items(flattened(), self._executors)

annotate

annotate(name: str, fn: AsyncOrSyncFn, *, concurrency: int = 1) -> Pipeline[T]

Compute fn(value) and store it under metadata[name], leaving value itself unchanged.

Deliberately metadata-BLIND on input, matching map/filter/flat_map's "models operate on values" principle -- use :meth:annotate_with_metadata for a derived value that itself needs the item's existing metadata as input.

Source code in causalatee/mining/_pipeline.py

def annotate(self, name: str, fn: AsyncOrSyncFn, *, concurrency: int = 1) -> Pipeline[T]:
    """Compute ``fn(value)`` and store it under ``metadata[name]``, leaving ``value`` itself unchanged.

    Deliberately metadata-BLIND on input, matching ``map``/``filter``/``flat_map``'s "models operate on
    values" principle -- use :meth:`annotate_with_metadata` for a derived value that itself needs the item's
    existing metadata as input.
    """

    async_fn = self._as_async(fn, concurrency)

    async def item_fn(item: PipelineItem[T]) -> PipelineItem[T]:
        computed = await async_fn(item.value)
        return PipelineItem(item.value, {**item.metadata, name: computed})

    # aiostream's stubs model map's fn as accepting *args (for multi-source zipping we don't use);
    # verified correct at runtime, see module probe.
    mapped: Any = stream.map(self._source, item_fn, task_limit=concurrency)  # type: ignore[arg-type]

    async def annotated() -> AsyncIterator[PipelineItem[T]]:
        async with mapped.stream() as streamer:
            async for result in streamer:
                yield result

    return Pipeline._from_items(annotated(), self._executors)

annotate_with_metadata

annotate_with_metadata(name: str, fn: Callable[[T, Mapping[str, object]], Any], *, concurrency: int = 1) -> Pipeline[T]

Like :meth:annotate, but fn(value, metadata) also receives the item's CURRENT metadata -- for a derived value that needs existing provenance as input (e.g. a recency score computed from metadata["source"]["timestamp"] together with the text itself).

Source code in causalatee/mining/_pipeline.py

def annotate_with_metadata(
    self,
    name: str,
    fn: Callable[[T, Mapping[str, object]], Any],
    *,
    concurrency: int = 1,
) -> Pipeline[T]:
    """Like :meth:`annotate`, but ``fn(value, metadata)`` also receives the item's CURRENT metadata -- for a
    derived value that needs existing provenance as input (e.g. a recency score computed from
    ``metadata["source"]["timestamp"]`` together with the text itself)."""

    async_fn = self._as_async(fn, concurrency)

    async def item_fn(item: PipelineItem[T]) -> PipelineItem[T]:
        computed = await async_fn(item.value, item.metadata)
        return PipelineItem(item.value, {**item.metadata, name: computed})

    # aiostream's stubs model map's fn as accepting *args (for multi-source zipping we don't use);
    # verified correct at runtime, see module probe.
    mapped: Any = stream.map(self._source, item_fn, task_limit=concurrency)  # type: ignore[arg-type]

    async def annotated() -> AsyncIterator[PipelineItem[T]]:
        async with mapped.stream() as streamer:
            async for result in streamer:
                yield result

    return Pipeline._from_items(annotated(), self._executors)

map_metadata

map_metadata(fn: Callable[[Mapping[str, object]], Mapping[str, object]], *, concurrency: int = 1) -> Pipeline[T]

Replace an item's metadata with fn(metadata) -- for arbitrary restructuring (adding, removing, or reshaping fields), unlike annotate's always-additive single key. fn's return value REPLACES the metadata entirely; merge in **metadata yourself if you only mean to add to it.

Source code in causalatee/mining/_pipeline.py

def map_metadata(
    self,
    fn: Callable[[Mapping[str, object]], Mapping[str, object]],
    *,
    concurrency: int = 1,
) -> Pipeline[T]:
    """Replace an item's metadata with ``fn(metadata)`` -- for arbitrary restructuring (adding, removing, or
    reshaping fields), unlike ``annotate``'s always-additive single key. ``fn``'s return value REPLACES the
    metadata entirely; merge in ``**metadata`` yourself if you only mean to add to it."""

    async_fn = self._as_async(fn, concurrency)

    async def item_fn(item: PipelineItem[T]) -> PipelineItem[T]:
        new_metadata = await async_fn(item.metadata)
        return PipelineItem(item.value, new_metadata)

    # aiostream's stubs model map's fn as accepting *args (for multi-source zipping we don't use);
    # verified correct at runtime, see module probe.
    mapped: Any = stream.map(self._source, item_fn, task_limit=concurrency)  # type: ignore[arg-type]

    async def transformed() -> AsyncIterator[PipelineItem[T]]:
        async with mapped.stream() as streamer:
            async for result in streamer:
                yield result

    return Pipeline._from_items(transformed(), self._executors)

reduce async

reduce(sink: Callable[[T], Awaitable[None] | None]) -> Any

Drain the pipeline, calling sink(item.value) (sync or async) for every item -- metadata-BLIND, the ergonomic default every ordinary sink already expects. Shuts down every executor any stage in this chain created, then returns sink itself (most sinks, e.g. this package's graph sink, expose their accumulated result as an attribute/method after draining). See :meth:reduce_items for a sink that needs the full PipelineItem (value AND metadata).

Source code in causalatee/mining/_pipeline.py

async def reduce(self, sink: Callable[[T], Awaitable[None] | None]) -> Any:
    """Drain the pipeline, calling ``sink(item.value)`` (sync or async) for every item -- metadata-BLIND, the
    ergonomic default every ordinary sink already expects. Shuts down every executor any stage in this chain
    created, then returns ``sink`` itself (most sinks, e.g. this package's graph sink, expose their
    accumulated result as an attribute/method after draining). See :meth:`reduce_items` for a sink that needs
    the full ``PipelineItem`` (value AND metadata)."""

    return await self._drain(sink, lambda item: item.value)

reduce_items async

reduce_items(sink: Callable[[PipelineItem[T]], Awaitable[None] | None]) -> Any

Like :meth:reduce, but calls sink(item) with the full PipelineItem -- for a metadata-aware sink such as causalatee.mining.graph_sink, which needs provenance to compute per-edge evidence/support.

Source code in causalatee/mining/_pipeline.py

async def reduce_items(self, sink: Callable[[PipelineItem[T]], Awaitable[None] | None]) -> Any:
    """Like :meth:`reduce`, but calls ``sink(item)`` with the full ``PipelineItem`` -- for a metadata-aware
    sink such as ``causalatee.mining.graph_sink``, which needs provenance to compute per-edge evidence/support.
    """

    return await self._drain(sink, lambda item: item)

PipelineItem dataclass

PipelineItem(value: T, metadata: Mapping[str, object] = dict())

Bases: Generic[T]

One value flowing through a Pipeline, paired with its accumulated provenance/metadata.

metadata is intentionally generic and source-dependent (a source URL, a WARC record id, a derived sentence index or topic label, ...) -- Pipeline itself has no opinion on its shape, only on how it propagates.

Document dataclass

Document(id: str, text: str, metadata: Mapping[str, object] = dict())

A single whole document flowing into a mining Pipeline.

Functions:

graph_sink

graph_sink(*, support: Mapping[str, Callable[[Mapping[str, object]], Hashable]] | None = None, evidence: EvidencePolicy | None = None) -> GraphSink

Create a Pipeline.reduce_items sink that aggregates every relation-carrying item it receives into a causalatee.graph.Graph, by (cause, effect) pair across the whole corpus. See GraphSink for the aggregation/lifecycle/provenance details.

Source code in causalatee/mining/_graph_sink.py

def graph_sink(
    *,
    support: Mapping[str, Callable[[Mapping[str, object]], Hashable]] | None = None,
    evidence: EvidencePolicy | None = None,
) -> GraphSink:
    """Create a ``Pipeline.reduce_items`` sink that aggregates every relation-carrying item it receives into a
    ``causalatee.graph.Graph``, by ``(cause, effect)`` pair across the whole corpus. See ``GraphSink`` for the
    aggregation/lifecycle/provenance details.
    """

    return GraphSink(support=support, evidence=evidence)

causal_predicate

causal_predicate(result: Any) -> bool

predicate= for :meth:Pipeline.filter when fn is a causalatee.models.Detection model directly: true unless labeled "Uncausal" (case-insensitive) -- the exact check causalatee.models._ComposedExtraction makes internally, so a mining pipeline's detection-gating stage agrees with the end-to-end Extraction composition rather than drifting from it.

Source code in causalatee/mining/_pipeline.py

def causal_predicate(result: Any) -> bool:
    """``predicate=`` for :meth:`Pipeline.filter` when ``fn`` is a ``causalatee.models.Detection`` model directly: true
    unless labeled "Uncausal" (case-insensitive) -- the exact check ``causalatee.models._ComposedExtraction`` makes
    internally, so a mining pipeline's detection-gating stage agrees with the end-to-end ``Extraction`` composition
    rather than drifting from it."""

    return str(result["label"]).lower() != "uncausal"