API reference¶

Cairn is a library as well as a command. Everything the CLI and the service do goes through Pipeline, so a script behaves the same as a command.

Pipeline¶

from cairn.core.config import load_config
from cairn.pipeline import Pipeline

cfg = load_config("configs/default.yaml")
pipeline = Pipeline.from_config(cfg)

manifest = pipeline.ingest()          # build the index, return the manifest
answer = pipeline.answer("How many days to request a credit?")

print(answer.status.value, answer.confidence)
for claim in answer.claims:
    print(claim.supported, claim.text)
    for citation in claim.citations:
        print(f"  {citation.quote!r}")

Method	Returns	Notes
`Pipeline.from_config(cfg)`	`Pipeline`	builds every component the configuration names
`ingest()`	`IndexManifest`	parses, chunks, embeds and writes the index atomically
`load_index()`	`IndexManifest`	raises `ManifestMismatch` when the configuration and the index disagree
`retrieve(question, k=None)`	`list[ScoredChunk]`	`k` defaults to `retrieval.k`
`retrieve_timed(question, k=None)`	`(list[ScoredChunk], dict[str, int])`	also per-stage latency
`answer(question, threshold=None)`	`Answer`	uses the calibrated threshold when one exists
`calibration_threshold()`	`float \| None`	`None` when absent or fitted for a different configuration
`calibration_record()`	`CalibrationRecord \| None`	the record and its evidence

The index loads on first use, so retrieve and answer work without an explicit load_index().

Evaluating in a script¶

from cairn.eval.golden import load_golden, select_split
from cairn.eval.report import write_report
from cairn.eval.runner import run_eval

cases = load_golden(cfg.golden_path())
evaluation = select_split(cases, "eval", cfg.calibration.split, cfg.calibration.seed)
report = run_eval(cfg, evaluation, pipeline.calibration_threshold(), pipeline)
write_report(report, "reports")

print(report.headline())

Gating in a script¶

from cairn.core.types import GateThresholds
from cairn.eval.gate import gate
from cairn.eval.report import read_report

result = gate(
    read_report("reports/latest/default.json"),
    read_report("reports/ci/default.json"),
    GateThresholds(max_quality_drop=0.01, max_cost_increase=0.2),
)
if not result.promote:
    for reason in result.reasons:
        print(reason)
    raise SystemExit(1)

The data model¶

Everything components exchange is a pydantic model, so it validates at the boundary and serialises without ceremony.

Type	What it is
`Document`	a parsed file: text, page offsets, content hash
`Chunk`	a slice with offsets into its document, plus page and section
`ScoredChunk`	a chunk with a score, the stage that produced it and its rank
`Citation`	a span inside a chunk, with the quote
`Claim`	an assertion, its citations, and whether they verified
`Answer`	status, text, claims, confidence, passages, usage, cost, latencies
`GoldenCase`	one evaluation row
`IndexManifest`	what built an index, and what makes it compatible
`EvalReport`	the four metric groups plus per-case rows
`GateResult`	promote or not, with reasons and deltas
`CalibrationRecord`	the threshold and the evidence for it

ScoredChunk.stage matters: a BM25 score, a cosine similarity and a reranker logit live on different scales and must never be compared directly.

Components and the registry¶

from cairn.core.registry import build, load_builtins, names

load_builtins()
print(names("chunker"))     # ['fixed', 'sentence_window', 'structure']
chunker = build("chunker", "sentence_window", {"window": 4, "overlap": 1})

Kinds: parser, chunker, embedder, vector_index, lexical_index, retriever, reranker, rewriter, provider, judge, confidence. See Add a component.

Errors¶

Every exception derives from CairnError, and each names a stage so a failure says where it happened.

Exception	Raised when
`ConfigError`	a configuration or a component name is invalid
`IngestError`	a document could not be parsed or chunked
`IndexError_`	an index could not be built, saved or loaded
`ManifestMismatch`	the index was built with a different corpus, chunker or embedder; carries `fields`
`RetrievalError`	a retrieval stage failed
`ProviderError`	a provider failed, refused, or returned output that did not validate; carries `retryable`
`MissingDependency`	an optional extra is needed; the message names it
`EvalError`	the harness or the gate hit an inconsistency

from cairn.core.errors import ManifestMismatch

try:
    pipeline.load_index()
except ManifestMismatch as e:
    print("rebuild the index; these differ:", e.fields)

Serving¶

from cairn.serve.app import create_app

app = create_app(cfg)   # a FastAPI application, mountable in your own

Observability¶

from cairn.observability.drift import DriftMonitor, load_baseline
from cairn.observability.tracing import configure_tracing, span

configure_tracing()                 # reads the environment; a no-op when nothing is set
with span("cairn.retrieve", k=8):
    ...

monitor = DriftMonitor(load_baseline(pipeline.drift_baseline_path()))
monitor.observe(embedding, top1_score)
print(monitor.drift())

See Serve.