ek

ek – a framework for building Knowledge Evaluation systems.

ek evaluates the outputs of information-extraction systems. OCR is treated as the noisiest special case of a general problem, so the core is source-agnostic (PDF, DOCX, tables, DB responses, LLM extractors) and the OCR specifics live in ek.ocr.

Two facades cover the two halves of evaluation, both operating on the same typed schema:

• score() / evaluate() – reference-based (offline): compare against a gold answer, one item or a whole corpus, the metric chosen by output type.

• estimate_quality() – reference-free (online): gather signals, calibrate, validate, and decide accept/flag/block with no gold answer.

Everything swappable is a typing.Protocol strategy resolved from ek.registry and injected as a keyword-only argument with a smart default – so the simple path Just Works and every layer stays replaceable (open-closed).

Quickstart:

>>> import ek
>>> round(ek.score("hello wrld", "hello world").value, 3)   # CER by default
0.091
>>> ek.evaluate([("ct", "cat"), ("dg", "dog")], metric="cer").n
2

The two-layer data model (GraphGrammar for the schema + cost weights, AnnotatedExtraction for per-extraction verification metadata) is the SSOT every component plugs into – see ek.base. Persistence is local-file dol stores under ~/.local/share/ek/ – see ek.stores.

class ek.AgreementSignal(cost_tier: int = 3, use_confidence: bool = True, conf_weight: float = 0.5, null_conf: float = 0.7, tokenize: Callable[[Any], List[Tuple[str | None, float]]] | None = None, max_tokens: int | None = 5000)

ROVER multi-engine agreement as a reference-free Signal.

Cost tier 3 (N engine runs): try the deterministic verifier layer and any free intrinsic confidence first. Called on a collection of hypotheses, it runs rover() and returns the mean per-position agreement as the raw signal – uncalibrated, like every signal, so it must pass through a Calibrator before any gate reads it.

Example

>>> sig = AgreementSignal()
>>> round(sig(["the cat sat", "the cat sit", "the bat sat"]), 3)
0.778

class ek.AnlsMetric(*, threshold: float = 0.5)

ANLS / ANLS* (nested-JSON) similarity as a Metric.

Handles a bare string (classic ANLS) or an arbitrarily nested dict/list/tuple/None structure (ANLS*); the backend dispatches on type.

Parameters:

threshold – ANLS zeroes a per-leaf similarity below this value before averaging (the classic ANLS 0.5 cut tolerates minor OCR/spelling noise). anls_star exposes no parameter for this – it reads a class constant ANLSTree.THRESHOLD (default 0.5) – so the metric applies the threshold by setting that constant around the call and restoring it. Caveat: this mutates a process-global, so it is not thread-safe; concurrent AnlsMetric calls with different thresholds in the same process can interfere. Defaults to 0.5.

aggregate(scores: Sequence[Score]) → float

Corpus ANLS = mean of per-item ANLS (each is already length-normalized).

Unlike CER/WER (which accumulate global edit counts) or field-F1 (micro-TP/ FP/FN), ANLS is defined as the average normalized similarity over the evaluation set – the “A” in ANLS – so the corpus statistic is the plain mean of the per-item scores.

class ek.AnnotatedExtraction(grammar: GraphGrammar, estimates: NodePath, ~ek.base.FieldEstimate]=<factory>)

Layer B: the verification metadata, referencing a frozen Layer-A grammar.

The grammar is held by reference and never mutated; estimates maps a NodePath to the FieldEstimate for that node/field.

class ek.Calibrator(*args, **kwargs)

Maps a raw scalar signal to a probability after being fit on labels.

class ek.Canonicalizer(steps: Sequence[str | Callable[[str], str]] = (), *, version: str = '', audit_marks: bool = False)

An ordered, named, versioned pipeline of str -> str steps.

Parameters:

• steps – Step references – names registered under "normalizers" or plain callables. Applied left to right.

• version – Override the auto-derived version string (the +-joined names).

• audit_marks – If true, warn when a step would run on text carrying combining marks (the cross-script damage guard).

property steps: list

The resolved step names, in order.

property version: str

Stable identifier of this pipeline (persist it next to a gold set).

class ek.Cell(text: str = '', rowspan: int = 1, colspan: int = 1)

One logical table cell: its text plus its row/column span (>=1 each).

class ek.ConformalGate(alpha: float = 0.1, nonconformity: ~typing.Callable[[float], float] = <function ConformalGate.<lambda>>, q: float = inf, _cal: ~typing.List[float] = <factory>)

Split-conformal accept/flag with a finite-sample marginal guarantee.

Fit on (calibrated_prob, correct): the nonconformity scores of the correct calibration items set a threshold q such that, among truly-correct test items exchangeable with them, at most alpha are flagged. The guarantee is marginal, not per-field-type – use GroupConformalGate for that.

Parameters:

• alpha – Target false-flag rate on correct items (e.g. 0.1).

• nonconformity – Maps calibrated confidence -> nonconformity (default 1 - confidence: low confidence is nonconforming).

fit(probs: Sequence[float], correct: Sequence[bool]) → ConformalGate

Fit q from the nonconformity scores of the correct calibration items.

Non-finite confidences are dropped (a single NaN would otherwise sort to the end and corrupt the quantile, silently breaking the coverage guarantee).

p_value(confidence: float) → float

Per-instance conformal p-value (small = atypical of correct items).

class ek.Corrector(*args, **kwargs)

A Validator that may also CORRECT.

Same callable shape as a Validator, but it can emit a severity=CORRECT Finding whose suggestion is the proposed replacement value. The layer attribute names its place on the cheap->expensive spine.

class ek.CostSensitiveGate(rho: float = 1.0, block_threshold: float = 0.0, accept_threshold: float | None = None)

Accept/flag/block from the cost ratio rho = c_FN / c_FP on calibrated p.

Accept when the expected cost of accepting is below that of routing to a human. Accepting risks an undetected error: cost (1 - p) * c_FN. Flagging a value that was actually correct wastes a review: cost p * c_FP (the false-positive cost is incurred only with probability p, not unconditionally). So accept iff (1 - p) * c_FN <= p * c_FP, i.e. the Bayes-optimal threshold p >= c_FN / (c_FN + c_FP) = rho / (1 + rho). A larger rho (misses much costlier than reviews) raises the bar to auto-accept (tau -> 1 as rho -> inf); rho = 1 gives the symmetric tau = 0.5. block_threshold adds an optional hard-fail band at the bottom.

Parameters:

• rho – Cost ratio c_FN / c_FP (>= 0). The single lever; no magic numbers.

• block_threshold – Calibrated probability at or below which to block rather than flag (default 0 -> never block on probability alone; a failed verifier is what forces a block, upstream).

• accept_threshold – Override the derived accept threshold explicitly.

property tau: float

The accept threshold on calibrated probability (derived from rho).

class ek.Decision(value)

Terminal action a selective-prediction gate emits for an extracted value.

class ek.DecisionPolicy(*args, **kwargs)

Turns a (calibrated) confidence into an accept/flag/block Decision.

class ek.EdgeType(name: str, src: str, dst: str, importance: float = 1.0)

A directed relation kind between two NodeType names.

class ek.Episode(task_id: str = '', trajectory: Trajectory = <factory>, output: Any = None, final_state: Any = None, cost: Cost | None = None, success: bool | None = None, run: RunProvenance | None = None, provenance: Provenance | None = None, meta: dict = <factory>)

Layer B: one agent run over one task – the object agent evaluation scores.

The direct analog of AnnotatedExtraction: it rides alongside the frozen Layer-A grammar and carries the run’s metadata. It is scored offline against gold (ek.score() / ek.evaluate()) and estimated online against a judge or a consensus (ek.estimate_quality()) – the same object, both halves.

Parameters:

• task_id – Which task this episode answers (the k-trial grouping key).

• trajectory – The tool calls taken.

• output – The agent’s final user-facing answer.

• final_state – The end state of the world (the DB, the filesystem) – what a state-based oracle actually grades. Success is a final-state check, not a surface-text check.

• cost – The Cost consumed.

• success – Filled in by a checker; None until graded.

• run – The RunProvenance (seed, model, simulator, suite version).

• provenance – Optional core Provenance for click-to-source audit.

• meta – Anything else the bridge captured.

property calls: tuple

The (tool, args) pairs taken, in order.

graded(success: bool) → Episode

A copy with success set (never mutates – grading stays idempotent).

meta is copied too: dataclasses.replace is a shallow copy, so without this the “copy” would share the original’s mutable meta dict.

class ek.FieldEstimate(value: Any, raw_signals: dict = <factory>, confidence: float | None = None, findings: tuple = (), provenance: Provenance | None = None, decision: Decision | None = None)

One extracted value plus all of its reference-free verification metadata.

class ek.FieldMetric(*, canonicalizer=None)

Per-field precision/recall/F1 for two dict records.

Parameters:

canonicalizer – Optional str -> str applied to string values before comparison (e.g. casefold/whitespace folding).

aggregate(scores: Sequence[Score]) → float

Micro-averaged F1 over the corpus (sum TP/FP/FN, then divide).

class ek.FieldSpec(name: str, type: str = 'string', importance: float = 1.0, domain: tuple = (), normalizer: str | None = None)

A leaf attribute of a node: its type, validation domain and cost weight.

Parameters:

• name – Attribute name.

• type – A free-form type tag (e.g. "string", "number", "date"); ek does not impose a type system – the tag drives metric/validator dispatch and is yours to define.

• importance – Per-field cost weight; the lever that makes “two extra digits on a donation amount” outweigh “a misspelled city”. Defaults to 1.0.

• domain – Optional allowed values: an enum tuple, a (lo, hi) range, or a regex string – consumed by validators, never by metrics.

• normalizer – Optional key naming a canonicalizer (resolved from the registry) to apply to this field before comparison.

class ek.Finding(field: str, layer: str, severity: Severity = Severity.FLAG, message: str = '', suggestion: Any | None = None)

One validator observation about a value: a flag, optionally a correction.

A suggestion is present iff the producing layer can correct (only the deterministic canonicalizer and the gated LLM corrector freely do so). The original value is always retained elsewhere; a Finding never mutates.

class ek.GateResult(passed: bool, metric: str, higher_is_better: bool, tolerance: float, aggregate_current: float | None = None, aggregate_baseline: float | None = None, regressions: dict = <factory>)

Outcome of a regression_gate() check.

class ek.GraphGrammar(node_types: Mapping[str, ~ek.base.NodeType]=<factory>, edge_types: Mapping[str, ~ek.base.EdgeType]=<factory>)

The frozen schema SSOT: typed nodes/edges plus their importance weights.

Methods return the cost weight for a referenced type, defaulting to 1.0 for anything not declared, so a partial grammar is always usable.

cost(ref: TypeRef) → float

Importance weight for any TypeRef (the default CostWeight).

edge_cost(name: str) → float

Importance weight of an edge type (1.0 if undeclared).

field_cost(node: str, field_name: str) → float

Importance weight of a field on a node type (1.0 if undeclared).

node_cost(name: str) → float

Importance weight of a node type (1.0 if undeclared).

class ek.GritsMetric(*, structure_only: bool = False)

GriTS-Top / GriTS-Con as a Metric (pure-Python, P/R aware).

Parameters:

structure_only – When True, score cell topology only (GriTS-Top); when False (default), average topology with cell-content similarity (GriTS-Con).

aggregate(scores: Sequence[Score]) → float

Corpus GriTS F1: pool overlap and per-table cell counts, then divide once.

GriTS precision = overlap / |pred cells| and recall = overlap / |gold cells|; the corpus statistic micro-averages by summing the numerators and denominators globally – never a mean of per-table F1s.

class ek.GroupCalibrator(factory: Callable[[], ~typing.Any]=<class 'ek.qe.calibrate.PlattCalibrator'>, by_group: dict = <factory>, pooled: Any = None, kind: str = 'group')

Per-group (Mondrian) calibration: one calibrator per NodeType/FieldSpec.

Distribution-free conditional (per-field-type) coverage is impossible in general (Barber et al. 2019); calibrating separately per group restores it approximately. fit takes a parallel groups sequence; __call__ routes by group key, falling back to a pooled calibrator for unseen groups.

Parameters:

factory – Zero-arg callable producing a fresh per-group calibrator (default PlattCalibrator).

factory

alias of PlattCalibrator

fit(scores: Sequence[float], correct: Sequence[bool], *, groups: Sequence[Any]) → GroupCalibrator

Fit one calibrator per distinct group key, plus a pooled fallback.

to_dict() → dict

Serialize each per-group calibrator (keyed by stringified group) plus the pooled fallback, so a fitted Mondrian calibrator round-trips like the others. Group keys are stringified (they are NodeType/FieldSpec names).

class ek.GroupConformalGate(alpha: float = 0.1, nonconformity: ~typing.Callable[[float], float] = <function GroupConformalGate.<lambda>>, by_group: dict = <factory>, pooled: ~ek.qe.decide.ConformalGate | None = None)

Mondrian (class-conditional) conformal: one ConformalGate per group.

Restores approximate per-field-type validity by calibrating separately per group key (e.g. NodeType/FieldSpec), with a pooled fallback for unseen groups.

class ek.IntrinsicConfidenceSignal(pool: str = 'min', cost_tier: int = 2)

Aggregate an extractor’s per-unit confidences into a field score.

Parameters:

• pool – "min" (weakest unit), "mean", or "geo_mean" over the per-unit confidences (already probabilities in [0, 1]).

• cost_tier – 2 – free (the posteriors are emitted at inference).

Example

>>> IntrinsicConfidenceSignal(pool="min")([0.99, 0.4, 0.95])
0.4

class ek.IsotonicCalibrator(x: List[float] = <factory>, y: List[float] = <factory>, kind: str = 'isotonic')

Isotonic (monotonic non-decreasing) calibration via pool-adjacent-violators.

More flexible than Platt; needs more calibration data and can overfit small sets. Predicts by linear interpolation between fitted points, clipped at the ends.

fit(scores: Sequence[float], correct: Sequence[bool]) → IsotonicCalibrator

Fit the monotonic step function to (score, correct) pairs.

class ek.LogprobSignal(aggregator: Any = <function geo_mean>, alpha: float = 0.6, cost_tier: int = 2)

Aggregate token log-probabilities into one field score (a Signal).

Parameters:

• aggregator – A logprob aggregator (name registered under aggregators or a callable). Defaults to geo_mean().

• alpha – Length penalty passed to length_normalized() only.

• cost_tier – 2 – free intrinsic signal (already computed at inference).

aggregator() → float

Geometric mean of token probabilities, exp(mean log p) (perplexity-inverse).

class ek.MatchScheme(value)

The (backend, scheme) contract a span/slot F1 is computed under.

The two backends legitimately disagree (seqeval ignores other-type tags at the tag level; nervaluate includes them), so the scheme is always explicit. The SEQEVAL_* members consume BIO/IOBES tag sequences; the others consume SemEval-2013 span lists via nervaluate.

EXACT = 'exact'

exact span boundaries, type ignored.

Type:

nervaluate exact

PARTIAL = 'partial'

boundary overlap (half credit), type ignored.

Type:

nervaluate partial

SEQEVAL_CONLL = 'seqeval_conll'

seqeval, CoNLL conlleval-compatible entity F1 (the default seqeval mode).

SEQEVAL_STRICT = 'seqeval_strict'

span boundaries AND type must match, keyed to a scheme.

Type:

seqeval strict mode

STRICT = 'strict'

exact span boundaries AND correct type.

Type:

nervaluate SemEval-2013 strict

TYPE = 'type'

correct type with at least some span overlap.

Type:

nervaluate type

class ek.Metric(*args, **kwargs)

Compares one prediction to one gold reference, returning a Score.

class ek.NodePath(node_id: str, node_type: str, field: str | None = None)

Addresses a node (or one of its fields) in an extracted graph; keys Layer B.

class ek.NodeType(name: str, fields: Mapping[str, ~ek.base.FieldSpec]=<factory>, importance: float = 1.0)

A node kind in the typed graph (e.g. an “invoice” or a “line item”).

class ek.PlattCalibrator(a: float = 1.0, b: float = 0.0, max_iter: int = 100, kind: str = 'platt')

Platt scaling: sigmoid(a * score + b), fit by Newton/IRLS on labels.

The default calibrator: works on any scalar (aggregated confidence/logprob), no logits required. Uses Platt’s target smoothing so it does not overfit small calibration sets.

Parameters:

max_iter – Newton iterations (converges in a handful for a 2-parameter model).

fit(scores: Sequence[float], correct: Sequence[bool]) → PlattCalibrator

Fit a, b to maximise the likelihood of correct given scores.

class ek.Provenance(engine: str = '', source_span: tuple | None = None, bbox: Any = None, raw_transcripts: Sequence[str] = ())

Where an extracted value came from, for audit and click-to-source review.

bbox is intentionally typed Any so ek core need not import any OCR package; the OCR instance stores normalized [0, 1] top-left-origin geometry.

class ek.QualityReport(calibrated_confidence: float | None = None, decision: Decision | None = None, findings: tuple = (), raw_signals: dict = <factory>, provenance: dict = <factory>, per_field: Mapping | None = None)

The result of a reference-free quality estimate for one extraction.

For a single value the scalar fields carry the verdict. For a whole AnnotatedExtraction, per_field maps each NodePath to its updated FieldEstimate (confidence/findings/decision filled in), and the scalar fields summarize: calibrated_confidence is the weakest field, and decision the most severe (block > flag > accept).

class ek.Report(metric: str = '', aggregate: float | None = None, n: int = 0, scores: list = <factory>, per_slice: dict = <factory>, detail: dict = <factory>)

Aggregate of many Score s over a corpus, with optional per-slice cuts.

aggregate is the corpus-level headline (computed by the metric’s own aggregator – e.g. globally accumulated WER – not a naive mean of per-item values). per_slice maps a slice label to its own aggregate.

class ek.RiskControlGate(target: float = 0.05, loss: ~typing.Callable[[bool], float] = <function RiskControlGate.<lambda>>, loss_bound: float = 1.0, lam: float = 1.0)

Conformal Risk Control: accept p >= lambda bounding the accepted-error rate.

Picks the smallest threshold lambda (the most coverage) whose finite-sample risk bound is at or below target. The controlled quantity is E[ loss(item) * 1(accepted) ] – with the default 0/1 loss, the population rate of accepted-yet-wrong items.

Parameters:

• target – Upper bound on the accepted-error rate (e.g. 0.05).

• loss – correct -> loss in [0, loss_bound]; default 0 if correct else 1. Any monotone loss bounded by loss_bound works.

• loss_bound – The loss’s upper bound B – the CRC finite-sample slack is +B/(n+1). Must match loss’s range or the guarantee is invalid; keep the default 1.0 for the 0/1 loss.

fit(probs: Sequence[float], correct: Sequence[bool]) → RiskControlGate

Find the smallest accept threshold whose risk bound is <= target.

class ek.RoverConsensus(tokens: List[str] = <factory>, slots: List[RoverSlot] = <factory>, agreement: List[float] = <factory>, n_engines: int = 0)

The result of a ROVER pass over N hypotheses.

property mean_agreement: float

Mean per-position agreement over consensus tokens, a reference-free confidence in [0, 1]: 1.0 means every engine agreed at every emitted position; lower means at least one engine dissented.

An empty consensus (no token survived voting – NULL won every slot) means the engines agreed on nothing, so this returns 0.0 when 2+ engines were fused, NOT a misleading 1.0 (which would feed a maximal raw confidence to the calibrator/gate). A single engine – or none – is vacuously 1.0.

property text: str

The consensus transcription (space-joined winning tokens).

class ek.Score(value: float, precision: float | None = None, recall: float | None = None, f1: float | None = None, metric: str = '', detail: dict = <factory>)

The result of comparing one prediction against one reference.

value is the headline number (higher is better unless a metric documents otherwise). The optional decomposition fields let structured metrics carry precision/recall/F1, an alignment, and raw counts (e.g. edit operations) so that corpus aggregation can be done correctly (e.g. global WER accumulation) rather than by averaging per-item scores. float(score) returns value.

class ek.Severity(value)

Whether a Finding merely flags or can also correct a value.

class ek.Signal(*args, **kwargs)

A reference-free quality signal: extractor output -> raw scalar score.

The cost_tier attribute (lower is cheaper) lets an escalation policy run cheap signals first and pay for expensive ones only on residual uncertainty.

class ek.SpanF1Metric(scheme: MatchScheme | None = None, *, tagging_scheme: str = 'IOB2')

Entity/slot precision/recall/F1 under an explicit MatchScheme.

Parameters:

• scheme – REQUIRED. The (backend, scheme) contract – there is deliberately no default, because seqeval and nervaluate disagree and a bare F1 is ambiguous (misc/docs/ek_02 §2.2). Pass a MatchScheme.

• tagging_scheme – For MatchScheme.SEQEVAL_STRICT only: the BIO/IOBES tagging scheme name (e.g. "IOB2", "IOBES", "BILOU") that strict-mode entity decoding is keyed to. Defaults to "IOB2".

Raises:

TypeError – if scheme is omitted (the whole point of this metric).

aggregate(scores: Sequence[Score]) → float

Micro-averaged F1 over the corpus (sum the raw counts, then divide once).

seqeval schemes carry TP/FP/FN; nervaluate schemes carry COR/INC/PAR/MIS/SPU (PAR worth half). Both micro-average by summing counts globally – never a mean of per-document F1s. The two backends’ counts are not commensurable, so a corpus mixing them is rejected rather than silently dropping one.

class ek.StringMetric(mode: str = 'cer', *, canonicalizer=None)

CER (mode='cer') or WER (mode='wer') as a Metric.

Parameters:

• mode – "cer" (character) or "wer" (word) error rate.

• canonicalizer – Optional str -> str applied to both sides before scoring (see ek.canonicalize). Usually supplied by the facade instead.

aggregate(scores: Sequence[Score]) → float

Corpus error rate = total edits / total reference length (the correct way).

class ek.Table(rows: Sequence[Sequence[Cell]] = ())

A table as an ordered list of rows of Cell s (the logical cells).

rows[i] is the i-th row’s logical cells in reading order; spans are carried on each Cell. Use from_html() / from_grid() to build one, and as_grid() for the dense 2-D occupancy view GriTS scores.

as_grid() → list[list[Cell | None]]

Expand spans into a dense 2-D grid; a spanned position repeats its cell.

Each physical grid position holds the (shared) Cell covering it, so two tables with the same topology have grids of the same shape – the view GriTS aligns position-by-position.

classmethod coerce(obj: Any) → Table

Coerce a Table, an HTML string, or a 2-D grid into a Table.

classmethod from_grid(grid: Sequence[Sequence[Any]]) → Table

Build a Table from a 2-D grid of cell texts (None = spanned-over).

classmethod from_html(html: str) → Table

Parse an HTML <table> (with colspan/rowspan) into a Table.

class ek.TaskSpec(task_id: str, input: Any = None, gold: Any = None, value: float = 1.0, tools: Sequence[ToolSpec] = (), slice: str | None = None)

One task in a suite: its input, its oracle, its value, and its allowed tools.

Parameters:

• task_id – Stable identifier (the grouping key for k-trial reliability).

• input – Whatever the agent under test is called with.

• gold – The reference the checker grades against (a goal state, an expected answer, a gold trajectory – the checker decides how to read it).

• value – Task-level cost weight: how much a completed task is worth. The task-level analog of FieldSpec.importance; feeds value-weighted cost accounting.

• tools – The tools this task permits (the Layer-A grammar for the task).

• slice – Optional slice label (domain, difficulty, language) – per-slice cuts are mandatory, not optional (a low pass^k concentrated in one hard slice is a different problem from uniform flakiness).

grammar() → GraphGrammar

The Layer-A grammar for this task’s tool set.

class ek.TedsMetric(*, structure_only: bool = False)

TEDS / TEDS-Struct as a Metric (clean-room on apted).

Parameters:

structure_only – When True, ignore cell text and score structure alone (TEDS-Struct), isolating table-structure accuracy from OCR noise. Defaults to False (full TEDS: structure + content).

aggregate(scores: Sequence[Score]) → float

Corpus TEDS = mean of per-table TEDS (the PubTabNet/OmniDocBench convention). Each per-table TEDS is already node-count-normalized, so the reported corpus statistic is the average, not a pooled edit-rate. The pooled 1 - sum(dist)/sum(nodes) is available from the per-item detail if a size-weighted variant is wanted instead.

class ek.TemperatureCalibrator(T: float = 1.0, kind: str = 'temperature', t_min: float = 0.05, t_max: float = 10.0, max_iter: int = 60)

Temperature scaling: sigmoid(logit / T) with one T fit on a holdout.

Use only when you have logits: __call__ expects a logit, not a probability. T > 1 softens overconfidence; the argmax is unchanged.

fit(logits: Sequence[float], correct: Sequence[bool]) → TemperatureCalibrator

Fit T by minimising NLL with a bounded 1-D search.

t_min: float = 0.05

Golden-section search bounds and iteration count (config, not magic numbers).

class ek.ToolSpec(name: str, params: Mapping[str, ~ek.base.FieldSpec]=<factory>, importance: float | None = None, destructive: bool = False)

One tool the agent may call: its argument schema and its error costs.

Parameters:

• name – The tool/function name (the AST-match ground truth).

• params – Argument name -> FieldSpec. Each spec’s importance is the cost of getting that argument wrong, and its domain feeds validators – the same contract as an extracted field.

• importance – Cost weight of the call itself (a missing or spurious call). Defaults to DESTRUCTIVE_WEIGHT when destructive is set, else 1.0.

• destructive – Whether calling this tool mutates the world irreversibly (a refund, a delete, a send). Drives the default weight and the safety validators.

to_node_type() → NodeType

Render this tool as a Layer-A NodeType (args as fields).

property weight: float

The call-level cost weight (explicit, else destructive-aware default).

class ek.TypeRef(kind: str, name: str, field: str | None = None)

A reference into the schema, so an injected cost function can read weights.

kind is one of "node", "edge" or "field"; for a field, name is the node-type name and field the field name.

class ek.TypedEdge(src: str, dst: str, edge_type: str = '')

A directed, typed relation between two node ids.

class ek.TypedGraph(nodes: Sequence[TypedNode] = (), edges: Sequence[TypedEdge] = ())

A typed graph: typed nodes plus directed typed edges between them.

to_networkx()

Build a networkx.DiGraph carrying type/fields as node/edge attrs.

class ek.TypedGraphMetric(grammar: GraphGrammar | None = None, weights: Callable[[GraphGrammar, TypeRef], float] | None = None, *, canonicalizer=None, timeout: float = 10.0, max_nodes: int = 60)

Cost-weighted, type-aware typed-graph edit distance as a Metric.

Parameters:

• grammar – Layer-A GraphGrammar supplying cost weights (may also be passed per-call to __call__).

• weights – Optional CostWeight overriding the schema weights.

• canonicalizer – Optional str -> str applied to field values before the equality check (the facade passes normalize= through here); a schema’s per-field FieldSpec.normalizer takes precedence for the fields it names.

• timeout – Seconds budget for the (NP-hard) GED search; on timeout the best cost found so far is used (an approximation). Default 10.0.

aggregate(scores: Sequence[Score]) → float

Corpus normalized distance = total raw distance / total max-distance.

class ek.TypedNode(node_id: str, node_type: str, fields: Mapping[str, ~typing.Any]=<factory>)

A node in a typed graph: an id, a node-type tag, and its field values.

class ek.ValidationResult(original: Any, value: Any, findings: tuple[Finding, ...] = ())

The outcome of running a value through a validation_pipeline().

value is the final value after any applied corrections; original is the input; findings is the full audit trail (FLAG and CORRECT), in layer order.

property clean: bool

No findings at all – the value passed every layer untouched.

property corrected: bool

Whether any correction changed the value.

property flagged: bool

Whether any layer raised a FLAG (something a human should look at).

class ek.Validator(*args, **kwargs)

Reference-free check on a value, yielding zero or more Finding s.

class ek.VerifierSignal(validators: ~typing.List[~typing.Callable[[...], ~typing.Iterable[~ek.base.Finding]]] = <factory>, cost_tier: int = 1)

Deterministic verifier evidence as a reference-free Signal.

Cost tier 1 – the free, first-to-run layer. Runs a list of Validator s on a value and returns the fraction that passed in [0, 1] as the raw signal (1.0 = every check passed). It also exposes the Finding s it produced via findings(), so the same object feeds both the score path (-> calibrate) and the audit path (-> review). Like every signal it is uncalibrated: a verifier failing is strong evidence, not a probability, so a Calibrator still runs before any gate.

Example

>>> v = VerifierSignal([checksum_validator("luhn")])
>>> v("79927398713")          # valid -> all checks pass
1.0
>>> v("79927398710")          # invalid -> the one check fails
0.0

findings(value: Any, *, spec: FieldSpec | None = None) → List[Finding]

All findings the validators produce for value (empty == all clear).

ek.app_folder() → Path

The root ek data folder (e.g. ~/.local/share/ek).

ek.benford_findings(numbers: Iterable[Any], *, field: str = '', tol: float = 0.15, min_n: int = 30, layer: str = 'anomaly') → list

FLAG a numeric field whose first-digit distribution deviates from Benford’s law.

A reference-free anomaly check for naturally-occurring magnitudes (amounts, populations, counts): the first significant digit should follow P(d)=log10(1+1/d). A large deviation flags fabricated or systematically-wrong data. Corpus-level (takes the whole column of values) and FLAG-only – it routes a field to review, never auto-edits, and can false-positive on legitimately bounded/skewed fields. Skipped (returns []) below min_n usable values, since the law is asymptotic.

ek.cache_this(func: Callable[[Any], VT] = None, *, cache: str | MutableMapping[KT, VT] | None = None, key: KT | Callable[[str], KT] | None = None, pre_cache: bool | MutableMapping = False, as_property: bool | None = None, ignore: str | list[str] | None = None, serialize: Callable[[Any], Any] | None = None, deserialize: Callable[[Any], Any] | None = None)

Unified caching decorator for properties and methods with persistent storage support.

cache_this extends the capabilities of Python’s built-in functools.cached_property and functools.lru_cache by providing:

• Persistent caching: Store cached values in files, databases, or any MutableMapping

• Flexible cache backends: Use instance attributes, external stores, or cache factories

• Smart key generation: Automatic argument-based keys for methods with parameter filtering

• Serialization support: Custom serialize/deserialize functions for complex data

• Auto-detection: Automatically chooses property vs method caching based on signature

• No LRU eviction: Unlike lru_cache, values persist until explicitly removed

Unlike functools.cached_property (properties only) and lru_cache (memory-only with eviction), cache_this provides a unified interface for both use cases with persistent storage options.

Parameters:

• func – The function to be decorated (usually left empty).

• cache –

  The cache storage. Can be: - A MutableMapping instance (shared across instances) - A string naming an instance attribute containing a MutableMapping - A callable taking (instance) and returning a MutableMapping

  This enables instance-specific caching, e.g.: cache=lambda self: Files(f’/cache/{self.user_id}/’)

• key – For properties: the key to store the cache value, can be a callable that will be applied to the method name to make a key, or an explicit string. For methods: a callable that takes (self, *args, **kwargs) and returns a cache key.

• pre_cache – Default is False. If True, adds an in-memory cache to the method to (also) cache the results in memory. If a MutableMapping is given, it will be used as the pre-cache. This is useful when you want a persistent cache but also want to speed up access to the method in the same session.

• as_property – If True, force use of CachedProperty. If False, force use of CachedMethod. If None (default), auto-detect based on function signature.

• ignore – Parameter name(s) to exclude from cache key computation. Can be a string (single parameter) or list of strings (multiple parameters). Commonly used to ignore ‘self’ or parameters like ‘verbose’ that don’t affect the result.

• serialize – Optional function to serialize values before caching. Example: serialize=pickle.dumps for binary file storage

• deserialize – Optional function to deserialize cached values. Example: deserialize=pickle.loads

Returns:

The decorated function.

## Comprehensive Example

Here’s a complete example showcasing all major features of cache_this:

>>> import tempfile
>>> import os
>>> from pathlib import Path
>>>
>>> class DataProcessor:
...     def __init__(self, user_id="user123"):
...         self.user_id = user_id
...         self.memory_cache = {}  # In-memory cache
...         self.call_counts = {}   # Track function calls for demo
...
...     # 1. Basic property caching (like functools.cached_property)
...     @cache_this
...     def basic_property(self):
...         '''Cached in instance.__dict__ by default'''
...         self.call_counts['basic_property'] = self.call_counts.get('basic_property', 0) + 1
...         return f"computed_value_{self.call_counts['basic_property']}"
...
...     # 2. Property with custom cache and key
...     @cache_this(cache='memory_cache', key='custom_prop_key')
...     def custom_cached_property(self):
...         '''Cached in instance.memory_cache with custom key'''
...         self.call_counts['custom_cached_property'] = self.call_counts.get('custom_cached_property', 0) + 1
...         return f"custom_value_{self.call_counts['custom_cached_property']}"
...
...     # 3. Method caching with argument-based keys
...     @cache_this(cache='memory_cache')
...     def compute_result(self, x, y, mode='fast'):
...         '''Cached based on arguments (x, y, mode)'''
...         key = ('compute_result', x, y, mode)
...         self.call_counts[key] = self.call_counts.get(key, 0) + 1
...         return x * y * (2 if mode == 'fast' else 3)
...
...     # 4. Method caching with ignored parameters
...     @cache_this(cache='memory_cache', ignore={'verbose', 'debug'})
...     def process_data(self, data, algorithm='default', verbose=False, debug=False):
...         '''Cache ignores verbose and debug parameters'''
...         key = ('process_data', tuple(data), algorithm)
...         self.call_counts[key] = self.call_counts.get(key, 0) + 1
...         if verbose: print(f"Processing {data} with {algorithm}")
...         return sum(data) * (2 if algorithm == 'default' else 3)
...
...     # 5. Instance-specific cache factory
...     @cache_this(cache=lambda self: {f'{self.user_id}_cache': {}}.get(f'{self.user_id}_cache'))
...     def user_specific_computation(self, value):
...         '''Each instance gets its own cache based on user_id'''
...         key = ('user_specific_computation', value)
...         self.call_counts[key] = self.call_counts.get(key, 0) + 1
...         return value ** 2

Now let’s test all the features:

>>> processor = DataProcessor("alice")
>>>
>>> # Test basic property caching
>>> result1 = processor.basic_property
>>> result2 = processor.basic_property  # Should use cache
>>> assert result1 == result2 == "computed_value_1"
>>> assert 'basic_property' in processor.__dict__  # Cached in instance dict
>>>
>>> # Test custom cache and key
>>> result1 = processor.custom_cached_property
>>> result2 = processor.custom_cached_property  # Should use cache
>>> assert result1 == result2 == "custom_value_1"
>>> assert 'custom_prop_key' in processor.memory_cache
>>>
>>> # Test method caching with arguments
>>> result1 = processor.compute_result(3, 4, 'fast')
>>> result2 = processor.compute_result(3, 4, 'fast')  # Should use cache
>>> result3 = processor.compute_result(3, 4, 'slow')  # Different args, new computation
>>> assert result1 == result2 == 24  # 3 * 4 * 2
>>> assert result3 == 36  # 3 * 4 * 3
>>>
>>> # Test parameter ignoring
>>> result1 = processor.process_data([1, 2, 3], verbose=True)
Processing [1, 2, 3] with default
>>> result2 = processor.process_data([1, 2, 3], verbose=False)  # Should use same cache
>>> result3 = processor.process_data([1, 2, 3], debug=True)     # Should use same cache
>>> assert result1 == result2 == result3 == 12  # sum([1,2,3]) * 2
>>>
>>> # Test instance-specific caching
>>> result1 = processor.user_specific_computation(5)
>>> result2 = processor.user_specific_computation(5)  # Should use cache
>>> assert result1 == result2 == 25  # 5 ** 2
>>>
>>> # Different instance should have separate cache
>>> processor2 = DataProcessor("bob")
>>> result3 = processor2.user_specific_computation(5)  # Fresh computation
>>> assert result3 == 25

Used with no arguments, cache_this will cache just as the builtin cached_property does – in the instance’s __dict__ attribute.

>>> class SameAsCachedProperty:
...     @cache_this
...     def foo(self):
...         print("In SameAsCachedProperty.foo...")
...         return 42
...
>>> obj = SameAsCachedProperty()
>>> obj.__dict__  # the cache is empty
{}
>>> obj.foo  # when we access foo, it's computed and returned...
In SameAsCachedProperty.foo...
42
>>> obj.__dict__  # ... but also cached
{'foo': 42}
>>> obj.foo  # so that the next time we access foo, it's returned from the cache.
42

Not that if you specify cache=False, you get a property that is computed every time it’s accessed:

>>> class NoCache:
...     @cache_this(cache=False)
...     def foo(self):
...         print("In NoCache.foo...")
...         return 42
...
>>> obj = NoCache()
>>> obj.foo
In NoCache.foo...
42
>>> obj.foo
In NoCache.foo...
42

Specify the cache as a dictionary that lives outside the instance:

>>> external_cache = {}
>>>
>>> class CacheWithExternalMapping:
...     @cache_this(cache=external_cache)
...     def foo(self):
...         print("In CacheWithExternalMapping.foo...")
...         return 42
...
>>> obj = CacheWithExternalMapping()
>>> external_cache
{}
>>> obj.foo
In CacheWithExternalMapping.foo...
42
>>> external_cache
{'foo': 42}
>>> obj.foo
42

Specify the cache as an attribute of the instance, and an explicit key:

>>> class WithCacheInInstanceAttribute:
...
...     def __init__(self):
...         self.my_cache = {}
...
...     @cache_this(cache='my_cache', key='key_for_foo')
...     def foo(self):
...         print("In WithCacheInInstanceAttribute.foo...")
...         return 42
...
>>> obj = WithCacheInInstanceAttribute()
>>> obj.my_cache
{}
>>> obj.foo
In WithCacheInInstanceAttribute.foo...
42
>>> obj.my_cache
{'key_for_foo': 42}
>>> obj.foo
42

Now let’s see a more involved example that exhibits how cache_this would be used in real life. Note two things in the example below.

First, that we use functools.partial to fix the parameters of our cache_this. This enables us to reuse the same cache_this in multiple places without all the verbosity. We fix that the cache is the attribute cache of the instance, and that the key is a function that will be computed from the name of the method adding a ‘.pkl’ extension to it.

Secondly, we use the ValueCodecs from dol to provide a pickle codec for storying values. The backend store used here is a dictionary, so we don’t really need a codec to store values, but in real life you would use a persistent storage that would require a codec, such as files or a database.

Thirdly, we’ll use a pre_cache to store the values in a different cache “before” (setting and getting) them in the main cache. This is useful, for instance, when you want to persist the values (in the main cache), but keep them in memory for faster access in the same session (the pre-cache, a dict() instance usually). It can also be used to store and use things locally (pre-cache) while sharing them with others by storing them in a remote store (main cache).

Finally, we’ll use a dict that logs any setting and getting of values to show how the caches are being used.

>>> from dol import cache_this
>>>
>>> from functools import partial
>>> from dol import ValueCodecs
>>> from collections import UserDict
>>>
>>>
>>> class LoggedCache(UserDict):
...     name = 'cache'
...
...     def __setitem__(self, key, value):
...         print(f"In {self.name}: setting {key} to {value}")
...         return super().__setitem__(key, value)
...
...     def __getitem__(self, key):
...         print(f"In {self.name}: getting value of {key}")
...         return super().__getitem__(key)
...
>>>
>>> class CacheA(LoggedCache):
...     name = 'CacheA'
...
>>>
>>> class CacheB(LoggedCache):
...     name = 'CacheB'
...
>>>
>>> cache_with_pickle = partial(
...     cache_this,
...     cache='cache',  # the cache can be found on the instance attribute `cache`
...     key=lambda x: f"{x}.pkl",  # the key is the method name with a '.pkl' extension
...     pre_cache=CacheB(),
... )
>>>
>>>
>>> class PickleCached:
...     def __init__(self, backend_store_factory=CacheA):
...         # usually this would be a mapping interface to persistent storage:
...         self._backend_store = backend_store_factory()
...         self.cache = ValueCodecs.default.pickle(self._backend_store)
...
...     @cache_with_pickle
...     def foo(self):
...         print("In PickleCached.foo...")
...         return 42
...

>>> obj = PickleCached()
>>> list(obj.cache)
[]

>>> obj.foo
In CacheA: getting value of foo.pkl
In CacheA: getting value of foo.pkl
In PickleCached.foo...
In CacheA: setting foo.pkl to b'\x80\x04K*.'
42
>>> obj.foo
In CacheA: getting value of foo.pkl
In CacheB: setting foo.pkl to 42
42

As usual, it’s because the cache now holds something that has to do with foo:

>>> list(obj.cache)
['foo.pkl']
>>> # == ['foo.pkl']

The value of ‘foo.pkl’ is indeed 42:

>>> obj.cache['foo.pkl']
In CacheA: getting value of foo.pkl
42

But note that the actual way it’s stored in the _backend_store is as pickle bytes:

>>> obj._backend_store['foo.pkl']
In CacheA: getting value of foo.pkl
b'\x80\x04K*.'
>>> # == b'\x80\x04K*.'

ek.canonicalize_corrector(normalize: Any, *, field_name: str = '', layer: str = 'canonicalize') → Corrector

L0: fold a value to canonical form, emitting a CORRECT finding when it changes.

normalize is anything ek.canonicalize.resolve_canonicalizer() accepts (a registered name, a callable, a step list). The narrowest, safest corrector – run it first so later layers compare canonical forms.

ek.check_requirements(*, engine: str | None = None, extra: str | None = None) → dict

Report what an OCR engine or feature extra needs, without raising.

For an OCR engine and an installed ocracy, defers to ocracy’s own doctor (which knows each backend’s pip package, system binaries and credentials). Otherwise returns a generic hint.

Returns:

A dict with at least {"ok": bool, "hint": str}.

ek.checksum_validator(kind: str = 'luhn', *, field_name: str = '', layer: str = 'checksum') → Callable[[...], Iterable[Finding]]

A Validator that flags a value failing a named checksum.

ek.cohen_kappa(rater_a: Sequence, rater_b: Sequence) → float

Cohen’s kappa for two raters on nominal labels (chance-corrected agreement).

Use only for exactly two raters, nominal labels, complete data; prefer krippendorff_alpha() for the general case (missing data, any measurement level).

ek.cross_field_validator(predicate: Callable[[Any], bool], *, message: str, fields: Sequence[str] = (), layer: str = 'cross_field') → Callable[[...], Iterable[Finding]]

FLAG a record when predicate(record) is False (a general cross-field check).

Skips silently when predicate raises on a partial record (a missing field is not itself a cross-field violation).

ek.default_canonicalizer() → Canonicalizer

A conservative default: NFC -> casefold -> collapse whitespace.

ek.default_cost_weight(grammar: GraphGrammar, ref: TypeRef) → float

The default CostWeight: read the importance weight off the schema.

ek.engine_yields_tables(engine: str) → bool

Whether engine’s capability profile claims scoreable table structure.

A convenience over ek.ocr.profile(): returns the engine profile’s tables flag. ek treats the profile as a prior (Azure DI, AWS Textract, PaddleOCR and the markdown engines are expected to yield scoreable tables; Google Vision is not), to be verified empirically by actually running table_from_ocr(). Use it to skip table scoring for engines that cannot emit cells in the first place.

ek.enum_validator(members: Sequence[Any], *, field_name: str = '', layer: str = 'enum') → Callable[[...], Iterable[Finding]]

A Validator that flags a value not in an allowed set.

ek.estimate_quality(extraction: Any, *, sources: Iterable = (), signals: Iterable = (), calibrator=None, validators: Iterable = (), policy=None, agreement: bool = True, assume_calibrated: bool = False) → QualityReport

Estimate the quality of an extraction with no gold reference.

Composes the strict signal -> calibrate -> validate -> decide pipeline (misc/docs/ek_03) over a single value, a FieldEstimate, or a whole AnnotatedExtraction (scored per field, with per-field specs feeding the validators and the node type as the Mondrian group key). The simple call estimate_quality(value) Just Works; every stage is injectable.

Parameters:

• extraction – A raw value, a FieldEstimate, or an AnnotatedExtraction.

• sources – Additional hypotheses of the same content (strings or OcrResult-shaped objects) to fuse with ROVER – their mean agreement becomes a raw_signals["agreement"] entry (the flagship online signal).

• signals – Explicit Signal callables target -> float | Mapping producing further raw signals.

• calibrator – A Calibrator mapping raw score -> probability. Calibration is non-optional before gating (Hard Rule 1): passing a policy with no calibrator and assume_calibrated false raises – a raw, uncalibrated score must never reach a DecisionPolicy.

• validators – Validator callables yielding findings. A flat iterable runs on every field (use spec-driven validators like schema_validator()). For per-field scoping pass a Mapping keyed by field name or node type ("*" runs on all).

• policy – A DecisionPolicy producing accept/flag/block from the calibrated confidence.

• agreement – Auto-run ROVER over sources when any are given (default true).

• assume_calibrated – Treat the incoming confidence as already calibrated (silences the uncalibrated-gating warning).

ek.evaluate(cases: Iterable, *, metric: None | str | Metric = None, grammar: GraphGrammar | None = None, normalize: Any = None, weights: Any = None) → Report

Aggregate many comparisons into a Report (corpus level).

Parameters:

• cases – Iterable of (pred, gold) or (pred, gold, slice_label) tuples.

• metric – As in score() (resolved once from the first case’s types).

• grammar – Optional Layer-A grammar passed to every comparison.

• normalize – Optional canonicalizer applied before every comparison.

Returns:

A Report whose aggregate is computed by the metric’s own aggregator (e.g. globally accumulated CER/WER, micro-F1) – never a naive mean.

ek.evaluate_store(predict: Callable[[Any], Any], gold: Mapping, *, metric: str | None = None, grammar: Any = None, normalize: Any = None, weights: Any = None, input_key: str = 'input', reference_key: str = 'reference', slice_key: str = 'slice', persist: bool = False, run_id: str | None = None, rootdir: str | None = None)

Run predict over a gold store and score it per slice.

Parameters:

• predict – input -> prediction callable (the system under test).

• gold – Mapping key -> {input, reference, [slice], ...} (a dict or a ek gold store).

• metric – Metric name/callable (defaults to type-dispatch on the first case).

• grammar/normalize/weights – forwarded to scoring.

• input_key/reference_key/slice_key – field names within each gold record.

• persist/run_id/rootdir – persist the run to the results/runs stores.

Returns:

A Report; detail['per_item'] maps each key to its prediction, reference, slice, and score.

ek.expected_calibration_error(probs: Sequence[float], correct: Sequence[bool], *, n_bins: int = 10) → float

Expected Calibration Error: weighted mean gap between confidence and accuracy.

Bins predictions by confidence into n_bins equal-width bins and averages |mean_confidence - accuracy| weighted by bin population. 0 is perfect. Non-finite probs are skipped; out-of-range probs are clamped into [0, 1].

ek.get(namespace: str, name: str) → Any

Resolve a registered strategy by name (discovering entry points first).

Raises:

KeyError – with the list of available names, if name is unknown.

ek.has_table_structure(ocr_result: Any, *, parser: None | str | Callable[[Any], Any] = None) → bool

Whether table_from_ocr() recovers a non-empty table from ocr_result.

A thin predicate over table_from_ocr() (same parser semantics): true iff a table with at least one cell can be recovered. Use it to route an OCR result to the table metrics vs the text-only CER/WER fallback.

ek.iban_check(iban: Any) → bool

IBAN validity via the ISO 13616 / ISO 7064 mod-97 rule (must equal 1).

ek.isbn_check(value: Any) → bool

Validate an ISBN-10 or ISBN-13 by length (digits/hyphens/spaces ignored).

ek.json_store(kind: str, *, rootdir: str | None = None) → MutableMapping

A JSON MutableMapping for one kind of artifact.

Keys are plain names (no extension); values are JSON-serializable objects.

ek.krippendorff_alpha(reliability_data: Sequence[Sequence], *, level: str = 'nominal') → float

Krippendorff’s alpha – the general inter-annotator agreement coefficient.

Pure-Python (no dependency, permissive core): any number of raters, any measurement level ("nominal", "ordinal", "interval", "ratio"), and missing data (use None or float('nan') for an unlabelled cell). reliability_data is one row per rater, one column per item. Alpha is 1.0 for perfect agreement, 0.0 at chance, and goes negative for systematic disagreement. Computed via the coincidence-matrix method (Krippendorff 2011): observed vs expected disagreement under the level-specific difference metric.

Example

>>> data = [[1, 1, 1, 1], [1, 1, 1, 1]]   # two raters, perfect agreement
>>> krippendorff_alpha(data)
1.0

ek.lexicon_corrector(vocabulary: Iterable[str], *, threshold: float = 0.8, scorer: Any = None, field_name: str = '', layer: str = 'lexicon', flag_unmatched: bool = True) → Corrector

L2: resolve a value against a closed vocabulary by fuzzy match (rapidfuzz).

The safest corrector in the stack when the candidate set is closed (country codes, SKUs, enums): an exact member passes untouched; a single close match (similarity >= threshold) is applied as a CORRECT finding; otherwise the value is FLAGged as out-of-vocabulary (when flag_unmatched). Against an open vocabulary, set flag_unmatched=False and treat matches as suggestions only.

Parameters:

• vocabulary – The closed set of valid values.

• threshold – Minimum rapidfuzz similarity in [0, 1] to auto-correct.

• scorer – A rapidfuzz scorer (default fuzz.ratio).

• flag_unmatched – FLAG a value with no close match (default True).

ek.llm_corrector(correct_fn: Callable[[str], str | None], *, only_flagged: bool = True, field_name: str = '', layer: str = 'llm_correct') → Corrector

L5: a gated neural/LLM corrector – the only layer that invents content.

correct_fn is an injected (value) -> Optional[str] (bring your own LLM/seq2seq call; return a corrected string, or None to leave the value unchanged). It is the most expensive and stochastic layer, so by default (only_flagged) it fires only on a value that a cheaper layer already FLAGged – gate it to the residual, never let it silently rewrite clean fields (this layer reads the pipeline’s accumulated findings via wants_findings). Emits a CORRECT finding; the pipeline applies the rewrite and keeps the original for audit. Always verify its output downstream – it can make text worse.

Recipe (Anthropic): correct_fn=lambda v: client.messages.create(model=..., messages=[{"role": "user", "content": prompt(v)}]).content[0].text.strip() with client = anthropic.Anthropic() (MIT). Constrain/verify the result and keep an audit trail.

ek.lm_surprisal_validator(scorer: Callable[[str], float], *, threshold: float, higher_is_worse: bool = True, field_name: str = '', layer: str = 'lm_prior') → Callable[[...], Iterable[Finding]]

L3: FLAG a value whose language-model surprisal crosses threshold.

scorer is an injected (str) -> float returning a surprisal / perplexity / negative-log-likelihood (dependency injection: bring your own in-domain n-gram or masked-LM scorer – an in-domain prior is what makes this useful). FLAG-only on its own; pair it with a candidate generator (lexicon_corrector / llm_corrector) to actually correct. Set higher_is_worse=False if your scorer returns a probability (higher = better).

Recipe (masked-LM pseudo-log-likelihood): wrap minicons – from minicons import scorer; m = scorer.MaskedLMScorer("bert-base-uncased", "cpu") – and pass scorer=lambda s: -m.sequence_score([s])[0] (surprisal = negative PLL). minicons is MIT; install your own LM backend (it pulls torch).

ek.load_baseline(name: str, *, rootdir: str | None = None) → dict | None

Load a named baseline (or None if it does not exist).

ek.load_calibrator(name: str, *, rootdir: str | None = None) → Any

Reconstruct a persisted calibrator by name (dispatched on its kind).

Validates the stored record so a malformed or unknown kind fails with an actionable error rather than a raw KeyError or a load-then-crash-later.

ek.luhn_check(number: Any) → bool

Luhn (mod-10) checksum: credit cards, IMEIs, some national IDs.

ek.mall(*, rootdir: str | None = None) → Mall

The full stores-of-stores mall (one JSON store per KINDS).

ek.names(namespace: str) → Iterable[str]

All registered names in a namespace (after entry-point discovery).

ek.ordering_validator(keys: Sequence[str], *, strict: bool = True, layer: str = 'cross_field') → Callable[[...], Iterable[Finding]]

FLAG a record whose comparable values at keys are not ascending.

The canonical date/sequence check (start <= end, issue before due). Only the keys actually present and mutually comparable are checked, so it is safe on partial records.

ek.pass_at_k(*, n: int, c: int, k: int) → float

Unbiased pass@k: probability that at least one of k samples succeeds.

The HumanEval estimator 1 - C(n-c, k) / C(n, k) over n trials with c successes. This is the capability metric – right when a single success is enough (offline candidate generation behind a verifier).

Example

>>> round(pass_at_k(n=10, c=1, k=1), 3)
0.1
>>> pass_at_k(n=10, c=10, k=5)
1.0

ek.pass_hat_k(*, n: int, c: int, k: int) → float

Unbiased pass^k: probability that all k independent trials succeed.

The tau-bench estimator C(c, k) / C(n, k); for a per-trial success probability p it decays to p**k. This is the reliability metric – the one that matters when consistency is the product.

Example

>>> pass_hat_k(n=8, c=8, k=8)
1.0
>>> pass_hat_k(n=8, c=4, k=2)
0.21428571428571427

ek.percent_agreement(rater_a: Sequence, rater_b: Sequence) → float

Raw fraction of items two raters labelled identically.

ek.persistent_cache(func: Callable | None = None, *, kind: str = 'runs', key: Any = None, rootdir: str | None = None)

Memoize a plain function’s result in a kind JSON store, across sessions.

Use as @persistent_cache(kind="runs", key=...). Returned values must be JSON-serializable. key may be a constant or a callable receiving the call arguments; it defaults to the function’s qualified name (so provide a key for functions that take arguments). For caching a class property/method, prefer cache_this (dol’s descriptor) instead.

Example

>>> import tempfile; root = tempfile.mkdtemp()
>>> calls = []
>>> @persistent_cache(kind="runs", key="answer", rootdir=root)
... def compute():
...     calls.append(1)
...     return {"answer": 42}
>>> compute(), compute(), len(calls)
({'answer': 42}, {'answer': 42}, 1)

ek.range_validator(lo: float, hi: float, *, field_name: str = '', layer: str = 'range') → Callable[[...], Iterable[Finding]]

A Validator that flags a number outside [lo, hi].

ek.regex_validator(pattern: str, *, field_name: str = '', layer: str = 'format') → Callable[[...], Iterable[Finding]]

A Validator that flags a value not fully matching pattern.

ek.register(namespace: str, name: str, obj: Any | None = None) → Any

Register obj under namespace/name (usable as a decorator).

Parameters:

• namespace – A strategy family, e.g. "metrics", "signals", "calibrators", "policies", "normalizers", "ocr".

• name – The lookup key within the namespace.

• obj – The object to register; omit to use as a decorator.

Returns:

obj (so it can be used as a decorator).

ek.regression_gate(report, baseline: Any, *, tolerance: float = 0.0, higher_is_better: bool | None = None, rootdir: str | None = None) → GateResult

Fail if report regresses beyond tolerance vs a baseline, per slice.

Parameters:

• report – The current Report.

• baseline – A baseline name (loaded from the baselines store) or a baseline dict from save_baseline().

• tolerance – Allowed drift before a change counts as a regression.

• higher_is_better – Override metric-direction inference (CER/WER/graph are lower-is-better; F1/similarity are higher-is-better).

• rootdir – data root (when baseline is a name).

Returns:

A GateResult (falsy if any regression was found).

ek.reliability_curve(probs: Sequence[float], correct: Sequence[bool], *, n_bins: int = 10) → List[dict]

Per-bin {confidence, accuracy, count} for a reliability diagram.

ek.requires_extra(extra: str, *, packages: Iterable[str] | None = None) → Callable

Decorator: fail with an actionable install hint if an extra is missing.

The wrapped callable runs only once every package in packages imports; the common case (everything installed) adds a single cheap import check.

Parameters:

• extra – The extra name, used in the hint pip install ek[<extra>].

• packages – Import names to probe (defaults to [extra]).

Example

>>> @requires_extra("ocr", packages=["definitely_not_installed_pkg"])
... def run():
...     return "ran"
>>> try:
...     run()
... except MissingExtraError as e:
...     print("ek[ocr]" in str(e))
True

ek.resolve(namespace: str, ref: Any, *, default: Any | None = None) → Any

Coerce a strategy reference to a callable.

ref may be a registered name (str), an already-resolved object, or None (in which case default, itself a name or object, is used).

ek.resolve_table_parser(ref: str | Callable[[Any], Any]) → Callable[[Any], Any]

Resolve a table parser reference (a registered name or a callable) to a callable.

Mirrors how other ek strategies resolve from the registry (see ek.registry.resolve()). A str is looked up under the "table_parsers" namespace; an already-callable parser is returned unchanged.

ek.risk_coverage_curve(probs: Sequence[float], correct: Sequence[bool], *, thresholds: Sequence[float] | None = None) → List[dict]

The risk-coverage trade-off: {threshold, coverage, selective_risk} per point.

Coverage is the fraction auto-accepted at each threshold; selective risk is the error rate among the accepted. Lower-left is better; pick the operating point against a target risk or coverage.

ek.rover(hypotheses: Iterable[Any], *, use_confidence: bool = True, conf_weight: float = 0.5, null_conf: float = 0.7, tokenize: Callable[[Any], List[Tuple[str | None, float]]] | None = None, max_tokens: int | None = 5000) → RoverConsensus

Align N hypotheses, vote per slot, and emit consensus + per-position agreement.

Parameters:

• hypotheses – The recognizer/extractor outputs to fuse. Each may be a string, an OcrResult-shaped object (.text/.mean_confidence), a list of token strings, or a list of (token, confidence) pairs.

• use_confidence – Blend confidence into the vote (True) or vote purely by frequency (False, which forces conf_weight to 0).

• conf_weight – Weight on average confidence vs vote frequency in the slot score (Fiscus’s 1 - alpha); 0 is frequency-only, 1 is confidence-only. Ignored when use_confidence is False.

• null_conf – Confidence credited to a NULL vote (an engine that produced no token at a slot) – the lever for how readily a deletion wins.

• tokenize – Optional hypothesis -> [(token, confidence), ...] override; by default _as_units() handles strings/OcrResults/token lists.

• max_tokens – Reject any hypothesis longer than this (the aligner is O(N*l*L*L'); an unbounded input is a quadratic time/memory DoS). None disables the guard.

Returns:

A RoverConsensus with the consensus tokens/text, the per-slot breakdown, and the per-consensus-token agreement usable as a raw signal.

ek.run_suite(agent: Callable[[TaskSpec], Any], tasks: Any, *, k: int = 1, check: Any = None, metrics: Mapping | None = None, price: ModelPrice | None = None, prices: Mapping[str, ModelPrice] | None = None, run: RunProvenance | None = None, seed: int | None = None, persist: bool = False, run_id: str | None = None, rootdir: str | None = None) → ReliabilityReport

Run agent over a task suite, k trials per task, and report reliability + cost.

Parameters:

• agent – The system under test: TaskSpec -> Episode (a bare answer is wrapped). It receives the whole spec – .input, .tools, .gold – not just the input. If it accepts a keyword seed, the per-trial seed is passed to it.

• tasks – TaskSpec s (or a {task_id: spec} mapping).

• k – Trials per task. k > 1 is what makes pass^k meaningful – and it only means anything if the agent is genuinely stochastic (the report warns if it is not).

• check – The success oracle – a registered checker name or an (episode, gold) -> bool callable. Defaults to "output"; inject a state-based/executable oracle for a real suite, and run it isolated from the agent.

• metrics – Optional {name: Metric} scored on every episode against task.gold, with the suite’s Layer-A grammar injected – this is how the tool/argument cost weights reach a tool_call/trajectory metric. Results land in report.detail["metrics"].

• price/prices – Rates for costing the episodes (see ek.agents.cost).

• run – RunProvenance for this run (model, simulator, suite version, scaffold). Recorded on every episode and on any baseline saved from it.

• seed – Base RNG seed. Trial i of each task uses seed + i: it is recorded on the episode’s provenance and passed to the agent when the agent accepts a seed keyword.

• persist/run_id/rootdir – Persist the run to the runs/results stores.

Returns:

A ReliabilityReport carrying pass@k, pass^k (with a bootstrap CI), the success rate (with a Wilson CI), the per-slice cuts, and the cost report – because reliability without cost is only half the answer.

ek.save_baseline(report, name: str, *, rootdir: str | None = None) → dict

Freeze a report’s aggregate + per-slice scores as a named baseline.

ek.save_calibrator(calibrator: Any, name: str, *, rootdir: str | None = None) → dict

Persist a fitted calibrator’s parameters to the calibrators store.

ek.schema_validator(value: Any, *, spec: FieldSpec | None = None) → Iterable[Finding]

A spec-driven Validator: type and domain checks from a FieldSpec.

Reads the FieldSpec type and domain – a (lo, hi) numeric range, an enum tuple, or a regex string – and applies the matching check. With no spec it is a no-op, so it composes safely in any validator list.

ek.score(pred: Any, gold: Any, *, grammar: GraphGrammar | None = None, metric: None | str | Metric = None, normalize: Any = None, weights: Any = None) → Score

Score one prediction against one gold reference (reference-based).

Parameters:

• pred – The predicted output (string, record dict, or anything with .text).

• gold – The gold reference, same shape as pred.

• grammar – Optional Layer-A GraphGrammar (carries cost weights).

• metric – A registered metric name ("cer", "wer", "fields", …), a callable Metric, or None to dispatch by type.

• normalize – Optional canonicalizer (name, callable, step list, or Canonicalizer) applied before comparison.

• weights – A CostWeight for cost-weighted metrics (the typed-graph distance); overrides the schema’s importance weights.

Returns:

A Score.

ek.split_conformal_quantile(scores: Sequence[float], alpha: float) → float

The split-conformal threshold: the ceil((n+1)(1-alpha)) / n empirical quantile.

scores are nonconformity scores on an exchangeable calibration set. Returns +inf when n is too small to guarantee 1 - alpha (so nothing is flagged – the honest behaviour at that sample size), and -inf at the alpha -> 1 limit (flag everything).

ek.stop_on_correction(findings: Sequence[Finding]) → bool

Stop policy: end the pass as soon as any layer proposes a correction.

ek.stop_on_flag(findings: Sequence[Finding]) → bool

Stop policy: end the pass at the first FLAG (fail fast).

ek.table_from_ocr(ocr_result: Any, *, parser: None | str | Callable[[Any], Any] = None) → Table | None

Extract a normalized Table from an OcrResult.

The single seam ek’s table metrics (TEDS / GriTS) use to score OCR table output uniformly. Returns the recovered Table, or None when the result carries no recoverable table structure.

Parameters:

• ocr_result – Any OcrResult-shaped object. ek reads only its .raw (the engine-specific payload), .markdown (VLM/markdown engines), and .meta – never ocracy types, so any image -> OcrResult callable’s output works.

• parser –

  How to turn the raw payload into a table (dependency injection / open-closed). One of:

  • None (default): try the built-in safe heuristics – .raw is already a Table, a 2-D grid, or a <table>...</table> HTML string; or .raw/.meta is a mapping carrying one of {tables, table, cells, grid, html}; or .markdown contains an HTML <table>. Returns None if none match – it never guesses wildly.

  • a str: the name of a parser registered under "table_parsers" (per-engine extractor), resolved from ek.registry.

  • a callable (raw) -> Table | grid | html | None: an inline per-engine extractor. It receives the result’s .raw (falling back to the result itself when there is no .raw).

• through (Any non-None value an injected parser returns is fed)

:param Table.coerce(): :param so a parser may yield a Table: :param a 2-D grid: :param or HTML.:

Returns:

A non-empty Table, or None when no table structure is recoverable. An empty Table (no rows/cells) is normalized to None so the predicate has_table_structure() and downstream metrics see “no table”, not a degenerate one.

ek.totals_consistent(record: Mapping, *, total_key: str, item_keys: Sequence[str], tol: float = 0.01, layer: str = 'cross_field') → Iterable[Finding]

Flag when record[total_key] does not equal the sum of item_keys (±tol).

The canonical cross-field check (invoice total = Σ line items). Skipped (rather than raising or flagging) when the total or all item keys are absent or non-numeric, so it is safe on partial records.

ek.validation_pipeline(*layers: ~typing.Callable[[...], ~typing.Iterable[~ek.base.Finding]], apply_corrections: bool = True, stop_when: ~typing.Callable[[~typing.Sequence[~ek.base.Finding]], bool] = <function _never_stop>) → Callable[[...], ValidationResult]

Compose validators/correctors into a cheapest -> most-expensive pipeline.

layers run in the given order (the order encodes the cost spine). Each layer is a (value, *, spec) -> Iterable[Finding] callable. When apply_corrections (default), the first CORRECT finding a layer emits is applied to the value before the next layer runs – so a cheap deterministic fix is what the expensive layers then see (the noisy-channel chain). stop_when is an injectable early-exit policy over the findings accumulated so far (see stop_on_correction() / stop_on_flag()); the default runs every layer.

ek.zscore_anomaly_findings(numbers: Iterable[Any], *, field: str = '', threshold: float = 3.5, min_n: int = 30, layer: str = 'anomaly') → list

FLAG numeric values that are robust-z-score outliers (median + MAD based).

A reference-free, dependency-free anomaly check that complements benford_findings() (which checks the leading-digit distribution; this checks individual magnitudes). For a numeric column it computes each value’s modified z-score – 0.6745 * (x - median) / MAD – which is robust to the very outliers it is looking for, and FLAGs those whose absolute score exceeds threshold. One FLAG per outlying value (carrying its index); FLAG-only – it routes a value to review, never edits. Skipped below min_n (robust statistics on a handful of points are noise). When the MAD is 0 (a near-constant column, e.g. many identical values plus one outlier) it falls back to the mean-absolute-deviation scale (Iglewicz & Hoaglin), so a lone outlier is still caught; a truly constant column (no spread at all) yields no findings. For multivariate outliers, plug an isolation forest via pyod (already in ek[validation]) as a custom validator.