ef.source_manager

The search facade — one-shot ingest(), SearchableCorpus, SourceManager.

This is where ef’s layers come together. Phases 1–4 built the parts — a Corpus, a Segmenter, an Embedder, the content-addressed ArtifactGraph — and this module wires corpus → segment → embed → vd into a usable search facade with the progressive disclosure ef promises:

  • The light path — ingest() — one line. A list of strings (or any corpus) in, a SearchableCorpus out, search(query) ready.

  • The heavy path — SourceManager — the same machinery with the components named explicitly and multiple configs (a config is one segmenter+embedder pipeline). Two configs that share a step share that step’s artifacts in the ArtifactGraph for free — config branching costs only the divergent cone.

  • Keeping it fresh — SourceManager.diagnose() reports how far the index has drifted from the corpus (the four staleness conditions); SourceManager.refresh() re-syncs it. Pass auto_refresh=True to keep

    the index live as the corpus is edited. See ef.diagnostics and

    ef.refresh.

How the wiring works (one source, one config):

  1. the source is a graph leaf, keyed by content_hash();

  2. a segment ProducerSpec over that leaf → a list of Segments;

  3. an embed ProducerSpec over the segment artifact → an (n, dim) array;

  4. each (segment, vector) pair is written into a vd collection as a vd.Document, with source_id / source_hash / config_hash

    recorded in its metadata.

ef owns the embedding: it always supplies vectors to vd (the index is vd’s job, the vectors are ef’s — design notes §7), and at query time it embeds the query with input_type="query" and hands vd the vector.

Example — the light path, end to end, offline:

>>> import numpy as np
>>> from ef import ingest, as_embedder
>>> toy = as_embedder(
...     lambda texts: np.array(
...         [[t.count('a'), t.count('e'), t.count('o')] for t in texts], dtype=float
...     ),
...     model_id='toy-vowels@3',
... )
>>> idx = ingest(['apple orchard', 'ocean breeze', 'green meadow'], embedder=toy)
>>> hits = idx.search('ocean breeze', limit=2)
>>> hits[0].segment['text']
'ocean breeze'
>>> round(hits[0].score, 6)
1.0
>>> hits[0].source_id is not None
True

And the true light path — no embedder at all, nothing to install beyond ef itself (the default is the dependency-free HashingEmbedder):

>>> idx = ingest(['apple orchard', 'ocean breeze', 'green meadow'])
>>> idx.search('ocean breeze', limit=1)[0].segment['text']
'ocean breeze'

Module Attributes

DEFAULT_EMBEDDER

The model name ingest() / SourceManager resolve an absent embedder to — HashingEmbedder, the dependency-free default (numpy only, no extra to install).

Functions

hits_to_segments(hits)

Project SearchHits to plain Segments.

ingest(sources, *[, segmenter, embedder, ...])

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

Classes

SearchHit(segment, score[, source_id])

One ranked search result — a Segment and its score.

SearchableCorpus(collection, embedder, *[, ...])

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

SourceManager([corpus, segmenter, embedder, ...])

The heavy search facade — multi-config corpus indexing over one corpus.
ef.source_manager.DEFAULT_EMBEDDER = 'hashing'

The model name ingest() / SourceManager resolve an absent embedder to — HashingEmbedder, the dependency-free default (numpy only, no extra to install). It is a lexical embedder; pass an explicit embedder ("st:...", "openai:...", …) for neural semantic quality.

class ef.source_manager.SearchHit(segment, score, source_id=None)[source]

Bases: object

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.

score

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

source_id

the corpus key of the source document, if recorded.

>>> 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.source_manager.SearchableCorpus(collection, embedder, *, config=None)[source]

Bases: object

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 (Any) – the vd collection holding this config’s vectors.

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

  • config (str | None) – the ConfigId of the indexed pipeline, if known.

>>> # built for you by ingest(); see the module docstring for a full example.
retrieve(query, *, limit=10, filter=None)[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"].

Return type:

list[Segment]

>>> # idx.retrieve('a query') -> [{'text': ..., 'id': ..., 'metadata': ...}, ...]
search(query, *, limit=10, filter=None)[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.

Return type:

list[SearchHit]

class ef.source_manager.SourceManager(corpus=None, *, segmenter=None, embedder=None, store=None, cache=None, auto_refresh=False, embedder_api_key=None)[source]

Bases: object

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 (Any) – the source corpus — coerced by as_corpus() (a mapping, a directory path, an iterable of sources, or None).

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

  • embedder (Any) – 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 (str | None) – 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 (Any) – 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 (MutableMapping[str, Any] | None) – the ArtifactGraph’s value cache — any MutableMapping; None → an in-RAM dict.

  • auto_refresh (bool) – 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=None)[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.

Return type:

StalenessReport

>>> 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=None)[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').

Return type:

int

lineage(key)[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.

Return type:

frozenset[str]

materialize(config=None, *, sources=None)[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 (str | None) – the config to index — a name, a ConfigId, or None for every registered config.

  • sources (Iterable[str] | None) – 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.

Return type:

dict[str, Any]

Returns:

a report dict with configs / sources / segments counts.

pipeline(config=None)[source]

The PipelineSpec of a registered config.

Return type:

PipelineSpec

rebuild(config=None)[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.

Return type:

RefreshReport

Returns:

a RefreshReport.

refresh(config=None, *, sources=None, mode='full')[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 (str | None) – the config to refresh (name / id / None for the default or sole config).

  • sources (Iterable[str] | None) – 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 (Literal['none', 'incremental', 'full', 'scoped_full']) – 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'.

Return type:

RefreshReport

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='default', *, segmenter=None, embedder=None, embedder_api_key=None)[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).

Return type:

str

>>> 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, *, config=None, limit=10, filter=None)[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.

Return type:

list[Segment]

scan()[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).

Return type:

CorpusDiff

search(query, *, config=None, limit=10, filter=None)[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.

Return type:

list[SearchHit]

searchable(config=None)[source]

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

Return type:

SearchableCorpus

ef.source_manager.hits_to_segments(hits)[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.

Return type:

list[Segment]

>>> 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.source_manager.ingest(sources, *, segmenter=None, embedder=None, store=None, cache=None)[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 (Any) – the corpus — coerced by as_corpus() (a mapping, a directory path, an iterable of strings/mappings, or None).

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

  • embedder (Any) – 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 (Any) – the vector store — None → an in-memory vd backend. See SourceManager for the other accepted forms.

  • cache (MutableMapping[str, Any] | None) – the artifact-graph value cache — any MutableMapping.

Return type:

SearchableCorpus

Returns:

a SearchableCorpus over the freshly indexed corpus.

See the module docstring for a complete offline example.