Current
Typed knowledge pipelines
Benchmark first
Corpora: BEIR, FEVER, and LongMemEval form the end-to-end matrix; deterministic fixture suites cover stores, serialization, compiler output, and failure recovery.
Protocol: Pin every input, dependency, model, prompt, and seed. Preserve per-case traces and report quality beside p50/p95 latency, tokens, storage, index-build time, refresh work, and failure count. Compare outputs after restart and across supported stores; compiled configurations must reproduce the same artifact graph and evaluator inputs.
Composable Stage values transform immutable batches and return outputs plus a complete StageTrace. Domain stages may emit reviewable artifact mutations; the runner itself performs no storage writes.
How it works
Each stage declares input/output types, a versioned configuration fingerprint, and whether it is pure or calls an injected service. The runner topologically orders stages, passes immutable batches, records input revisions and stage results, and stops dependent stages after failure. Outputs are mutation proposals; a final policy validates evidence, scope, and expected artifact revision before the application commits them.
Research basisPipeline provenance research ties reproducibility to captured inputs, transformations, and configuration. Data Cascades documents how upstream data failures compound downstream. This motivates stage identities, dependency traces, and visible failures; the generic stage and mutation types are Mari’s composition boundary.
extract*→resolve→link→review→*index
from mari_components.platform import Pipeline, Stage
pipeline = Pipeline(
stages=(
Stage(
name="normalize",
version="1",
transform=lambda rows: (row.strip() for row in rows),
),
Stage(
name="discard-empty",
version="2",
transform=lambda rows: (row for row in rows if row),
configuration={"preserve_order": True},
),
)
)
result = pipeline.run([" policy ", ""])
assert result.outputs == ("policy",)
assert result.succeeded
assert result.trace[0].fingerprint