ef

ef (Embedding Flow) — a facade for semantic-embeddings user journeys.

ef makes the modern semantic-search pipeline — corpus, segmenter, embedder, vector store, retriever — usable with progressive disclosure: the light case (a list of strings, search in one or two lines) and the heavy case (huge corpora, many segmentations/embedders, varied sources and vector DBs) share one facade. ef is not RAG — it returns ranked segments; bring your own LLM.

The embedder facade is the core surface and is always importable (it costs only numpy):

The segmenter facade is the other always-importable core surface (it needs no numpy at all):

The corpus facade is ef’s source layer (L0) — also always importable:

The artifact graph is ef’s corpus-indexing core (also always importable) — a content-addressed producer graph in which cascade invalidation and config branching are the same operation:

The search facade is where the layers come together — corpus → segment → embed → vd → ranked search, with progressive disclosure:

The refresh layer keeps an indexed corpus in sync as its sources change:

The RAG-plug-in & evaluation layer hands a corpus to an external RAG/agent framework and measures its quality — ef returns ranked context, it does not synthesize answers:

The explore layer (L5) is ef’s visualization heritage, kept as a secondary “see the shape of the corpus” surface — search/RAG/indexing is the primary one. It imports numpy-only; UMAP, HDBSCAN and imbed are lazy extras:

  • project() — reduce embeddings to 2-D/3-D coordinates (PCA → UMAP); cluster() — group them (k-means / HDBSCAN); label_clusters() — name clusters with an LLM via imbed.

  • explore() — the orchestrator: project + cluster (+ label) in one call, returning a structured, JSON-friendly ExploreResult (ids / coords / labels).

The service layer bridges ef’s stateful objects to a stateless transport — hand its methods to qh.mk_app() to serve ef over HTTP:

Example — wrap a plain function and embed two strings:

>>> import numpy as np
>>> from ef import as_embedder
>>> embedder = as_embedder(
...     lambda texts: np.array([[len(t)] for t in texts]), model_id='len@1'
... )
>>> embedder(['hello', 'hi']).ravel().tolist()
[5.0, 2.0]
class ef.ArtifactGraph(*, store: MutableMapping[str, Any] | None = None, producers: MutableMapping[str, ProducerSpec] | None = None, edges: MutableMapping[str, tuple[str, ...]] | None = None, ops: MutableMapping[str, Callable[[...], Any]] | None = None)[source]

A content-addressed producer graph — the corpus-indexing & refresh core.

The graph holds two kinds of node:

and four small MutableMapping stores, every one injectable so the graph can be backed by RAM (the default) or by a persistent dol store:

  • store — the content-addressed cache: artifact_id -> value.

  • producers — the recipe registry: artifact_id -> ProducerSpec.

  • edges — the reverse index: artifact_id -> (dependent ids, ...), so the downstream cone of any node is one lookup away.

  • ops — the op registry: op_key -> callable. The one store that is not persisted (callables are not serializable) — repopulate it with register_op() each process.

The four contract operations:

  • materialize() — lazy backward: compute a node by recursively materializing its inputs, then memoize.

  • mark_stale() — forward: drop the cached values of a node’s produced downstream cone, keeping the recipes (they will rebuild).

  • delete_cascade() — forward: remove a node and its whole downstream cone entirely (values, recipes and edges).

  • freshness() — query a node’s state (Freshness).

plus the reachability queries descendants() (the downstream cone) and ancestors() (lineage — what an artifact was produced from).

Parameters:

  • store – the content-addressed value cache. Defaults to a fresh dict.

  • producers – the recipe registry. Defaults to a fresh dict. If passed already-populated (a resumed/persisted graph) and edges is left empty, the reverse index is rebuilt from it on construction.

  • edges – the reverse-dependency index. Defaults to a fresh dict; if passed already-populated it is trusted as-is (not rebuilt).

  • ops – the op registry. Defaults to a fresh dict; register_op() adds to it.

>>> graph = ArtifactGraph()
>>> _ = graph.register_op('join', lambda a, b, *, sep: f'{a}{sep}{b}')
>>> x = graph.put('x', 'hello')
>>> y = graph.put('y', 'world')
>>> both = graph.add(producer_spec('join', x, y, op_version='1', sep=', '))
>>> graph.materialize(both)
'hello, world'
>>> sorted(graph.descendants('x'))
[...]
>>> graph.delete_cascade('y')                     # the 'y' source is removed
frozenset(...)
>>> graph.freshness(both)                         # its consumer is gone too
'unknown'
add(spec: ProducerSpec) → str[source]

Register a producer recipe; return its content-addressed id.

Idempotent: the id is the recipe, so adding an equal ProducerSpec twice (or adding the shared upstream step of two branching configs) registers one node and returns the same id — the basis of free config branching. Inputs need not exist yet; the graph is declarative and materialize() resolves them lazily.

>>> graph = ArtifactGraph()
>>> a = graph.add(producer_spec('op', 'leaf', op_version='1'))
>>> a == graph.add(producer_spec('op', 'leaf', op_version='1'))  # idempotent
True
>>> len(graph.producers)
1
ancestors(key: str) → frozenset[str][source]

The lineage of key — every artifact it was transitively produced from.

Excludes key itself. Answers “what produced this?” — provenance / lineage queries. The leaves of the returned set are the original sources the artifact ultimately derives from.

>>> graph = ArtifactGraph()
>>> _ = graph.register_op('id', lambda x: x)
>>> a = graph.add(producer_spec('id', 'src', op_version='1'))
>>> b = graph.add(producer_spec('id', a, op_version='1'))
>>> graph.ancestors(b) == {'src', a}
True
delete_cascade(key: str) → frozenset[str][source]

Remove key and its whole downstream cone — forward; return what was removed.

Deletes every artifact in {key} ∪ descendants(key) entirely — cached value, recipe and edges — and detaches the cone from any upstream inputs that survive it. Unlike mark_stale() this removes the recipes too: use it when an artifact can never be rebuilt, the canonical case being a deleted source (its segments, vectors and index entries cascade away with it).

Returns the ids actually removed (those that had a value or a recipe).

>>> graph = ArtifactGraph()
>>> _ = graph.register_op('id', lambda x: x)
>>> graph.put('s', 'v')
's'
>>> a = graph.add(producer_spec('id', 's', op_version='1'))
>>> graph.delete_cascade('s') == {'s', a}
True
>>> 's' in graph, a in graph
(False, False)
descendants(key: str) → frozenset[str][source]

The downstream cone of key — every artifact transitively produced from it.

Excludes key itself. This is the set of artifacts a change to key would invalidate; mark_stale() and delete_cascade() are built on it.

>>> graph = ArtifactGraph()
>>> _ = graph.register_op('id', lambda x: x)
>>> a = graph.add(producer_spec('id', 'src', op_version='1'))
>>> b = graph.add(producer_spec('id', a, op_version='1'))
>>> graph.descendants('src') == {a, b}
True
freshness(key: str) → Literal['materialized', 'stale', 'unknown'][source]

Report the state of key — Freshness.

"materialized" if a value is cached; "stale" if a recipe exists but no value (a materialize() will build it); "unknown" if the graph has neither.

Because the graph is content-addressed, a cached value is always correct for its id — staleness here means “not yet computed”, never “computed wrong”. The richer staleness conditions (orphan / missing / misconfigured, computed against vd index metadata) belong to a later diagnostics phase, not to the graph core.

>>> graph = ArtifactGraph()
>>> _ = graph.register_op('id', lambda x: x)
>>> graph.put('s', 'v')
's'
>>> a = graph.add(producer_spec('id', 's', op_version='1'))
>>> graph.freshness('s'), graph.freshness(a), graph.freshness('?')
('materialized', 'stale', 'unknown')
mark_stale(key: str) → frozenset[str][source]

Invalidate key’s produced downstream cone — forward; return that cone.

Drops the cached values of every produced artifact in {key} ∪ descendants(key), keeping the recipes — so the next materialize() recomputes them. Leaf nodes (inputs, no recipe) are never touched: they cannot be recomputed, so dropping their value would be data loss — use delete_cascade() to remove an input.

This is the invalidation half of a refresh: when an op’s implementation changes, or to force a recompute. (A source change is usually a delete_cascade() of the old content hash — the new content yields a new leaf id.) Returns the produced artifacts that are now stale.

>>> graph = ArtifactGraph()
>>> _ = graph.register_op('id', lambda x: x)
>>> graph.put('s', 'v')
's'
>>> a = graph.add(producer_spec('id', 's', op_version='1'))
>>> _ = graph.materialize(a)
>>> graph.mark_stale('s') == {a}        # the leaf 's' itself is not "stale"
True
>>> graph.freshness(a)
'stale'
materialize(key: str) → Any[source]

Compute (or fetch the cached value of) the artifact key — lazy backward.

If key is already cached in store its value is returned directly. Otherwise its recipe’s inputs are materialized recursively (depth-first), the op is called op(*input_values, **params), and the result is memoized in store before being returned — so a second materialize() is a cache hit, and a shared upstream artifact is computed exactly once however many configs depend on it.

Raises:

  • KeyError – if key is neither cached nor has a producer recipe (an unresolved leaf — supply it with put()).

  • LookupError – if a recipe’s op is not in the ops registry.

  • ValueError – if the producer graph contains a cycle (a content- addressed graph cannot — this guards a hand-corrupted registry).

>>> graph = ArtifactGraph()
>>> _ = graph.register_op('double', lambda x: x * 2)
>>> graph.put('n', 21)
'n'
>>> out = graph.add(producer_spec('double', 'n', op_version='1'))
>>> graph.materialize(out)
42
put(key: str, value: Any) → str[source]

Store a leaf value (an external input) under key; return key.

A leaf has a value but no recipe — typically an L0 source keyed by its ef.corpus.content_hash(). Returns key so the call composes:

src = graph.put(content_hash(text), text)

>>> graph = ArtifactGraph()
>>> graph.put('doc-1', 'some text')
'doc-1'
>>> graph.freshness('doc-1')
'materialized'
register_op(name: str, fn: Callable[[...], Any]) → Callable[[...], Any][source]

Register the callable fn under the op key name; return fn.

materialize() resolves a ProducerSpec’s string op through this registry. The op is called fn(*input_values, **params): the materialized inputs positionally, the spec’s params as keywords.

>>> graph = ArtifactGraph()
>>> graph.register_op('upper', str.upper)
<method 'upper' of 'str' objects>
ef.ArtifactId

alias of str

class ef.BaseEmbedder[source]

Implementation base for embedders (optional — Embedder is structural).

Subclasses set model_id / dim / normalized / honored_input_types and implement __call__; they then satisfy Embedder structurally. In return they get:

  • a default synchronous embed_batch() (ready_handle(self(...))) — override it only for a genuinely async provider Batch API;

  • a readable repr.

embed_batch(texts: Iterable[str], *, input_type: Literal['query', 'document', 'classification', 'clustering'] | None = None, **backend: Any) → BatchHandle[source]

Default: embed synchronously and return a finished handle.

class ef.BaseSegmenter[source]

Implementation base for segmenters (optional — Segmenter is structural).

A subclass implements __call__; in return it gets a default batch() (one-document-at-a-time — override it for a genuinely batched backend) and a readable repr. Subclassing is purely a convenience; satisfying Segmenter structurally is what matters.

batch(docs: Iterable[str | Mapping[str, Any]], /) → Iterator[Iterable[Segment]][source]

Default: segment each document independently.

class ef.BatchHandle(poll: ~typing.Callable[[], ~typing.Literal['pending', 'done', 'failed']], result: ~typing.Callable[[], ~numpy.ndarray], cancel: ~typing.Callable[[], None] = <function BatchHandle.<lambda>>)[source]

A poll/result handle over an in-flight (or already-finished) embed job.

Batch-and-poll is just a special case of async. A synchronous backend returns a ready handle (via ready_handle()); a provider Batch API returns a genuinely pending one. Either way the caller sees one shape.

poll

() -> BatchStatus — non-blocking status check.

Type:

Callable[[], Literal[‘pending’, ‘done’, ‘failed’]]

result

() -> np.ndarray — blocks until done, returns (n, dim).

Type:

Callable[[], numpy.ndarray]

cancel

() -> None — best-effort cancellation.

Type:

Callable[[], None]

class ef.BatchedSegmenter(*args, **kwargs)[source]

Optional trait: a segmenter that can process many documents at once.

Plain Segmenter handles one document per call. A segmenter that can amortize per-call cost across documents (a shared model load, a vectorized pass) also offers batch. BaseSegmenter supplies a correct — if un-optimized — default, so every BaseSegmenter subclass is already a BatchedSegmenter.

batch(docs: Iterable[str | Mapping[str, Any]], /) → Iterable[Iterable[Segment]][source]

Segment each document in docs, yielding one segment-stream each.

class ef.CachedEmbedder(inner: Embedder, store: MutableMapping[str, ndarray], *, cache_queries: bool = False)[source]

Memoize embeddings in a MutableMapping store.

On each call, texts already in the store are served from it; only the misses are forwarded to the inner embedder (in a single batch) and then written back. The store can be any dol-style mapping — an in-RAM dict, a directory of .npy files, a key-value DB.

Document embeddings are cached unconditionally; query embeddings are skipped by default (they are rarely repeated) — pass cache_queries=True to cache them too.

Parameters:

  • inner – The embedder to memoize.

  • store – A MutableMapping[str, ndarray] keyed by cache_key().

  • cache_queries – Whether to also cache input_type="query" calls.

property honored_input_types: tuple[Literal['query', 'document', 'classification', 'clustering'], ...]

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable’s items.

If the argument is a tuple, the return value is the same object.

property normalized: bool

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

class ef.ChangeDetectingCorpus(corpus: Any = None, *, on_change: Callable[[ChangeEvent], Any] | None = None, hashes: MutableMapping[str, str] | None = None, content_keys: Iterable[str] | None = None, fingerprint: Callable[[str], Hashable] | None = None)[source]

A corpus wrapper that content-hashes values and reports what changed.

Wraps any inner corpus (a MutableMapping[source_id, Source]) and adds change detection without altering its mapping behavior — reads, writes and deletes pass straight through to the inner store. It is to a corpus what CachedEmbedder is to an embedder: a transparent, composable layer.

Changes are noticed two ways:

  • Through the wrapper. A corpus[id] = source or del corpus[id] that actually changes content fires the injected on_change callback with a ChangeEvent. Writing identical content is a no-op (no event) — the basis of idempotent ingestion.

  • Out of band. When the backing store is edited directly (a file changed on disk, an S3 object replaced), scan() re-reads every source, compares hashes, fires events for what differs and returns a CorpusDiff. diff() runs the same comparison with no side effects.

The on_change callback is the seam the downstream ArtifactGraph plugs into: a "modified" / "deleted" event is exactly the trigger for a cascade invalidation.

On construction a baseline is established: the current content hashes of the inner corpus are recorded silently (no events). Pass an already-populated hashes registry to instead resume from a persisted baseline — then call scan() to reconcile any drift.

Parameters:

  • corpus – the inner corpus to wrap. Coerced via as_corpus(), so a directory path / dict / iterable of sources / None all work.

  • on_change – called with each ChangeEvent. Defaults to a no-op. It should be cheap and not raise; an exception propagates to the caller after the corpus and registry are already consistent.

  • hashes – the hash registry — a MutableMapping[source_id, str] of the last-known content hash per source. Injectable so it can be persisted (e.g. a dol store); defaults to an in-RAM dict. If passed already-populated it is trusted as the baseline.

  • content_keys – forwarded to content_hash() — metadata keys that participate in the hash.

  • fingerprint – an optional cheap source_id -> Hashable callback (an mtime / etag prefilter). When given, scan() / diff() skip re-hashing a source whose fingerprint is unchanged. The content hash always remains the source of truth — the fingerprint is only a speed optimization, never trusted to declare a change.

>>> backing = {'a': 'one', 'b': 'two'}
>>> corpus = ChangeDetectingCorpus(backing)   # baseline adopts 'a' and 'b'
>>> backing['a'] = 'ONE'                      # edited behind the wrapper's back
>>> backing['c'] = 'three'
>>> del backing['b']
>>> d = corpus.scan()
>>> (d.added, d.modified, d.deleted)
(('c',), ('a',), ('b',))
diff() → CorpusDiff[source]

Compare current corpus content against the registry — no side effects.

Detects out-of-band edits (changes made to the backing store directly) without firing events or updating the registry. scan() is the effectful counterpart.

scan() → CorpusDiff[source]

Reconcile the registry with the corpus, fire events, return the diff.

Use this to pick up out-of-band edits — changes made to the backing store directly rather than through this wrapper. Every detected change fires a ChangeEvent and the registry (and fingerprints) are advanced to the new state, so a second scan() with nothing further changed is a no-op. diff() is the side-effect-free version.

class ef.ChangeEvent(source_id: str, kind: Literal['added', 'modified', 'deleted'], old_hash: str | None = None, new_hash: str | None = None)[source]

A single detected change to a source in a corpus.

Emitted by ChangeDetectingCorpus to its on_change callback. old_hash is None for an "added" event, new_hash is None for a "deleted" one.

source_id

the key of the source that changed.

Type:

str

kind

"added", "modified" or "deleted".

Type:

Literal[‘added’, ‘modified’, ‘deleted’]

old_hash

the source’s previous content hash, if any.

Type:

str | None

new_hash

the source’s new content hash, if any.

Type:

str | None

ef.ConfigId

alias of str

class ef.CorpusDiff(added: tuple[str, ...] = (), modified: tuple[str, ...] = (), deleted: tuple[str, ...] = (), unchanged: tuple[str, ...] = ())[source]

The result of comparing a corpus’s content against known content hashes.

The four fields partition the corpus’s source ids; each is a sorted tuple, so a CorpusDiff is deterministic and JSON-friendly. Truthiness reports whether anything changed.

>>> d = CorpusDiff(added=('b',), modified=('a',))
>>> bool(d), d.changed
(True, ('b', 'a'))
>>> bool(CorpusDiff(unchanged=('a', 'b')))
False
property changed: tuple[str, ...]

The source ids that are new or modified — the set to (re-)materialize.

class ef.CorpusInfo[source]

A JSON-friendly summary of one registered corpus.

The return shape shared by EfService.create_corpus(), EfService.corpus_info() and (as a list) EfService.list_corpora() — a plain dict an HTTP client and qh’s schema both understand.

Keys:

corpus_id: the registry handle — how every other method addresses it. n_sources: the number of source documents in the corpus. n_segments: the number of indexed segments (sources cut by the segmenter). embedder: the embedder’s model_id (e.g. "hashing:v1@512"). dim: the embedding dimensionality. config_id: the content hash of the segmenter+embedder pipeline.

class ef.EfService(*, default_embedder: str | None = None)[source]

A registry-backed facade bridging stateful ef to a stateless transport.

Construct one EfService per process, register corpora with create_corpus(), then query them by corpus_id. Hand the seven bound methods to qh.mk_app() to expose them over HTTP — see the module docstring.

The handle registry is private instance state — never a module global. Corpora are isolated: each create_corpus() builds its own SourceManager over its own in-memory vd backend, so one corpus’s vectors never leak into another’s search.

default_embedder sets the embedder create_corpus() resolves when its own embedder argument is None — the per-instance hook a host (e.g. app_ef) uses to pick a policy default without passing embedder= on every call. It defaults to DEFAULT_EMBEDDER ("hashing"), so a bare EfService() stays dependency-free and offline.

>>> service = EfService()
>>> len(service)
0
>>> _ = service.create_corpus(['hello world'], corpus_id='greetings')
>>> 'greetings' in service
True
>>> len(service)
1

A host can swap the default embedder at construction (the string is resolved lazily, per corpus, by the DI seam — constructing the service touches no network):

>>> EfService(default_embedder='openai:text-embedding-3-small')
<EfService: 0 corpus(es) registered>
corpus_info(corpus_id: str) → CorpusInfo[source]

Return the CorpusInfo of a registered corpus.

Raises:

KeyError – if corpus_id is not registered.

create_corpus(sources: list[str], *, embedder: str | None = None, segmenter: str | None = None, corpus_id: str | None = None, embedder_api_key: str | None = None) → CorpusInfo[source]

Index sources into a new corpus, register it, return its CorpusInfo.

Parameters:

  • sources – the corpus — a list of text documents.

  • embedder – the embedder, as a string the DI seam resolves (as_embedder()) — "hashing", "openai:text-embedding-3-small", "cohere:...", an http(s):// URL, …. None → the service’s default_embedder (chosen at construction; itself DEFAULT_EMBEDDER, the dependency-free HashingEmbedder, unless a host overrode it).

  • segmenter – the segmenter, as a string (as_segmenter()) — None → the recursive-character default.

  • corpus_id – the handle to register the corpus under; None → a fresh random id. Reusing a live id is an error.

  • embedder_api_key – an optional API key for the embedder — the bring-your-own-key seam. When given and embedder is a hosted-API spec ("openai:…", "cohere:…", …), the key is forwarded to that adapter, so the service itself never needs to hold one. None → the adapter’s usual environment-variable resolution. Ignored for key-less embedders (e.g. "hashing"). The corpus’s embedder keeps the key for the life of the corpus, so search() / retrieve() / explore_corpus() need no key of their own.

Returns:

the CorpusInfo of the freshly indexed corpus.

Raises:

ValueError – if corpus_id is already registered.

delete_corpus(corpus_id: str) → None[source]

Drop a corpus from the registry, releasing its index.

Raises:

KeyError – if corpus_id is not registered.

explore_corpus(corpus_id: str, *, dims: int = 2, projection_method: str = 'auto', cluster_method: str = 'kmeans', n_clusters: int = 8, label: bool = False) → ExploreResult[source]

Project & cluster a registered corpus — the corpus-map surface.

Runs ef.exploration.explore() over the corpus: every indexed segment is projected to dims coordinates and assigned a cluster, returned as a row-aligned ExploreResult (ids / coords / labels / cluster_titles) — the JSON-friendly shape an app_ef corpus map consumes.

Parameters:

  • corpus_id – the corpus to explore.

  • dims – projection target dimensionality — 2 or 3.

  • projection_method – "auto" / "umap" / "pca".

  • cluster_method – "kmeans" / "hdbscan".

  • n_clusters – number of k-means clusters.

  • label – when True, also name each cluster with an LLM (needs the ef[imbed] extra and a key); default False.

Raises:

  • KeyError – if corpus_id is not registered.

  • ValueError – if the corpus has fewer than 2 indexed segments.

list_corpora() → list[CorpusInfo][source]

Return the CorpusInfo of every registered corpus.

retrieve(corpus_id: str, query: str, *, limit: int = 10) → list[Segment][source]

Retrieve ranked Segments — the RAG-plug-in shape.

Like search(), but returns plain segments in rank order (the score dropped, the source_id folded into metadata["source"]) — clean context to hand to an external RAG/agent framework. ef returns context; it does not synthesize answers.

Raises:

KeyError – if corpus_id is not registered.

search(corpus_id: str, query: str, *, limit: int = 10) → list[SearchHit][source]

Search a registered corpus — up to limit ranked SearchHits.

Each hit carries the matched Segment, its similarity score (higher = closer) and the source_id it was cut from.

Raises:

KeyError – if corpus_id is not registered.

class ef.Embedder(*args, **kwargs)[source]

Structural type for an embedder: a batch callable plus metadata.

An embedder maps Iterable[str] to an ndarray of shape (n, dim). The four metadata attributes are the single source of truth on the object itself — no global registry, no config singleton.

model_id

Identity that bakes in everything affecting the vector — "openai:text-embedding-3-large@1024" (provider:model@dim). Works directly as a cache namespace.

Type:

str

dim

Output dimensionality. May be None until first call for lazily-probed embedders (e.g. a bare callable of unknown width).

Type:

int | None

normalized

True iff ||v|| == 1 by construction.

Type:

bool

honored_input_types

The InputType values this embedder acts on (empty if it ignores the task hint).

Type:

tuple[Literal[‘query’, ‘document’, ‘classification’, ‘clustering’], …]

embed_batch(texts: Iterable[str], *, input_type: Literal['query', 'document', 'classification', 'clustering'] | None = None, **backend: Any) → BatchHandle[source]

Submit an embed job, returning a BatchHandle.

exception ef.EmbedderError[source]

Raised for embedder-specific failures (shape mismatch, batch job failure).

class ef.ExploreResult[source]

A structured, JSON-friendly corpus-exploration result.

What explore() returns and ef.service.EfService.explore_corpus() serves. Every list is row-aligned: ids[i], coords[i] and labels[i] all describe the same item.

Keys:
ids: the per-item identifiers — a corpus’s keys, or positional "0",

"1", … for an id-less vector matrix.

coords: the projected coordinates — one [x, y] (or [x, y, z])

row per item.

labels: the cluster id of each item (HDBSCAN marks noise -1). cluster_titles: {cluster_id: title} — empty unless explore()

was called with label=True.

class ef.FunctionEmbedder(func: Callable[[...], Any], *, model_id: str | None = None, dim: int | None = None, normalized: bool = False, honored_input_types: Iterable[Literal['query', 'document', 'classification', 'clustering']] = (), pass_backend: bool = True)[source]

Promote a bare texts -> vectors callable to a full Embedder.

This is the bridge for embedder functions that carry no metadata — most importantly imbed’s registered embedders. The wrapped callable is asked only for func(texts) (it knows nothing of input_type); the wrapper advertises honored_input_types=() accordingly.

Parameters:

  • func – A callable mapping a list of strings to an array-like of shape (n, dim) (a numpy array or a nested sequence — both accepted).

  • model_id – Vector-identity string. Defaults to "function:<name>".

  • dim – Output width. If None it is inferred (and frozen) on the first call.

  • normalized – Whether func already returns unit vectors.

  • honored_input_types – Task hints func honors (usually none).

  • pass_backend – If True (default), forward any **backend kwargs to func; set False for callables with a strict (texts,) signature.

>>> import numpy as np
>>> e = FunctionEmbedder(lambda ts: np.ones((len(ts), 4)), model_id='ones@4')
>>> e(['a', 'b', 'c']).shape
(3, 4)
class ef.FunctionSegmenter(func: ~typing.Callable[[...], ~typing.Any], *, pass_doc: bool = False, count_tokens: ~typing.Callable[[str], int] | None = <function approx_token_count>, tokenizer: str = 'char-approx')[source]

Promote a bare text -> pieces callable to a full Segmenter.

The bridge for segmenter functions that carry no segment schema — most importantly imbed’s registered segmenters, which map text to an iterable of plain strings. Each piece the callable yields is normalized:

  • a str piece becomes a Segment; its character offsets are recovered by locating it in the source text, and a token count plus tokenizer metadata are attached;

  • a Mapping / SegmentRecord piece is coerced with as_segment(), gaining only an index / parent_id where it lacks them.

Parameters:

  • func – The callable to wrap. By default it receives the document’s text; with pass_doc=True it receives the whole document.

  • pass_doc – Pass the raw document instead of its text.

  • count_tokens – Token counter for str pieces (None disables the tokens / tokenizer annotation).

  • tokenizer – Name recorded as the tokenizer metadata key.

>>> fs = FunctionSegmenter(lambda text: text.split(','))
>>> [s['text'] for s in fs('a,b,c')]
['a', 'b', 'c']
>>> s = list(fs('a,b,c'))[1]
>>> (s['start'], s['end'], s['index'])
(2, 3, 1)
class ef.HashingEmbedder(*, dim: int = 512, ngram_range: tuple[int, int] = (1, 2), sublinear_tf: bool = True)[source]

A dependency-free embedder — the feature-hashing trick, numpy only.

HashingEmbedder is ef’s zero-install default: the embedder ingest() resolves to when none is given, so the headline ingest([...]).search(query) works on a bare pip install ef — no torch, no model download, no network.

It is a lexical embedder, not a neural one. Each text is tokenized into lowercased word n-grams; every token is hashed into one of dim buckets with a sign (the hashing trick — no vocabulary, no fitting, fixed memory); the signed, sublinear-tf-weighted bucket counts are L2-normalized. The cosine similarity of two vectors then ranks by shared vocabulary — enough for demos, doctests and offline tests, and a sane fallback in production. For genuine semantic quality install a real embedder (ef[sentence-transformers], ef[openai]) and pass it explicitly.

Every parameter that changes a vector is baked into model_id, so a HashingEmbedder’s artifacts stay correctly content-addressed in the ArtifactGraph.

Parameters:

  • dim – Output width. Wider → fewer hash collisions, larger vectors.

  • ngram_range – Inclusive (min, max) word-n-gram sizes. The default (1, 2) mixes single words with adjacent pairs — a little phrase sensitivity at no real cost.

  • sublinear_tf – Weight a token by 1 + log(count) rather than its raw count, damping a word repeated many times in one text.

>>> e = HashingEmbedder()
>>> e.model_id
'hashing:v1@512'
>>> import numpy as np
>>> v = e(['the quick brown fox'])
>>> v.shape
(1, 512)
>>> bool(np.isclose(np.linalg.norm(v[0]), 1.0))  # rows are L2-normalized
True
>>> a, b = e(['hello world']), e(['hello world'])
>>> round(float(a[0] @ b[0]), 5)  # identical text → cosine similarity 1.0
1.0
>>> isinstance(e, Embedder)
True
class ef.IndexedSource(source_id: str, doc_ids: tuple[str, ...], source_hashes: frozenset[str], config_hashes: frozenset[str])[source]

The provenance a vd collection records for one source document.

A source is segmented into several documents; they all carry the same source_id and should carry one source_hash / config_hash each. indexed_state() groups a collection’s documents into one of these per source — collecting the hashes as sets so a partially re-indexed source (documents disagreeing on a hash) is detectable rather than papered over.

source_id

the corpus key of the source these documents were cut from.

Type:

str

doc_ids

the ids of every vd document belonging to the source.

Type:

tuple[str, …]

source_hashes

the distinct source_hash values its documents record — a single-element set for a cleanly indexed source.

Type:

frozenset[str]

config_hashes

likewise the distinct config_hash values.

Type:

frozenset[str]

class ef.MultiEmbedder(routes: Mapping[Hashable, Embedder], predicate: Callable[[str], Hashable], *, default: Embedder | None = None)[source]

Route each text to one of several embedders by a predicate.

Useful for domain-specific models (e.g. a code embedder for code, a prose embedder for prose). All routes must share the same dim — the output is a single (n, dim) array, with each row produced by its text’s route.

Parameters:

  • routes – {route_key: Embedder}.

  • predicate – text -> route_key — picks a route per text.

  • default – Embedder for texts whose route_key is not in routes (otherwise an unknown key raises EmbedderError).

>>> import numpy as np
>>> from ef.embedders import FunctionEmbedder
>>> short = FunctionEmbedder(lambda ts: np.zeros((len(ts), 2)), model_id='s@2')
>>> long = FunctionEmbedder(lambda ts: np.ones((len(ts), 2)), model_id='l@2')
>>> m = MultiEmbedder({'s': short, 'l': long},
...                   predicate=lambda t: 's' if len(t) < 4 else 'l')
>>> m(['hi', 'hello']).tolist()
[[0.0, 0.0], [1.0, 1.0]]
class ef.NormalizingEmbedder(inner: Embedder)[source]

L2-normalize an inner embedder’s output.

Use this for MRL-truncated vectors (which lose unit norm when sliced to a smaller dimension) or any backend that does not normalize by construction. Reports normalized=True; cache_key() keys on normalized, so a normalized and an un-normalized view of the same model never collide in a shared cache.

Parameters:

inner – The embedder whose output to renormalize.

property honored_input_types: tuple[Literal['query', 'document', 'classification', 'clustering'], ...]

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable’s items.

If the argument is a tuple, the return value is the same object.

ef.OpKey

alias of str

class ef.PipelineSpec(segment: TransformSpec, embed: TransformSpec)[source]

A whole indexing pipeline as data — the segment→embed chain.

A PipelineSpec is the declarative description of one config: a TransformSpec for the segmenter and one for the embedder. It is fully serializable (as_dict() / from_dict()) and content-hashed by config_id(). Registering a second PipelineSpec that differs only in its embed step shares the segment artifacts in the ArtifactGraph — that is config branching, and the graph gives it for free.

The segment``→``embed ordering is the input wiring: the embed step consumes the segment step’s output. (Phase 5 fixes this two-step shape; a later phase may generalize it to an arbitrary step list.)

>>> seg = TransformSpec(op='segment:r', op_version='1', params={})
>>> emb = TransformSpec(op='embed:m@8', op_version='1', params={'input_type': 'document'})
>>> spec = PipelineSpec(segment=seg, embed=emb)
>>> spec.embed.params['input_type']
'document'
>>> PipelineSpec.from_dict(spec.as_dict()) == spec
True
as_dict() → dict[str, Any][source]

Return a plain-dict (JSON-serializable) form of the pipeline.

>>> seg = TransformSpec(op='segment:r', op_version='1', params={})
>>> emb = TransformSpec(op='embed:m', op_version='1', params={})
>>> PipelineSpec(segment=seg, embed=emb).as_dict()['segment']['op']
'segment:r'
classmethod from_dict(d: Mapping[str, Any]) → PipelineSpec[source]

Rebuild a PipelineSpec from its as_dict() form.

class ef.ProducerSpec(op: str, op_version: str, inputs: tuple[str, ...]=(), params: Mapping[str, ~typing.Any]=<factory>)[source]

A declarative recipe for one produced artifact — the graph’s node type.

A ProducerSpec says: this artifact is produced by op op at version op_version, applied to the artifacts inputs (passed positionally), with the keyword parameters params. It does not hold the producer callable — only the string op key — so a spec is fully serializable (as_dict() / from_dict()) and the graph’s producers registry can be persisted.

The recipe is the artifact’s identity: artifact_id() hashes exactly these four fields. Treat an instance as immutable — it is frozen, and mutating params in place would silently desynchronize it from its id.

op

the op key, resolved to a callable via the graph’s ops registry.

Type:

str

op_version

the op’s version string. It participates in the id, so a behavior change to an op must be paired with a version bump — reusing a version for changed behavior silently serves stale cached values (the graph trusts the version; it cannot inspect the code).

Type:

str

inputs

the input artifact ids, in the order the op receives them positionally. Order is significant — it is part of the id.

Type:

tuple[str, …]

params

the op’s keyword arguments. Key order is not significant (the id is computed over canonical, sorted-key JSON); pass defaults-filled, fully-named params so that two semantically equal calls hash equal.

Type:

collections.abc.Mapping[str, Any]

>>> spec = producer_spec('embed', 'seg-1', op_version='3', model='small')
>>> spec.op, spec.inputs, spec.op_version
('embed', ('seg-1',), '3')
>>> spec.params == {'model': 'small'}
True
as_dict() → dict[str, Any][source]

Return a plain-dict form of the spec — JSON-serializable.

The inverse of from_dict(). Use it to persist a producers registry to a dol JSON store.

>>> producer_spec('embed', 'a', 'b', op_version='1', dim=256).as_dict()
{'op': 'embed', 'op_version': '1', 'inputs': ['a', 'b'], 'params': {'dim': 256}}
classmethod from_dict(d: Mapping[str, Any]) → ProducerSpec[source]

Rebuild a ProducerSpec from its as_dict() form.

>>> spec = producer_spec('embed', 'a', 'b', op_version='1', dim=256)
>>> ProducerSpec.from_dict(spec.as_dict()) == spec
True
class ef.RagEvalReport(metrics: Mapping[str, float], coverage: Mapping[str, int], n_samples: int)[source]

The outcome of evaluate_rag() — lexical RAG metrics over a sample set.

metrics maps each metric to its mean over the samples it applied to; coverage records how many samples that was (a metric needing a reference skips samples without one). n_samples is the total scored.

>>> report = RagEvalReport(
...     metrics={'exact_match': 0.5}, coverage={'exact_match': 4}, n_samples=4)
>>> report.metrics['exact_match']
0.5
class ef.RagSample[source]

One end-to-end RAG interaction — the unit evaluate_rag() scores.

The field names match Ragas’ SingleTurnSample exactly, so a RagSample is the interchange shape app_ef / srag / raglab produce and as_ragas_dataset() forwards unchanged. ef itself fills only retrieved_contexts (from retrieve()); response is the caller’s LLM output — ef does not synthesize it.

Keys:

user_input: the user’s question. response: the generated answer (the caller’s LLM produced it). retrieved_contexts: the context passages handed to the LLM — what

retrieve() returned, as text.

reference: the ground-truth answer, if known. reference_contexts: the ground-truth context passages, if known.

class ef.RecursiveCharacterSegmenter(*, chunk_size: int = 512, chunk_overlap: int = 64, separators: ~typing.Sequence[str] = ('\n\n', '\n', '. ', ', ', ' ', ''), count_tokens: ~typing.Callable[[str], int] = <function approx_token_count>, tokenizer: str = 'char-approx')[source]

The default segmenter — recursive character splitting.

Splits a document on a descending ladder of separators (paragraph → line → sentence → clause → word → character), merging the resulting pieces into chunks of about chunk_size tokens with chunk_overlap tokens shared between neighbours. This “good enough” default is, on realistic corpora, within a point or two of far more elaborate methods — upgrade only if you can measure the win.

Sizes are measured with count_tokens. The default counter is the dependency-free approx_token_count(); inject a real tokenizer (and its tokenizer name) when an exact count matters. Every emitted segment records the tokenizer in its metadata and carries exact start / end character offsets into the source.

Parameters:

  • chunk_size – Target chunk size, in count_tokens units.

  • chunk_overlap – Tokens of overlap between consecutive chunks (must be smaller than chunk_size).

  • separators – The separator ladder, coarsest first; the last should be "" so an unsplittable run is still broken between characters.

  • count_tokens – Maps a string to a token count.

  • tokenizer – Name recorded as the tokenizer metadata key.

>>> seg = RecursiveCharacterSegmenter(chunk_size=3, chunk_overlap=1)
>>> [c['text'] for c in seg('one two three four')]
['one two three', 'three four']
class ef.RefreshPlan(to_materialize: tuple[str, ...] = (), to_delete: tuple[str, ...] = ())[source]

What a refresh will do — plan_refresh()’s output.

The two fields are disjoint in intent but may overlap in membership: a stale source is in both — its old documents are deleted and it is re-materialized. An orphan is only in to_delete; a brand-new source only in to_materialize.

to_materialize

source ids to (re-)run through the pipeline.

Type:

tuple[str, …]

to_delete

source ids whose currently-indexed documents are removed (before re-materializing, for the ones that also reappear above).

Type:

tuple[str, …]

class ef.RefreshReport(config: str, mode: Literal['none', 'incremental', 'full', 'scoped_full'], added: tuple[str, ...] = (), modified: tuple[str, ...] = (), deleted: tuple[str, ...] = (), unchanged: tuple[str, ...] = (), documents_written: int = 0, documents_removed: int = 0, artifacts_removed: int = 0)[source]

The outcome of a refresh — what SourceManager.refresh() returns.

added / modified / deleted / unchanged partition the source ids the refresh considered (mirroring CorpusDiff field names): newly indexed, re-indexed (was stale/misconfigured), removed (orphan), and already-fresh-left-alone. The three counts report the work done. Truthiness reports whether anything changed.

>>> r = RefreshReport(config='c0', mode='full', added=('a',), modified=('b',))
>>> bool(r), r.changed
(True, ('a', 'b'))
>>> bool(RefreshReport(config='c0', mode='full', unchanged=('a',)))
False
property changed: tuple[str, ...]

The source ids that were added, modified or deleted.

class ef.Reranker(*args, **kwargs)[source]

A callable that re-scores (query, segments) — higher means more relevant.

The structural contract a reranker satisfies: given the query and a list of candidate Segments, return one score per segment, in the same order. rerank() sorts by these scores. It is a @runtime_checkable Protocol, not a base class — a plain function of the right shape already is a Reranker.

>>> def by_length(query, segments):
...     return [len(s['text']) for s in segments]
>>> isinstance(by_length, Reranker)
True
class ef.RetrievalEvalReport(metrics: Mapping[str, float], per_query: Mapping[str, Mapping[str, float]], n_queries: int, k_values: tuple[int, ...])[source]

The outcome of evaluate_retrieval() — metrics over a query set.

metrics maps a "<name>@<k>" key (e.g. "ndcg@10") to the mean of that metric over every scored query; per_query keeps the same keys per query for drill-down. n_queries is how many queries were actually scored (a query with no positive judgement is skipped).

>>> report = RetrievalEvalReport(
...     metrics={'ndcg@10': 0.83}, per_query={}, n_queries=12, k_values=(10,))
>>> report.primary
0.83
property primary: float | None

The headline number — mean NDCG@10, or None if it was not computed.

primary_at(k: int) → float | None[source]

Mean NDCG at cutoff k — None if NDCG@``k`` was not computed.

class ef.RetryPolicy(max_attempts: int = 4, base_delay: float = 1.0, max_delay: float = 30.0, jitter: float = 0.3, retry_on: ~typing.Callable[[BaseException], bool] = <function _default_is_retryable>)[source]

Exponential-backoff retry policy for RetryingEmbedder.

max_attempts

Total tries (the first call counts as attempt 1).

Type:

int

base_delay

Seconds before the first retry; doubles each attempt.

Type:

float

max_delay

Upper bound on any single backoff sleep.

Type:

float

jitter

Random fraction (0..1) of the delay added on top, to de-correlate concurrent clients.

Type:

float

retry_on

Predicate deciding whether an exception is worth retrying.

Type:

Callable[[BaseException], bool]

retry_on() → bool

Heuristic: retry transient failures (429, 5xx, connection/timeout).

Per-request 400-class errors are not retried — the same request would fail identically. Recognizes the status_code/status attribute carried by most HTTP-client SDK exceptions (incl. the OpenAI SDK).

class ef.RetryingEmbedder(inner: ~ef.embedders.Embedder, policy: ~ef.embedder_wrappers.RetryPolicy | None = None, *, sleep: ~typing.Callable[[float], None] = <built-in function sleep>)[source]

Retry an inner embedder’s calls on transient failures.

Parameters:

  • inner – The embedder to guard.

  • policy – A RetryPolicy (sensible defaults if omitted).

  • sleep – The sleep function — injectable so tests run instantly.

property honored_input_types: tuple[Literal['query', 'document', 'classification', 'clustering'], ...]

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable’s items.

If the argument is a tuple, the return value is the same object.

property normalized: bool

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

class ef.SearchHit(segment: Segment, score: float, source_id: str | None = None)[source]

One ranked search result — a Segment and its score.

SearchableCorpus.search() / SourceManager.search() return a ranked list of these. Keeping Segment (the interchange TypedDict) pure of a result-only score key, a SearchHit wraps it alongside the similarity score and the source_id of the document the segment was cut from.

segment

the matched segment — the canonical Segment.

Type:

ef.segments.Segment

score

the similarity score from the vector store (higher = closer).

Type:

float

source_id

the corpus key of the source document, if recorded.

Type:

str | None

>>> hit = SearchHit(segment={'text': 'hi', 'id': 'x'}, score=0.9, source_id='doc-1')
>>> hit.segment['text'], hit.score, hit.source_id
('hi', 0.9, 'doc-1')
class ef.SearchableCorpus(collection: Any, embedder: Embedder, *, config: str | None = None)[source]

A ready-to-search view over one indexed config — the thin facade.

What ingest() returns and what SourceManager.searchable() hands out: a small, read-mostly object that knows just enough to answer a query — the vd collection holding the vectors and the Embedder that embeds the query. It does not re-index; mutation, refresh and multi-config live on SourceManager.

Parameters:

  • collection – the vd collection holding this config’s vectors.

  • embedder – the embedder — used to embed queries with input_type="query".

  • config – the ConfigId of the indexed pipeline, if known.

>>> # built for you by ingest(); see the module docstring for a full example.
retrieve(query: str | Iterable[float], *, limit: int = 10, filter: Mapping[str, Any] | None = None) → list[Segment][source]

Retrieve ranked Segments — the RAG-plug-in surface.

The name an external RAG/agent framework reaches for. Where search() returns scored SearchHits (for inspecting the ranking), retrieve returns plain segments in rank order — clean context to hand to an LLM, with no ef-specific type to learn (design notes §F5). ef returns the context; it does not synthesize answers. See hits_to_segments() for how the score is dropped and the source_id provenance is preserved in metadata["source"].

>>> # idx.retrieve('a query') -> [{'text': ..., 'id': ..., 'metadata': ...}, ...]
search(query: str | Iterable[float], *, limit: int = 10, filter: Mapping[str, Any] | None = None) → list[SearchHit][source]

Search the indexed corpus — return up to limit ranked SearchHits.

query is text (embedded with input_type="query") or an already computed query vector. filter is a vd metadata filter (MongoDB-style) — see vd.filters.

class ef.Segment[source]

A piece of text carved from a source document — the interchange type.

A TypedDict: a Segment is a plain dict, which keeps it cheap to create and stream. total=False because most keys are optional; by convention text is always present and id is always set (derived from text if not supplied).

Keys:

text: The segment’s text. Required. id: Stable identifier — content-derived by default (segment_id()). parent_id: Id of the source document or parent segment this came from. start: Character offset of the segment’s start in the source text. end: Character offset of the segment’s end (exclusive) in the source. index: Ordinal position of the segment in its segmenter’s output. tokens: Token count of text (meaningful only with metadata key

tokenizer recording which tokenizer counted it).

metadata: Free-form mapping; framework-specific keys live here. See

PROMOTED_METADATA_KEYS for the conventional keys.

class ef.SegmentRecord(text: str, id: str = '', parent_id: str | None = None, start: int | None = None, end: int | None = None, index: int | None = None, tokens: int | None = None, metadata: Mapping[str, ~typing.Any]=<factory>)[source]

Frozen-dataclass view over a Segment — the convenience surface.

Use this when attribute access and immutability read better than dict access; use the Segment TypedDict directly on the hot path. The two convert losslessly via to_segment() and segment_record().

id is derived from text (and metadata) if left empty.

>>> rec = SegmentRecord('hello', index=2)
>>> rec.id == segment_id('hello')
True
>>> rec.to_segment()['index']
2
to_segment() → Segment[source]

Return an equivalent Segment dict (dropping empty keys).

class ef.Segmenter(*args, **kwargs)[source]

Structural type for a segmenter: str | Mapping -> Iterable[Segment].

A segmenter is just a callable. It accepts a document — a bare str or a Mapping with a text key (and optionally an id, which becomes the emitted segments’ parent_id) — and yields Segment pieces. It returns an iterable, not a list: streaming is the default so the heavy case scales.

Because the protocol’s only member is __call__, every callable satisfies isinstance(x, Segmenter). Treat it as documentation of intent, not a discriminating check — as_segmenter() is the seam that turns an arbitrary callable into a well-behaved segmenter.

class ef.SourceManager(corpus: Any = None, *, segmenter: Any = None, embedder: Any = None, store: Any = None, cache: MutableMapping[str, Any] | None = None, auto_refresh: bool = False, embedder_api_key: str | None = None)[source]

The heavy search facade — multi-config corpus indexing over one corpus.

A SourceManager holds a corpus, an ArtifactGraph and one or more configs. A config is a named segmenter+embedder pipeline (PipelineSpec); register_config() adds one, materialize() runs the corpus through it into a vd collection, and search() queries it. Because every artifact is content-addressed in the shared graph, two configs that share a step (e.g. the same segmenter) compute that step’s artifacts once — config branching is free.

The constructor optionally registers a "default" config: pass an embedder (and optionally a segmenter) and it is registered eagerly; omit embedder to build every config explicitly via register_config().

Parameters:

  • corpus – the source corpus — coerced by as_corpus() (a mapping, a directory path, an iterable of sources, or None).

  • segmenter – the default config’s segmenter — coerced by as_segmenter() (None → the recursive default). Used only when embedder is also given.

  • embedder – the default config’s embedder — coerced by as_embedder(). If given, a "default" config is registered; if None, no default config is registered.

  • embedder_api_key – an optional API key for the default config’s embedder — forwarded to a hosted-API adapter ("openai:…" etc.) when the embedder is such a spec. The bring-your-own-key seam; ignored for key-less embedders. See register_config().

  • store – where the vectors go. None → an in-memory vd backend; a vd client → collections are created on it; a vd collection → used directly (then only one config is supported); a backend-name string → vd.connect of that backend.

  • cache – the ArtifactGraph’s value cache — any MutableMapping; None → an in-RAM dict.

  • auto_refresh – when True, the corpus is wrapped in a ChangeDetectingCorpus and every edit made through it is incrementally re-indexed into each materialized config — the index stays live without an explicit refresh(). scan() then also picks up out-of-band edits.

>>> import numpy as np
>>> from ef import as_embedder
>>> e1 = as_embedder(lambda ts: np.array([[len(t)] for t in ts], float), model_id='len@1')
>>> sm = SourceManager(['hello', 'hi there'], embedder=e1)
>>> report = sm.materialize()
>>> report['segments']
2
>>> len(sm.search('hello')) >= 1
True
property configs: dict[str, str]

The registered configs as a {name: config_id} mapping.

diagnose(config: str | None = None) → StalenessReport[source]

Report how far one config’s index has drifted from the corpus.

Computes the four staleness conditions — orphan / missing / stale / misconfigured — by comparing the config’s vd collection against the current corpus (see ef.diagnostics). Read-only: it changes nothing. refresh() acts on the result.

>>> sm = SourceManager({'a': 'alpha', 'b': 'beta'}, embedder='hashing')
>>> _ = sm.materialize()
>>> bool(sm.diagnose())                  # freshly indexed — in sync
False
>>> sm.corpus['a'] = 'alpha rewritten'
>>> sm.diagnose().stale
('a',)
gc_orphans(config: str | None = None) → int[source]

Delete the documents of sources that left the corpus — return the count.

The garbage-collection slice of refresh(): it removes only orphan documents (a source gone from the corpus) and prunes the dead leaves out of the ArtifactGraph, touching nothing else. gc_orphans() is the orphan-deletion half of refresh(mode='full').

lineage(key: str) → frozenset[str][source]

The artifacts key was produced from — its provenance.

A thin pass-through to ArtifactGraph.ancestors: “what produced this vector / segment?”. The leaves of the returned set are the source content hashes the artifact ultimately derives from.

materialize(config: str | None = None, *, sources: Iterable[str] | None = None) → dict[str, Any][source]

Index the corpus through one config (or all) — corpus → segment → embed → vd.

For each source and each selected config: the source is added to the ArtifactGraph as a content-addressed leaf, its segment and embed artifacts are materialize()d (a cache hit if already computed — so a shared segment step runs once across configs), and every (segment, vector) pair is written into the config’s vd collection. Idempotent: re-materializing rewrites identical documents.

Parameters:

  • config – the config to index — a name, a ConfigId, or None for every registered config.

  • sources – an optional subset of source ids to index — None (the default) indexes the whole corpus. refresh() uses this to re-index only the sources that changed; ids not in the corpus are skipped.

Returns:

a report dict with configs / sources / segments counts.

pipeline(config: str | None = None) → PipelineSpec[source]

The PipelineSpec of a registered config.

rebuild(config: str | None = None) → RefreshReport[source]

Drop one config’s index entirely and re-index the whole corpus.

The heavy hammer: every document in the config’s vd collection is deleted, then the whole corpus is re-materialize()d from scratch. Use it to recover from a config whose vd documents were produced by an earlier pipeline (every source misconfigured), or whenever a guaranteed-clean index is wanted over the surgical refresh(). Every source is reported as added — the collection was emptied and rebuilt.

Returns:

a RefreshReport.

refresh(config: str | None = None, *, sources: Iterable[str] | None = None, mode: Literal['none', 'incremental', 'full', 'scoped_full'] = 'full') → RefreshReport[source]

Re-sync one config’s vd index with the current corpus.

The explicit-refresh entry point. It diagnose()s the config, turns the staleness report into a RefreshPlan for the chosen mode (plan_refresh()), deletes the stale/orphan documents, re-materialize()s what changed, and prunes the dead leaves out of the ArtifactGraph. Because it re-reads and re-hashes the corpus, it catches out-of-band edits without any change-detection wrapper.

Parameters:

  • config – the config to refresh (name / id / None for the default or sole config).

  • sources – an optional subset of source ids to refresh — sources outside it are left untouched. Pair it with mode='scoped_full' (or 'incremental'); 'full' rejects it.

  • mode – one of REFRESH_MODES — 'none' (index new/changed content, delete nothing), 'incremental' (also replace changed sources), 'full' (also delete orphans — the corpus is authoritative), 'scoped_full' (full over the sources subset). Default 'full'.

Returns:

a RefreshReport.

>>> sm = SourceManager({'a': 'alpha text', 'b': 'beta text'}, embedder='hashing')
>>> _ = sm.materialize()
>>> sm.corpus['b'] = 'beta rewritten'        # edit
>>> sm.corpus['c'] = 'gamma text'            # add
>>> del sm.corpus['a']                       # delete
>>> report = sm.refresh()
>>> report.added, report.modified, report.deleted
(('c',), ('b',), ('a',))
>>> bool(sm.diagnose())                      # in sync again
False
register_config(name: str = 'default', *, segmenter: Any = None, embedder: Any = None, embedder_api_key: str | None = None) → str[source]

Register a named segmenter+embedder pipeline; return its ConfigId.

The segmenter and embedder are coerced through ef’s DI seams (as_segmenter() / as_embedder(); an absent embedder defaults to DEFAULT_EMBEDDER). Their ops are registered into the shared ArtifactGraph and a vd collection is created for the config. Registering a config does not index it — call materialize().

embedder_api_key is the per-call bring-your-own-key seam: when given and embedder is a hosted-API provider spec ("openai:…" etc.), the key is forwarded to that adapter instead of relying on a server environment variable. Ignored for key-less embedders.

Re-registering the same name replaces it. Two configs that resolve to the same PipelineSpec share one ConfigId (and one collection).

>>> import numpy as np
>>> from ef import as_embedder
>>> sm = SourceManager(['a', 'b'])
>>> cid = sm.register_config(
...     'mini', embedder=as_embedder(
...         lambda ts: np.ones((len(ts), 2)), model_id='ones@2'))
>>> len(cid)
64
retrieve(query: str | Iterable[float], *, config: str | None = None, limit: int = 10, filter: Mapping[str, Any] | None = None) → list[Segment][source]

Retrieve ranked Segments — the RAG-plug-in surface.

The SourceManager counterpart of SearchableCorpus.retrieve(): plain segments in rank order, the clean context shape an external RAG/agent framework consumes (ef does not synthesize answers). search() is the scored counterpart.

scan() → CorpusDiff[source]

Re-scan the corpus for out-of-band edits — needs auto_refresh=True.

Picks up changes made to the backing store directly (a file edited on disk, an S3 object replaced) rather than through the corpus wrapper. When the manager was built with auto_refresh=True the corpus is a ChangeDetectingCorpus; this calls its scan(), which fires a ChangeEvent per drift — each one auto-applied to every materialized config. Returns the CorpusDiff.

For a large bulk of out-of-band edits, prefer refresh() (one corpus pass) over scan (one collection pass per changed source).

Raises:

TypeError – if the corpus is not change-detecting (the manager was not built with auto_refresh=True).

search(query: str | Iterable[float], *, config: str | None = None, limit: int = 10, filter: Mapping[str, Any] | None = None) → list[SearchHit][source]

Search one config’s index — return up to limit ranked SearchHits.

config selects which config to query (a name / ConfigId, or None for the default or sole config). query is text or a pre-computed query vector; filter is a vd metadata filter.

searchable(config: str | None = None) → SearchableCorpus[source]

Return a thin SearchableCorpus bound to one config’s index.

class ef.StalenessReport(config: str, orphan: tuple[str, ...] = (), missing: tuple[str, ...] = (), stale: tuple[str, ...] = (), misconfigured: tuple[str, ...] = (), fresh: tuple[str, ...] = ())[source]

The four staleness conditions of one config — what diagnose() returns.

Each field is a sorted tuple of source_ids, so a report is deterministic and JSON-friendly. orphan and missing are disjoint from the rest (a source absent from one side cannot also be compared); a source can be both stale and misconfigured (its content changed and the config did) — the conditions are reported independently. fresh is the complement: indexed, current and built by this config.

Truthiness reports whether anything is wrong — bool(report) is False only for a perfectly in-sync index.

>>> r = StalenessReport(config='c0', stale=('a',), fresh=('b',))
>>> bool(r), r.needs_reindexing
(True, ('a',))
>>> bool(StalenessReport(config='c0', fresh=('a', 'b')))
False
property needs_reindexing: tuple[str, ...]

Sources whose vectors must be recomputed — stale ∪ misconfigured.

class ef.TransformSpec(op: str, op_version: str, params: Mapping[str, ~typing.Any]=<factory>)[source]

One step of a pipeline — a serializable recipe for an op application.

A TransformSpec names which op (op, a key resolved through the ArtifactGraph’s ops registry), at which version (op_version — it participates in every downstream hash), and with which keyword params (params). It does not record the step’s inputs: those are wired structurally by the PipelineSpec (the embed step consumes the segment step’s output). It is the per-step counterpart of a ProducerSpec.

The op key carries the component identity — e.g. "embed:openai:text- embedding-3-large@1024" or "segment:<segmenter-identity>" (see ef.ops) — so two configs with different components get disjoint artifact cones automatically.

>>> ts = TransformSpec(op='embed:m@8', op_version='2', params={'input_type': 'document'})
>>> ts.op, ts.op_version
('embed:m@8', '2')
>>> TransformSpec.from_dict(ts.as_dict()) == ts
True
as_dict() → dict[str, Any][source]

Return a plain-dict (JSON-serializable) form of the spec.

>>> TransformSpec(op='segment:r', op_version='1', params={}).as_dict()
{'op': 'segment:r', 'op_version': '1', 'params': {}}
classmethod from_dict(d: Mapping[str, Any]) → TransformSpec[source]

Rebuild a TransformSpec from its as_dict() form.

ef.approx_token_count(text: str) → int[source]

Rough token count — the common ~4 characters per token heuristic.

A dependency-free default so ef can talk in “tokens” without pulling in tiktoken or a model tokenizer. Inject a real counter (and its name) into a segmenter when an exact count matters.

>>> approx_token_count('a' * 40)
10
>>> approx_token_count('')
0
ef.artifact_id(spec: ProducerSpec) → str[source]

Content-addressed id of spec — sha256 over its four fields.

The id is sha256 of canonical (sorted-key) JSON of {op, op_version, inputs, params}. Two specs that describe the same computation hash identically; any change to the op, its version, its inputs (including their order) or its params yields a different id. That is what makes the graph content-addressed: the id is the recipe.

>>> a = producer_spec('embed', 'seg-1', op_version='2', model='small')
>>> artifact_id(a) == artifact_id(producer_spec(
...     'embed', 'seg-1', op_version='2', model='small'))
True
>>> len(artifact_id(a))
64
>>> artifact_id(a) == artifact_id(producer_spec(
...     'embed', 'seg-1', op_version='3', model='small'))   # version bump
False
>>> artifact_id(producer_spec('op', 'x', 'y', op_version='1')) == artifact_id(
...     producer_spec('op', 'y', 'x', op_version='1'))      # input order matters
False
ef.as_corpus(x: Any = None, /, **kwargs: Any)[source]

Normalize x into a corpus (MutableMapping[source_id, Source]).

The single dependency-injection seam through which every ef entry point accepts a user-supplied corpus — the mirror of as_embedder() and as_segmenter(). Accepts, in order:

  1. None — a fresh empty in-RAM dict;

  2. a Mapping — returned unchanged (a dict, a dol store, another corpus: it already is a corpus);

  3. a directory path (str / os.PathLike naming an existing directory) — a dol filesystem store of the text files under it (dol is imported lazily; only this branch needs it);

  4. an iterable of sources (strings or mappings) — an in-RAM dict keyed by each source’s content_hash() (content-addressed, so duplicate sources collapse to one entry).

Extra **kwargs are forwarded to the dol filesystem store (the directory branch only) and ignored otherwise.

Raises:

TypeError – if x is a non-directory string, or none of the above.

>>> as_corpus()
{}
>>> as_corpus({'a': 'x'})                 # a mapping passes straight through
{'a': 'x'}
>>> sorted(as_corpus(['alpha', 'beta']).values())
['alpha', 'beta']
>>> len(as_corpus(['dup', 'dup']))        # content-addressed: duplicates collapse
1
ef.as_embedder(x: Any, **kwargs: Any) → Embedder[source]

Normalize x into an Embedder — the DI seam.

The single place every ef entry point coerces a user-supplied embedder argument. Accepts, in order:

  1. an existing Embedder — returned unchanged;

  2. a URL string (http:// / https://) — http_embedder();

  3. a provider-prefixed string — "openai:<model>", "cohere:<model>", "voyage:<model>" or "gemini:<model>" — the matching hosted-API adapter;

  4. the bare string "hashing" — the dependency-free HashingEmbedder (ef’s zero-install default);

  5. an "st:<model>" string, or any other bare model name — sentence_transformers_embedder() (the local neural default);

  6. a bare callable — wrapped in FunctionEmbedder (the imbed-embedder bridge).

Extra **kwargs are forwarded to the chosen factory.

Raises:

TypeError – if x is none of the above.

>>> import numpy as np
>>> e = as_embedder(lambda ts: np.zeros((len(ts), 2)), model_id='zero@2')
>>> type(e).__name__
'FunctionEmbedder'
ef.as_ragas_dataset(samples: Iterable[RagSample | Mapping[str, Any]]) → Any[source]

Convert RagSamples into a Ragas EvaluationDataset.

The hookpoint for the LLM-judged RAG metrics ef deliberately does not compute itself (faithfulness, answer relevancy, …). RagSample already mirrors Ragas’ SingleTurnSample field-for-field, so this is a thin, lossless adapter. Run the evaluation with your own LLM:

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

dataset = as_ragas_dataset(samples)
result = evaluate(dataset, metrics=[faithfulness, answer_relevancy])
Parameters:

samples – an iterable of RagSamples (or plain mappings).

Returns:

a ragas.EvaluationDataset.

Raises:

ImportError – if the optional ragas package is not installed.

ef.as_segment(x: Any, **defaults: Any) → Segment[source]

Coerce x into a Segment — accepts a str, Mapping or record.

This is the segment-side normalizer that lets a Segmenter accept loosely-typed output (a bare string, a partial dict) and still emit a well-formed Segment. **defaults supplies fallback values: a default fills a field only when x does not already carry it.

Accepts:

  1. a SegmentRecord — converted via SegmentRecord.to_segment();

  2. a str — becomes the text of a fresh segment;

  3. a Mapping with a text key — its keys are carried through.

The id is always present on the result (derived if absent).

Raises:

  • TypeError – if x is none of the accepted forms.

  • ValueError – if x is a Mapping without a text key.

>>> as_segment('plain text', index=3)['index']
3
>>> as_segment({'text': 'kept', 'index': 0}, index=9)['index']  # x wins
0
ef.as_segmenter(x: Any = None, /, **kwargs: Any) → Segmenter[source]

Normalize x into a Segmenter — the DI seam.

The single place every ef entry point coerces a user-supplied segmenter argument. Accepts, in order:

  1. None or "default" / "recursive" — a RecursiveCharacterSegmenter (the default);

  2. "lines" — line_segmenter();

  3. any other string — looked up in imbed’s registry via imbed_segmenter();

  4. a ready-made segmenter (a BaseSegmenter or a callable marked by ef) — returned unchanged;

  5. a bare callable — wrapped in FunctionSegmenter.

Extra **kwargs are forwarded to the chosen factory (ignored when x is already a segmenter).

Raises:

TypeError – if x is none of the above.

>>> seg = as_segmenter(chunk_size=256, chunk_overlap=32)
>>> (seg.chunk_size, seg.chunk_overlap)
(256, 32)
ef.average_precision(ranked: Sequence[str], relevant: Mapping[str, float], k: int | None = None) → float[source]

Average precision of ranked — the per-query term of MAP.

The mean of the precision measured at each relevant result’s rank. k optionally caps the ranking.

>>> average_precision(['d1', 'x', 'd2'], {'d1': 1.0, 'd2': 1.0})
0.8333333333333333
ef.cache_key(embedder: Embedder, text: str, input_type: Literal['query', 'document', 'classification', 'clustering'] | None = None) → str[source]

Deterministic cache key for one (embedder, text, input_type) triple.

The key pins everything that changes the resulting vector. model_id is the embedder-identity SSOT — it already bakes in provider, model and dim — so dim is not keyed separately (keying it would also break FunctionEmbedder’s lazy dim inference). normalized is keyed because a NormalizingEmbedder shares its inner’s model_id yet yields different vectors.

Returns "<model_id>/<sha256>" — a namespaced key safe for any MutableMapping store.

>>> class _E:
...     model_id, normalized = 'm@8', False
>>> k1 = cache_key(_E(), 'hello', 'document')
>>> k2 = cache_key(_E(), 'hello', 'query')
>>> k1 == k2  # input_type participates
False
>>> k1.startswith('m@8/')
True
ef.cluster(data: Any, *, method: Literal['kmeans', 'hdbscan'] = 'kmeans', n_clusters: int = 8, min_cluster_size: int = 5, embedder: Any = None, normalize: bool = True, random_state: int = 42) → ndarray[source]

Cluster a corpus’s embeddings (use cases §G2).

method='kmeans' is the default and is dependency-free — a numpy Lloyd’s algorithm with k-means++ seeding. method='hdbscan' finds a density-based clustering (and a noise label -1) and needs the ef[explore] extra.

Parameters:

  • data – what to cluster — see project() / _resolve_vectors().

  • method – 'kmeans' (numpy-only) or 'hdbscan' (optional extra).

  • n_clusters – number of clusters for k-means (clamped to n_samples).

  • min_cluster_size – smallest cluster HDBSCAN will form.

  • embedder – embedder used when data is raw text.

  • normalize – L2-normalize the vectors first, so Euclidean distance behaves like cosine similarity — the right default for semantic embeddings. Set False to cluster raw vectors.

  • random_state – seed for k-means initialization.

Returns:

An ndarray of integer cluster labels, shape (n_samples,), in input order. HDBSCAN labels unclustered points -1.

>>> import numpy as np
>>> vectors = np.random.RandomState(2).rand(15, 8)
>>> labels = cluster(vectors, method='kmeans', n_clusters=3, random_state=0)
>>> labels.shape
(15,)
>>> set(labels.tolist()) <= {0, 1, 2}
True
ef.cohere_embedder(model: str = 'embed-v4.0', *, dim: int | None = None, api_key: str | None = None, batch_size: int = 96, timeout: float = 60.0) → Embedder[source]

Build a Cohere-embeddings Embedder.

Talks to Cohere’s /v2/embed REST endpoint directly — no cohere SDK, no extra dependency. Cohere’s v3+ models require a task hint; this adapter supplies one for every canonical InputType and defaults to the document role when none is given.

Parameters:

  • model – A Cohere embedding model name.

  • dim – Matryoshka-truncated width (embed-v4.0 only). Omit for the model’s native width. The dim is part of the embedder’s identity — set it once here, not per call.

  • api_key – Cohere API key. Falls back to CO_API_KEY / COHERE_API_KEY.

  • batch_size – Texts per request (capped at Cohere’s 96 limit).

  • timeout – Per-request timeout in seconds.

Returns:

An Embedder over the Cohere API.

ef.config_id(spec: PipelineSpec) → str[source]

Content-addressed id of a PipelineSpec — sha256 over its steps.

The id is sha256 of the canonical (sorted-key) JSON of PipelineSpec.as_dict(). Two pipelines that describe the same computation hash identically; any change to a step’s op, version or params yields a different id. ef writes this digest into vd index metadata as config_hash.

>>> seg = TransformSpec(op='segment:r', op_version='1', params={})
>>> emb = TransformSpec(op='embed:m@8', op_version='1', params={})
>>> cid = config_id(PipelineSpec(segment=seg, embed=emb))
>>> len(cid)
64
>>> cid == config_id(PipelineSpec(segment=seg, embed=emb))
True
ef.content_hash(source: Any, *, content_keys: Iterable[str] | None = None) → str[source]

Content hash of a source document — sha256 of its normalized content.

This is the basis of change detection: re-reading an unchanged source yields an identical hash, so a refresh that finds the same hash does nothing. Hashing normalized content (NFC, no BOM, \n line endings — see ef.hashing) means cosmetic encoding differences never look like a change.

The treatment of a Mapping source is deliberate:

  • a plain str source — the hash covers the text;

  • a Mapping with a text key — the hash covers source['text'] only. A change to title, tags, ACL or any other metadata does not change the hash, so it never triggers a spurious re-segment/re-embed (embedding is the pipeline’s most expensive step). Name the metadata keys that do feed downstream in content_keys and changes to those keys will be detected;

  • a Mapping without a text key — treated as a structured document and hashed via canonical (sorted-key) JSON of the whole mapping;

  • bytes — hashed raw (no text normalization is possible pre-parse).

Parameters:

  • source – a str, bytes, or a Mapping (ideally with a text key).

  • content_keys – metadata keys whose values join the text in the hash. Each key is looked up at the top level of source or inside a source['metadata'] sub-mapping. Use this only for metadata that genuinely affects the L1–L4 pipeline (e.g. a title prepended to the text before embedding); key order does not matter.

Raises:

TypeError – if source is not a str, bytes or Mapping.

>>> content_hash('hello') == content_hash('hello')
True
>>> content_hash('hello') == content_hash('hello world')
False
>>> content_hash('a\r\nb') == content_hash('a\nb')   # line endings normalized
True
>>> a = {'text': 'body', 'metadata': {'tags': ['x']}}
>>> b = {'text': 'body', 'metadata': {'tags': ['y']}}
>>> content_hash(a) == content_hash(b)                # metadata ignored by default
True
>>> content_hash(a, content_keys=['tags']) == content_hash(b, content_keys=['tags'])
False
>>> content_hash({'text': 'body'}) == content_hash('body')  # wrapper is transparent
True
ef.context_precision(reference: str, retrieved_contexts: Sequence[str]) → float[source]

Fraction of the retrieved contexts that lexically overlap the reference.

A deterministic proxy for retrieval precision — a context counts as useful if it shares at least one (normalized) token with the reference answer.

>>> context_precision('paris', ['Paris is in France.', 'Unrelated text.'])
0.5
ef.context_recall(reference: str, retrieved_contexts: Sequence[str]) → float[source]

Lexical recall of the reference tokens by the retrieved contexts.

The fraction of the (normalized) reference’s distinct tokens that appear anywhere in retrieved_contexts — a deterministic proxy for “did retrieval surface the information the answer needs”. For the LLM-judged context recall, use as_ragas_dataset() and Ragas.

>>> context_recall('paris france', ['Paris is the capital of France.'])
1.0
>>> context_recall('paris france', ['London is in England.'])
0.0
ef.cross_encoder_reranker(model_name: str = 'cross-encoder/ms-marco-MiniLM-L-6-v2', *, device: str | None = None, batch_size: int = 32, **predict_kwargs: Any) → Reranker[source]

Build a Reranker backed by a sentence-transformers CrossEncoder.

A cross-encoder scores a (query, passage) pair jointly — it is the accurate, slower second stage two-stage retrieval is built for. The model is a heavy, optional dependency: it is imported only when this factory is called, so importing ef.reranking itself stays cheap.

Example:

from ef import ingest, with_reranker, cross_encoder_reranker

index = ingest(corpus, embedder="st:all-MiniLM-L6-v2")
retrieve = with_reranker(index.retrieve, cross_encoder_reranker())
segments = retrieve("a query", limit=5)
Parameters:

  • model_name – a sentence-transformers CrossEncoder model name.

  • device – torch device ("cuda" / "cpu" / "mps"); None auto-selects.

  • batch_size – how many pairs to score per forward pass.

  • predict_kwargs – extra keyword arguments forwarded to CrossEncoder.predict.

Returns:

a Reranker — (query, segments) -> scores.

Raises:

ImportError – if the optional sentence-transformers package is absent.

ef.dcg_at_k(ranked: Sequence[str], relevant: Mapping[str, float], k: int) → float[source]

Discounted cumulative gain of ranked at cutoff k.

sum(grade_i / log2(i + 2)) over the top k results (i 0-indexed), with grade_i the relevance of the i-th doc (0 if unjudged) — the linear-gain DCG that BEIR / pytrec_eval use.

>>> round(dcg_at_k(['d1', 'x', 'd2'], {'d1': 1.0, 'd2': 1.0}, 3), 4)
1.5
ef.diagnose(corpus: MutableMapping[str, str | Mapping[str, Any]] | Mapping[str, Any], collection: Any, config: str) → StalenessReport[source]

Compute the four staleness conditions of collection against corpus.

The pure core of ef’s diagnostics — no side effects. It reads what the vd collection records (indexed_state()), what the corpus currently holds (corpus_state()), and partitions every source id into exactly the buckets it belongs in:

  • orphan — in the collection, not in the corpus;

  • missing — in the corpus, not in the collection;

  • stale — in both, but the recorded source_hash ≠ the current one;

  • misconfigured — in both, but a recorded config_hash ≠ config;

  • fresh — in both, content current and config matching.

Parameters:

  • corpus – the source corpus.

  • collection – the vd collection backing the config being diagnosed.

  • config – the ConfigId the collection should hold — documents recording any other config_hash are misconfigured.

Returns:

a StalenessReport.

ef.embed_length_sorted(texts: Sequence[str], encode: Callable[[Sequence[str]], Any], *, batch_size: int = 128) → ndarray[source]

Encode texts in length-sorted batches; return vectors in caller order.

Transformer self-attention is O(n²) in sequence length and a batch is padded to its longest member, so grouping similar-length texts together avoids wasted compute — a 5–10× speedup on skewed corpora. ef owns this so callers never sort, pad, or un-sort themselves.

Parameters:

  • texts – The strings to embed.

  • encode – Maps a list of strings to an array-like of shape (k, dim).

  • batch_size – Texts per call to encode.

Returns:

A float32 array of shape (len(texts), dim) in the original order of texts.

>>> import numpy as np
>>> def enc(batch):  # toy encoder: vector = [length]
...     return np.array([[len(t)] for t in batch], dtype=np.float32)
>>> out = embed_length_sorted(['aaa', 'a', 'aaaaa', 'aa'], enc, batch_size=2)
>>> out.ravel().tolist()  # order preserved despite internal sorting
[3.0, 1.0, 5.0, 2.0]
ef.evaluate_rag(samples: Iterable[RagSample | Mapping[str, Any]], *, metrics: Sequence[str] = ('exact_match', 'token_f1', 'context_recall', 'context_precision', 'non_empty_rate')) → RagEvalReport[source]

Score RAG samples with deterministic, reference-based lexical metrics.

Each sample is a RagSample — (user_input, response, retrieved_contexts, reference). evaluate_rag computes only metrics that need no LLM: exact_match(), token_f1(), context_recall(), context_precision() and non_empty_rate (the share of samples that retrieved any context). This keeps ef a facade — it never synthesizes an answer and never calls a model (§6).

For the LLM-judged metrics — faithfulness, answer relevancy, … — pass the same samples to as_ragas_dataset() and run Ragas with your own LLM.

Parameters:

  • samples – an iterable of RagSamples (or plain mappings of the same shape).

  • metrics – which metrics to compute — any of exact_match / token_f1 / context_recall / context_precision / non_empty_rate. The reference-based ones skip samples that carry no reference.

Returns:

a RagEvalReport.

Raises:

ValueError – for an unknown metric name, or if samples is empty.

>>> samples = [
...     {'user_input': 'capital of France?', 'response': 'Paris',
...      'reference': 'Paris',
...      'retrieved_contexts': ['Paris is the capital of France.']},
...     {'user_input': '2 + 2?', 'response': 'five', 'reference': 'four',
...      'retrieved_contexts': []},
... ]
>>> report = evaluate_rag(samples)
>>> report.metrics['exact_match']
0.5
>>> report.metrics['non_empty_rate']
0.5
ef.evaluate_retrieval(retriever: Any, qrels: Mapping[str, Mapping[str, float]], queries: Mapping[str, str], *, k_values: Sequence[int] = (1, 5, 10, 100), metrics: Sequence[str] = ('ndcg', 'recall', 'precision', 'mrr', 'map'), limit: int | None = None) → RetrievalEvalReport[source]

Score a retriever against a BEIR-shaped evaluation set.

For every query judged in qrels, the retriever is run, its results are mapped back to source document ids (_result_doc_id(), de-duplicated to the best rank per source), and each requested metric is computed at each cutoff in k_values. The report holds the mean of each "<metric>@<k>" over all scored queries — primary metric NDCG@10.

The natural pairing with ingest():

corpus, queries, qrels = read_beir("path/to/beir/dataset")
index = ingest(corpus, embedder="st:all-MiniLM-L6-v2")
report = evaluate_retrieval(index, qrels, queries)
report.primary            # mean NDCG@10
Parameters:

  • retriever – what answers a query — a SearchableCorpus, a SourceManager, or any callable query -> ranked results. Results may be SearchHits, Segments or bare doc-id strings.

  • qrels – the relevance judgements — {query_id: {doc_id: relevance}}. Queries with no positive judgement are skipped.

  • queries – the query texts — {query_id: query_text}. A judged query absent here cannot be run and is skipped.

  • k_values – the rank cutoffs to evaluate at.

  • metrics – which metrics to compute — any of ndcg / recall / precision / mrr / map.

  • limit – how many results to retrieve per query. None uses max(k_values); raise it when sources segment into many pieces (de-duplication to source level can otherwise shorten the ranking).

Returns:

a RetrievalEvalReport.

Raises:

ValueError – for an unknown metric name, or if no query can be scored.

>>> qrels = {'q1': {'d1': 1.0}, 'q2': {'d2': 1.0}}
>>> queries = {'q1': 'alpha', 'q2': 'beta'}
>>> def retriever(query, *, limit=10):
...     return {'alpha': ['d1', 'd2'], 'beta': ['d1', 'd2']}[query][:limit]
>>> report = evaluate_retrieval(retriever, qrels, queries, k_values=(1, 2))
>>> report.metrics['recall@2'], report.n_queries
(1.0, 2)
ef.exact_match(response: str, reference: str) → float[source]

1.0 if response equals reference after normalization, else 0.0.

>>> exact_match('The Eiffel Tower', 'eiffel tower')
1.0
>>> exact_match('Paris', 'London')
0.0
ef.explore(data: Any, *, dims: int = 2, projection_method: Literal['auto', 'umap', 'pca'] = 'auto', cluster_method: Literal['kmeans', 'hdbscan'] = 'kmeans', n_clusters: int = 8, min_cluster_size: int = 5, label: bool = False, context: str = ' ', n_words: int = 4, embedder: Any = None, random_state: int = 42) → ExploreResult[source]

Project, cluster and (optionally) label a corpus in one call.

The layer-L5 orchestrator — project() + cluster() (+ optionally label_clusters()) wired into one structured result. Where project() / cluster() return bare arrays in input order, explore keeps every row tied to its id: it returns an ExploreResult with ids / coords / labels row-aligned (plus cluster_titles) — the JSON-friendly shape an app_ef corpus map (or ef.service.EfService.explore_corpus()) consumes directly.

Parameters:

  • data – the corpus to explore — anything _resolve_explorable() understands (a SourceManager / SearchableCorpus, a Corpus mapping, an iterable of texts or Segments, or a vector matrix).

  • dims – projection target dimensionality — 2 or 3.

  • projection_method – forwarded to project() — 'auto' / 'umap' / 'pca'.

  • cluster_method – forwarded to cluster() — 'kmeans' / 'hdbscan'.

  • n_clusters – number of k-means clusters.

  • min_cluster_size – smallest cluster HDBSCAN will form.

  • label – when True, name each cluster with label_clusters() (needs the ef[imbed] extra, an LLM key, and text-bearing data); when False (the default) cluster_titles is empty.

  • context – the corpus topic passed to label_clusters().

  • n_words – maximum cluster-title length, in words.

  • embedder – embedder used when data is raw text.

  • random_state – seed — projection and clustering are reproducible.

Returns:

an ExploreResult — ids, coords and labels row-aligned, and cluster_titles (empty unless label=True).

Raises:

ValueError – if the corpus is empty, has fewer than 2 samples, or label=True is asked of a text-less vector matrix.

>>> import numpy as np
>>> vectors = np.random.RandomState(0).rand(12, 16)
>>> result = explore(vectors, projection_method='pca', n_clusters=3, random_state=0)
>>> len(result['ids']), len(result['coords']), len(result['labels'])
(12, 12, 12)
>>> result['ids'][:3]
['0', '1', '2']
>>> len(result['coords'][0])                 # each row has `dims` coordinates
2
>>> set(result['labels']) <= {0, 1, 2}
True
>>> result['cluster_titles']                 # empty unless label=True
{}
ef.full_kwargs(op: Callable[[...], Any], args: tuple[Any, ...] = (), kwargs: Mapping[str, Any] | None = None) → dict[str, Any][source]

Normalize a call into a canonical, fully-named, defaults-filled kwargs dict.

The i2.Sig recipe of design notes §3.8. A call expressed as positional args + keyword kwargs is mapped onto op’s signature: every parameter is named, and any parameter the call left implicit is filled with its default. The result is therefore stable — op(x) and op(x, size=512) (512 the default) yield the same dict, which is what makes a content hash built from it correct.

Parameters:

  • op – the callable whose signature defines the parameter names/defaults.

  • args – the call’s positional arguments — for an ef.ops op these are the upstream artifact ids; step_params() drops them again.

  • kwargs – the call’s keyword arguments.

>>> def embed(segments, *, input_type='document', dim=None): ...
>>> full_kwargs(embed, ('seg-1',), {'input_type': 'query'})
{'segments': 'seg-1', 'input_type': 'query', 'dim': None}
>>> full_kwargs(embed, ('seg-1',))                 # defaults filled in
{'segments': 'seg-1', 'input_type': 'document', 'dim': None}
ef.gemini_embedder(model: str = 'gemini-embedding-001', *, dim: int | None = None, api_key: str | None = None, batch_size: int = 100, timeout: float = 60.0) → Embedder[source]

Build a Google Gemini Embedder.

Talks to the Generative Language batchEmbedContents REST endpoint directly — no google-genai SDK, no extra dependency.

Parameters:

  • model – A Gemini embedding model name (a leading models/ is tolerated and stripped).

  • dim – Matryoshka-truncated width (gemini-embedding-001 only). Omit for the model’s native width. Note that Gemini, unlike OpenAI, does not re-normalize MRL-truncated output — a truncated embedder therefore reports normalized=False.

  • api_key – Gemini API key. Falls back to GEMINI_API_KEY / GOOGLE_API_KEY.

  • batch_size – Texts per request (capped at Gemini’s 100 limit).

  • timeout – Per-request timeout in seconds.

Returns:

An Embedder over the Gemini API.

ef.hierarchical(segmenters: Sequence[Any], /) → Segmenter[source]

Compose segmenters into levels — each segments the previous level’s output.

Level 0 applies segmenters[0] to the document. Level k applies segmenters[k] to every level-(k-1) segment’s text, setting each child’s parent_id to that segment’s id and shifting its offsets to stay absolute. The result is a flat stream of every segment of every level (level 0 first), with the tree expressed purely through parent_id pointers.

Each entry of segmenters is passed through as_segmenter(), so strings (e.g. "recursive") and bare callables are accepted alongside ready-made segmenters.

ef.hits_to_segments(hits: Iterable[SearchHit]) → list[Segment][source]

Project SearchHits to plain Segments.

The shape SearchableCorpus.retrieve() / SourceManager.retrieve() hand to an external RAG/agent framework — the RAG-plug-in surface (design notes §F5). A SearchHit carries an ef-specific score and source_id; a Segment is a plain TypedDict an external framework already understands (and trivially adapts to a LangChain Document / LlamaIndex TextNode / a Ragas retrieved_contexts list[str] via [s["text"] for s in segments]).

The similarity score is dropped — Segment is the interchange type and stays pure of a result-only key (call search() when the score matters). Provenance is not lost: each hit’s source_id is folded into the segment’s metadata under the conventional "source" key (one of PROMOTED_METADATA_KEYS), so a plain segment still records which source document it was retrieved from. A segment that already carries metadata["source"] keeps its own value.

>>> hit = SearchHit(segment={'text': 'hi', 'id': 'x'}, score=0.9, source_id='doc-1')
>>> segs = hits_to_segments([hit])
>>> segs[0]['text'], segs[0]['metadata']['source']
('hi', 'doc-1')
>>> 'score' in segs[0]                          # the score is not a Segment key
False
ef.http_embedder(url: str, *, model_id: str | None = None, dim: int | None = None, normalized: bool = False, batch_size: int = 64, headers: dict[str, str] | None = None, timeout: float = 30.0, payload_builder: ~typing.Callable[[~typing.Sequence[str]], bytes] = <function _tei_payload>, response_parser: ~typing.Callable[[~typing.Any], ~typing.Any] = <function _tei_parse>) → Embedder[source]

Build an Embedder over a remote HTTP service.

Defaults target the Text-Embeddings-Inference / infinity shape (a POST of {"inputs": [...]} returning a JSON list of vectors). For a different service, override payload_builder and/or response_parser.

Parameters:

  • url – The embedding endpoint.

  • model_id – Vector-identity string (defaults to "http:<url>").

  • dim – Output width; inferred from the first response if omitted.

  • normalized – Whether the service returns unit vectors.

  • batch_size – Texts per request (used with length-sorted batching).

  • headers – Extra HTTP headers (e.g. an auth token).

  • timeout – Per-request timeout in seconds.

  • payload_builder – texts -> request body bytes.

  • response_parser – decoded JSON -> array-like (n, dim).

Uses only the stdlib — no extra dependency.

ef.imbed_segmenter(name: str = 'default', /, **kwargs: Any) → Segmenter[source]

Build a Segmenter from imbed’s segmenter registry.

imbed’s registered segmenters are bare text -> Iterable[str] callables; this wraps the named one in a FunctionSegmenter so it emits well-formed Segment pieces.

Parameters:

  • name – A key in imbed.components.segmentation.segmenters ("default" resolves to imbed’s configured default segmenter).

  • **kwargs – Forwarded to FunctionSegmenter.

Raises:

  • ImportError – if imbed is not installed.

  • ValueError – if name is not a registered segmenter.

Requires the imbed package (pip install 'ef[imbed]').

ef.ingest(sources: Any, *, segmenter: Any = None, embedder: Any = None, store: Any = None, cache: MutableMapping[str, Any] | None = None) → SearchableCorpus[source]

Index sources and return a ready-to-search SearchableCorpus — one call.

The light path: corpus in, search-ready object out, every component defaulted. ingest(['a', 'b', 'c']).search('query') is the whole story. It builds a single-config SourceManager, materialize()s it, and hands back the thin SearchableCorpus.

Parameters:

  • sources – the corpus — coerced by as_corpus() (a mapping, a directory path, an iterable of strings/mappings, or None).

  • segmenter – the segmenter — as_segmenter() coerces it (None → the recursive default).

  • embedder – the embedder — as_embedder() coerces it; None → DEFAULT_EMBEDDER, the dependency-free HashingEmbedder (so this call needs nothing beyond pip install ef). Pass an explicit embedder — e.g. "st:all-MiniLM-L6-v2" — for neural semantic quality.

  • store – the vector store — None → an in-memory vd backend. See SourceManager for the other accepted forms.

  • cache – the artifact-graph value cache — any MutableMapping.

Returns:

a SearchableCorpus over the freshly indexed corpus.

See the module docstring for a complete offline example.

ef.label_clusters(segments: Iterable[Any], labels: Iterable[Any], *, context: str = ' ', n_words: int = 4, n_samples: int | None = None, **labeler_kwargs: Any) → dict[int, str][source]

Name each cluster with a short LLM-generated title (use cases §G3).

A thin wrapper over imbed’s imbed.tools.ClusterLabeler: it samples each cluster’s segments, asks an LLM for a few-word title, and returns {cluster_id: title}. ef does not reimplement this — it reuses imbed (design notes §7).

Parameters:

  • segments – the segment texts — plain strings or Segment mappings (the "text" field is used). Aligned with labels.

  • labels – the cluster id of each segment — typically the output of cluster().

  • context – the corpus’s overall topic, given to the LLM so titles describe how a cluster differs from the rest, not the shared topic.

  • n_words – maximum title length, in words.

  • n_samples – how many segments to sample per cluster when prompting; None keeps ClusterLabeler’s default.

  • **labeler_kwargs – forwarded to imbed.tools.ClusterLabeler (e.g. max_unique_clusters, prompt).

Returns:

{cluster_id: title} for every distinct label.

Raises:

  • ValueError – if segments and labels differ in length.

  • ImportError – if imbed (and its pandas/oa deps) is missing.

>>> # needs ef[imbed] + an LLM key — illustrative only:
>>> # label_clusters(['neural net training ...', ...], [0, 1, 0, ...])
>>> # -> {0: 'Neural network training', 1: 'Dataset preprocessing'}
ef.line_segmenter(doc: str | Mapping[str, Any], /) → Iterator[Segment][source]

Segment a document into its non-empty lines.

Each line is stripped of surrounding whitespace; blank lines are dropped. Character offsets point at the stripped content within the source.

A plain function — and therefore already a Segmenter. It carries the marker as_segmenter() uses to recognize a ready-made segmenter.

>>> [s['text'] for s in line_segmenter('  hello  \n\nworld\n')]
['hello', 'world']
ef.make_segment(text: str, *, id: str | None = None, parent_id: str | None = None, start: int | None = None, end: int | None = None, index: int | None = None, tokens: int | None = None, metadata: Mapping[str, Any] | None = None) → Segment[source]

Build a Segment, deriving a content-based id if none given.

Optional fields are omitted from the result when None — a segment dict carries only the keys it actually has.

>>> seg = make_segment('chunk one', index=0, start=0, end=9)
>>> sorted(seg)
['end', 'id', 'index', 'start', 'text']
>>> seg['id'] == segment_id('chunk one')
True
ef.materialise(x: Any) → Any[source]

Force a streaming segmenter (or segment stream) to a concrete list.

Streaming is the default, but the light case often wants everything in hand. materialise() is polymorphic:

  • given a segmenter (a callable), it returns a new segmenter whose output is a list rather than a lazy iterator;

  • given an iterable of segments, it returns a list of them.

>>> eager = materialise(line_segmenter)
>>> out = eager('a\nb')
>>> isinstance(out, list)
True
>>> [s['text'] for s in out]
['a', 'b']
>>> materialise(s for s in [{'text': 'x'}])
[{'text': 'x'}]
ef.ndcg_at_k(ranked: Sequence[str], relevant: Mapping[str, float], k: int) → float[source]

Normalized DCG at cutoff k — ef’s primary retrieval metric.

dcg_at_k() divided by the ideal DCG (the DCG of the best possible ranking of the judged docs). 1.0 is a perfect ranking; 0.0 when no relevant doc was judged.

>>> ndcg_at_k(['d1', 'd2'], {'d1': 1.0, 'd2': 1.0}, 2)   # perfect order
1.0
>>> round(ndcg_at_k(['x', 'd1'], {'d1': 1.0}, 2), 4)     # relevant doc at rank 2
0.6309
ef.openai_embedder(model: str = 'text-embedding-3-small', *, dim: int | None = None, api_key: str | None = None, client: Any | None = None, batch_size: int = 2048, poll_interval: float = 30.0) → Embedder[source]

Build an OpenAI-embeddings Embedder.

Parameters:

  • model – An OpenAI embedding model name.

  • dim – Matryoshka-truncated output width (text-embedding-3-* only). Omit for the model’s native dimension. The dim is part of the embedder’s identity — set it once here, not per call.

  • api_key – OpenAI API key (else the SDK’s usual env-var resolution).

  • client – A pre-built openai.OpenAI client (overrides api_key).

  • batch_size – Texts per request (capped at OpenAI’s 2048 limit).

  • poll_interval – Seconds between polls when blocking on a Batch-API job.

Returns:

An embedder whose embed_batch uses the async OpenAI Batch API.

Requires the openai package (pip install 'ef[openai]').

ef.plan_refresh(report: StalenessReport, *, mode: Literal['none', 'incremental', 'full', 'scoped_full'] = 'full', scope: Iterable[str] | None = None) → RefreshPlan[source]

Turn a StalenessReport into a RefreshPlan.

The pure, side-effect-free core of the refresh policy. Every source that is missing, stale or misconfigured is (re-)materialized; what gets deleted is what the mode dictates:

  • none — delete nothing (stale content is added alongside the old);

  • incremental — delete the stale/misconfigured sources’ old documents;

  • full / scoped_full — also delete orphans.

Parameters:

  • report – the staleness report to act on.

  • mode – one of REFRESH_MODES.

  • scope – an optional subset of source ids the plan is restricted to — sources outside it are left entirely alone. None means the whole report. (With scope=None, full and scoped_full coincide.)

Raises:

ValueError – if mode is not a recognized RefreshMode.

>>> from ef.diagnostics import StalenessReport
>>> report = StalenessReport(config='c0', orphan=('x',), missing=('a',),
...                          stale=('b',), misconfigured=('c',), fresh=('d',))
>>> plan_refresh(report, mode='full').to_materialize
('a', 'b', 'c')
>>> plan_refresh(report, mode='full').to_delete
('b', 'c', 'x')
>>> plan_refresh(report, mode='incremental').to_delete
('b', 'c')
>>> plan_refresh(report, mode='none').to_delete
()
>>> plan_refresh(report, mode='scoped_full', scope=['a', 'b']).to_delete
('b',)
ef.precision_at_k(ranked: Sequence[str], relevant: Mapping[str, float], k: int) → float[source]

Fraction of the top k results that are relevant.

>>> precision_at_k(['d1', 'x', 'd2'], {'d1': 1.0, 'd2': 1.0}, 2)
0.5
ef.producer_spec(op: str, *inputs: str, op_version: str, **params: Any) → ProducerSpec[source]

Build a ProducerSpec ergonomically — inputs positional, params keyword.

The natural way to write a recipe: the input artifact ids are positional (their order is significant — it is the order the op receives them), and every keyword argument is an op parameter. op_version is keyword-only and required — an op without a stated version cannot be content-addressed safely.

>>> spec = producer_spec('segment', 'doc-1', op_version='1', size=512, overlap=64)
>>> spec.op, spec.inputs
('segment', ('doc-1',))
>>> spec.params == {'size': 512, 'overlap': 64}
True
ef.project(data: Any, *, dims: int = 2, method: Literal['auto', 'umap', 'pca'] = 'auto', embedder: Any = None, n_neighbors: int = 15, min_dist: float = 0.1, metric: str = 'cosine', pca_components: int = 50, random_state: int = 42) → ndarray[source]

Project a corpus’s embeddings to dims coordinates for visualization.

The canonical “show me the shape of the corpus” operation (use cases §G1). The default method='auto' runs the design-notes §8.4 recipe — PCA → UMAP, cosine metric, seeded: PCA first compresses high-dimensional embeddings to pca_components (cheap denoising), then UMAP lays them out in dims dimensions.

Parameters:

  • data – what to project — anything _resolve_vectors() understands: an ndarray (or nested sequence) of vectors, an iterable of texts (embedded with embedder), a Corpus mapping, or a SearchableCorpus / SourceManager (vectors pulled from the index).

  • dims – target dimensionality — 2 or 3 for plotting.

  • method – 'auto' (UMAP when available and the corpus is large enough, else PCA), 'umap' (force UMAP — needs the ef[explore] extra), or 'pca' (numpy-only, always available).

  • embedder – embedder used when data is raw text; defaults to the dependency-free DEFAULT_EMBEDDER.

  • n_neighbors – UMAP neighborhood size (clamped to n_samples - 1).

  • min_dist – UMAP minimum point separation in the embedding.

  • metric – UMAP distance metric — 'cosine' for semantic embeddings.

  • pca_components – PCA width for the pre-UMAP compression step; PCA is skipped when the source dimensionality is already smaller.

  • random_state – seed — projection is reproducible.

Returns:

An ndarray of shape (n_samples, dims), rows in input order.

Raises:

  • ValueError – if there are fewer than 2 samples.

  • ImportError – if method='umap' and umap-learn is not installed.

>>> import numpy as np
>>> vectors = np.random.RandomState(1).rand(20, 32)
>>> project(vectors, dims=3, method='pca').shape
(20, 3)
ef.read_beir(directory: str, *, qrels: str | None = None) → tuple[dict[str, str], dict[str, str], dict[str, dict[str, float]]][source]

Load a BEIR-format dataset from disk into the in-memory evaluation triple.

Reads corpus.jsonl (records {"_id", "title", "text"}), queries.jsonl (records {"_id", "text"}) and a qrels TSV (query-id, corpus-id, score columns). The corpus text is the record’s title and text joined — the BEIR convention.

Parameters:

  • directory – the dataset directory.

  • qrels – the qrels file, relative to directory. None tries _DEFAULT_QRELS_PATHS in order.

Returns:

a (corpus, queries, qrels) triple — corpus and queries are {id: text} dicts, qrels is {query_id: {doc_id: relevance}}. Feed it straight to evaluate_retrieval() (after indexing corpus with ingest()).

Raises:

FileNotFoundError – if corpus.jsonl, queries.jsonl or the qrels file is missing.

ef.ready_handle(array: Any) → BatchHandle[source]

Wrap an already-computed array in a finished BatchHandle.

>>> import numpy as np
>>> h = ready_handle(np.zeros((2, 3)))
>>> h.poll()
'done'
>>> h.result().shape
(2, 3)
ef.recall_at_k(ranked: Sequence[str], relevant: Mapping[str, float], k: int) → float[source]

Fraction of the relevant docs that appear in the top k results.

>>> recall_at_k(['d1', 'x', 'd2'], {'d1': 1.0, 'd2': 1.0}, 2)
0.5
>>> recall_at_k(['d1', 'x', 'd2'], {'d1': 1.0, 'd2': 1.0}, 3)
1.0
ef.reciprocal_rank(ranked: Sequence[str], relevant: Mapping[str, float], k: int | None = None) → float[source]

Reciprocal of the rank of the first relevant result (0 if none).

Averaged over queries this is MRR. k optionally caps how far down the ranking to look.

>>> reciprocal_rank(['x', 'd1', 'd2'], {'d1': 1.0})
0.5
>>> reciprocal_rank(['x', 'd1'], {'d1': 1.0}, k=1)       # relevant doc past the cutoff
0.0
ef.refresh_on_change(manager: Any) → Callable[['ChangeEvent'], None][source]

Build a ChangeEvent handler that auto-refreshes manager.

The auto refresh seam. The returned callable is meant to be a ChangeDetectingCorpus’s on_change — each detected edit (through the wrapper, or surfaced out-of-band by scan()) is incrementally applied to every materialized config of manager: an added source is indexed, a modified one is re-indexed (delete-then-add), a deleted one is removed.

SourceManager wires this automatically when constructed with auto_refresh=True; call this directly only to attach a manager to a corpus you build yourself.

Parameters:

manager – a SourceManager (duck-typed — any object with an _apply_change(event) method).

Returns:

a Callable[[ChangeEvent], None] for on_change.

ef.rerank(query: str, segments: Sequence[Segment], reranker: Reranker, *, limit: int | None = None) → list[Segment][source]

Re-score segments against query with reranker and re-order them.

The pure heart of the module — no I/O, no retrieval. The reranker is called once with the whole candidate list; the segments are returned sorted by descending score (ties keep their incoming order), optionally trimmed to limit. Each returned segment is a copy carrying the reranker’s score in metadata["rerank_score"] — so the new ranking is auditable.

Parameters:

  • query – the search query the segments are scored against.

  • segments – the candidate segments (from a first-stage retriever).

  • reranker – a Reranker — a callable (query, segments) -> scores.

  • limit – keep only the top limit after reranking; None keeps all.

Returns:

the reranked segments — copies, best first.

Raises:

ValueError – if the reranker returns a number of scores that does not match the number of segments.

>>> segs = [{'text': 'a', 'id': '1'}, {'text': 'bbb', 'id': '2'}]
>>> ranked = rerank('q', segs, lambda q, s: [len(x['text']) for x in s])
>>> [r['text'] for r in ranked]
['bbb', 'a']
>>> ranked[0]['metadata']['rerank_score']
3.0
ef.segment_id(text: str, metadata: Mapping[str, Any] | None = None) → str[source]

Content-derived id for a segment — sha256 of its normalized content.

The id is sha256(normalize(text) + canonical_json(metadata)) — the normalization and hashing primitives live in ef.hashing, the single source of truth ef content-addresses through. Because the id depends only on content, re-segmenting an unchanged document produces identical ids — which is what makes ingestion idempotent (a re-run is a no-op, not a duplicate). Pass metadata only when two segments could share identical text yet must stay distinct (e.g. same boilerplate from different sources).

>>> segment_id('hello') == segment_id('hello')
True
>>> segment_id('hello') == segment_id('world')
False
>>> segment_id('x', {'source': 'a'}) == segment_id('x', {'source': 'b'})
False
ef.segment_record(x: Any) → SegmentRecord[source]

Coerce x (str / Mapping / SegmentRecord) into a record.

The dataclass counterpart of as_segment() — the convenience-surface constructor.

>>> segment_record('hello').text
'hello'
>>> segment_record({'text': 'hi', 'tokens': 1}).tokens
1
ef.sentence_transformers_embedder(model_name: str = 'all-MiniLM-L6-v2', *, device: str | None = None, normalize: bool = False, batch_size: int = 128, model_path: str | None = None, **st_kwargs: Any) → Embedder[source]

Build a local sentence-transformers Embedder.

Parameters:

  • model_name – A sentence-transformers model name (also used in model_id).

  • device – Torch device ("cuda" / "cpu" / "mps"); None auto-selects.

  • normalize – Whether to L2-normalize outputs. sentence-transformers does not normalize by default — set this truthfully.

  • batch_size – Encoding batch size (smaller on CPU).

  • model_path – A local filesystem path to load from instead of the hub — for air-gapped use. Loading still records model_name in model_id.

  • **st_kwargs – Forwarded to the SentenceTransformer constructor.

Requires sentence-transformers (pip install 'ef[sentence-transformers]').

ef.step_params(op: Callable[[...], Any], raw_params: Mapping[str, Any] | None = None, *, n_inputs: int = 1) → dict[str, Any][source]

The canonical keyword params of an op step — its inputs dropped.

An ef.ops op is called op(*input_values, **params): the first n_inputs positional parameters are upstream artifacts (they belong in ProducerSpec.inputs, not in the params), and every remaining parameter is a hashable step param. This runs full_kwargs() with placeholder inputs, then drops those input slots — leaving exactly the params, defaults filled, ready for a TransformSpec.

Parameters:

  • op – the op callable whose signature is the param contract.

  • raw_params – the keyword params the caller actually supplied; anything omitted is defaulted.

  • n_inputs – how many leading positional parameters are upstream inputs.

>>> def embed(segments, *, input_type='document'): ...
>>> step_params(embed, {'input_type': 'query'})
{'input_type': 'query'}
>>> step_params(embed)                              # default filled in
{'input_type': 'document'}
>>> def segment(source): ...
>>> step_params(segment)                            # no params at all
{}
ef.token_f1(response: str, reference: str) → float[source]

Token-overlap F1 of response against reference (the SQuAD F1).

The harmonic mean of token precision and recall over the normalized token bags — a graded credit where exact_match() is all-or-nothing.

>>> token_f1('the quick brown fox', 'a quick brown fox')
1.0
>>> round(token_f1('quick brown fox', 'quick brown dog'), 4)
0.6667
ef.voyage_embedder(model: str = 'voyage-3.5', *, dim: int | None = None, api_key: str | None = None, batch_size: int = 1000, timeout: float = 60.0) → Embedder[source]

Build a Voyage AI Embedder.

Talks to Voyage’s /v1/embeddings REST endpoint directly — no voyageai SDK, no extra dependency. Voyage embeddings are L2-normalized. Voyage distinguishes only query from document; a classification or clustering hint is simply not sent.

Parameters:

  • model – A Voyage embedding model name.

  • dim – Matryoshka-truncated width (the voyage-3.5 family and voyage-code-3 only). Omit for the model’s native width.

  • api_key – Voyage API key. Falls back to VOYAGE_API_KEY.

  • batch_size – Texts per request (capped at Voyage’s 1000 limit).

  • timeout – Per-request timeout in seconds.

Returns:

An Embedder over the Voyage API.

ef.with_overlap(segmenter: ~ef.segmenters.Segmenter, n_chars: int, /, *, count_tokens: ~typing.Callable[[str], int] = <function approx_token_count>, tokenizer: str = 'char-approx') → Segmenter[source]

Wrap segmenter so consecutive segments overlap by n_chars characters.

Each segment is extended forward by up to n_chars characters (clamped to the source length), so segment K shares its tail with segment K+1’s head. A segment whose text changes has its id re-derived and its tokens recounted; segments lacking start / end offsets pass through untouched.

Overlap is expressed in characters here (this is a generic post-hoc helper) — for token-measured overlap use a segmenter’s own chunk_overlap.

>>> base = RecursiveCharacterSegmenter(chunk_size=3, chunk_overlap=0)
>>> [c['text'] for c in base('one two three four')]
['one two three', 'four']
>>> [c['text'] for c in with_overlap(base, 100)('one two three four')]
['one two three four', 'four']
ef.with_reranker(retriever: Callable[[...], Sequence[Segment]], reranker: Reranker, *, fetch_k: int = 50) → Callable[[...], list[Segment]][source]

Decorate a retrieve-style callable with a two-stage reranking pass.

Wraps retriever (e.g. SourceManager.retrieve, or any callable query -> list[Segment]) so every call over-fetches fetch_k candidates cheaply, rerank()s them, and returns the top limit. The decorated callable keeps the retriever’s interface — query plus a keyword-only limit — so it is a drop-in replacement, including as the retriever argument of evaluate_retrieval().

Parameters:

  • retriever – the first-stage retriever — a callable retriever(query, *, limit=...) -> segments.

  • reranker – the Reranker applied to the over-fetched candidates.

  • fetch_k – how many candidates to retrieve before reranking. Larger trades latency for recall headroom; it should comfortably exceed the limit callers ask for.

Returns:

a new callable (query, *, limit=10, **kwargs) -> list[Segment] — extra keyword arguments (e.g. filter) pass straight through to retriever.

>>> pool = [{'text': t, 'id': t} for t in ['ab', 'abcd', 'abc']]
>>> def base(query, *, limit=10):
...     return pool[:limit]
>>> retrieve = with_reranker(
...     base, lambda q, segs: [len(s['text']) for s in segs], fetch_k=10)
>>> [s['text'] for s in retrieve('a query', limit=2)]
['abcd', 'abc']
ef.write_beir(directory: str, corpus: Mapping[str, str], queries: Mapping[str, str], qrels: Mapping[str, Mapping[str, float]]) → None[source]

Write an evaluation triple to disk in BEIR format — the inverse of read_beir().

Creates directory (and parents) and writes corpus.jsonl, queries.jsonl and qrels.tsv. Use it to snapshot an ef corpus as a reusable benchmark (ef_use_cases.md §H4).

Parameters:

  • directory – the output directory — created if absent.

  • corpus – the documents — {doc_id: text}.

  • queries – the queries — {query_id: query_text}.

  • qrels – the relevance judgements — {query_id: {doc_id: relevance}}.