Stage reference
Every stage: (records, **params) → records over
plain list[dict], deterministic, non-mutating, auto-traced when
a trace() is active. Contract, not implementation — inputs,
outputs, and the invariants held.
select(records, fields)
Keeps only fields on each record, in
keep-list order. Missing fields are simply absent (no error, no
None padding). Lossless: the dropped tokens were fields you
were never going to use. Also the stage that makes rows uniform — which
downstream table encodings reward. Everywhere a stage takes
fields, a bare string names one field
(fields="snippet" ≡ fields=["snippet"]) — it is
never iterated as characters.
dedupe(records, key=None)
Drops duplicates, first occurrence wins, order
preserved. key=None compares whole records (key-order
insensitive); a field name or list of names compares those fields.
Fingerprints are type-qualified — 1, 1.0,
True, "1" are four different values — with a
canonical-JSON fallback for unhashable values.
allowlist(records, field, allowed)
Provenance gate. Keeps only records whose
field value is in allowed; records missing the
field, or carrying an unhashable value, are dropped — fail closed. The
strongest injection control a shaper can offer is refusing content from
sources you did not approve: apply this first, before any
relevance logic, so untrusted records never compete for the budget.
Dropped records appear in the trace/audit with this stage as the reason.
quarantine(records, fields=("snippet",), patterns=None, action="drop", into="quarantined")
Injection tripwire. A deterministic, case-insensitive
pattern screen over the given text fields for the crudest prompt-injection
markers — "ignore all previous instructions", role resets ("you are now"),
system-prompt tags. action="drop" removes matches (audited);
action="flag" keeps everything and writes a boolean to
into so downstream policy decides.
rescore(records, query, fields=("snippet",), into="score", proximity=True, match="word")
Writes a deterministic lexical relevance score to
into on a copy of each record, so ranking stops inheriting the
search tool's score quality. BM25-style, computed inside the batch:
- per-term IDF
ln((N−df+0.5)/(df+0.5)+1)— rare terms weigh more; - saturating term frequency
tf/(tf+1.5)— covering all the query's terms beats repeating one; match="word"(default) counts exact word occurrences the way BM25/Lucene tokenize —value≠values, underscores split snake_case;match="substring"is the forgiving, decoy-prone alternative;proximity=True: consecutive query terms in order within 80 characters earn a bonus — the source a query was phrased from usually carries its terms in sequence.
rank(records, by, desc=True)
Sorts stably by a field name or key function; records
missing the field go last. Run it before trim_to_budget so the
budget drops the least important records, not arbitrary ones. Values of
by must be mutually comparable.
truncate_field(records, field, max_tokens, tokenizer=None, suffix="…")
Clips one long text field to the longest prefix that
fits the cap, then appends the suffix. For the built-in estimator the cut is
closed-form (ceil(len/4) is invertible: 4×budget characters);
for custom tokenizers it binary-searches. Non-string values pass through
untouched. Never exceeds the cap.
trim_to_budget(records, max_tokens, tokenizer=None, min_records=0)
Keeps the greedy prefix that fits the budget: each
record costs the token count of its serialized text plus 2 for list
framing; the walk stops before the budget is crossed. Assumes records are
already ordered by importance. If even the first record exceeds the budget
the result is empty — run truncate_field first for oversized
records, or set min_records: the first that-many records (the
best ones, post-rank) are kept even over budget, so a
too-small budget can never return an empty list that reads as "nothing
matched". The overrun stays visible in the trace/audit token counts.
merge(*sources, schema=None, dedupe_key=None)
Unifies differently-shaped tool outputs into one record
shape. schema maps a unified field name to a source field name
or a list of candidates tried in order; dedupe_key optionally
dedupes across the merged result. With schema=None the sources
are concatenated unchanged. A combiner rather than a single-collection
transform — the one stage that is not auto-traced.
merge(web, docs, schema={"title": ["title", "name", "headline"],
"url": ["url", "link", "href"]})
pipeline · stage · shaped · trace · tokens
pipeline([...])— plain left-to-right function composition; anyrecords → recordscallable (even a lambda) is a valid stage, so you can drop to code for anything irregular and still breakpoint inside it.stage(fn, **params)— binds keyword args so a stage slots into a pipeline.@shaped(pipe)— runs a tool's record-list return value through a pipeline before it returns; the model never sees raw output. Async tools work too: a coroutine function is awaited, then shaped.pipeline([...])exposes.fingerprint— a stable policy hash of stage names and bound parameters, recorded into any active trace so an audit can say which policy version shaped a context.trace(id_field=...)— context manager; every stage records records-in/out and tokens-before/after, and withid_fieldalso which records each stage dropped (records lacking the field are untracked).t.audit()emits the JSON-able governance record. Inactive overhead is a single ContextVar lookup per stage — single-digit milliseconds (49.8 vs 44.4 ms over 10k records in RESULTS.md); active tracing token-counts every stage boundary and is priced accordingly.tokens—count,use_tiktoken()(exact where tiktoken covers the model),tiktoken_tokenizer(),set_tokenizer(fn),set_serializer(fn),serialize(value), andscoped(...)— a context-local overlay that isolates concurrent tenants. Strings always bypass the serializer.
GitHub