Retrieval

Auto-generated from source docstrings (mkdocstrings). The complete public contract is agentmaker.__all__ plus each subpackage __all__.

agentmaker.retrieval: shared retrieval foundation for memory and rag (hybrid retrieval + rerank, isolated by scope).

Layering (ports & adapters): the pure framework (interfaces + orchestration + contracts) is kept separate from the swappable batteries (backends/): - types / scope : RetrievalResult unified result + MetadataFilter filter contract; Scope multi-dimensional ownership concept + guardrails (scope_sql holds the Scope->SQL mapping, shared across agentmaker) - base : the abstract ports Embedder / VectorStore / KeywordIndex / Reranker - hybrid : storage-agnostic orchestrator HybridRetriever + default fusion battery RRFFusion / reciprocal_rank_fusion - index_sync : source-of-truth -> derived index sync seam IndexSync / SyncIndexSync (shared by memory and rag) - backends/ : swappable adapters (batteries): openai_embedder / cohere_reranker / sqlite; changing backend or model touches only this layer

RetrievalError

Bases: AgentmakerError

Unified exception for the retrieval subsystem (embedding / vector store / retrieval). Configuration, missing dependencies, and invocation and storage failures are all normalized here.

CohereReranker

Bases: Reranker

Cross-encoder reranking backed by the Cohere Rerank API (multilingual, including Chinese).

__init__(model='rerank-v4.0-fast', api_key=None, *, timeout=30.0)

Parameters:

Name	Type	Description	Default
model	str	Cohere rerank model name, defaults to rerank-v4.0-fast (multilingual / cost-effective); use rerank-v4.0-pro for highest quality.	'rerank-v4.0-fast'
api_key	Optional[str]	API key; if omitted, read from the COHERE_API_KEY environment variable (note the Cohere SDK defaults to looking for CO_API_KEY, so we read it explicitly here).	None
timeout	float	Timeout in seconds.	30.0

rerank(query, results, *, top_k=5)

Hand the candidate texts to Cohere for precise reranking; map the returned index back to the original RetrievalResult and swap in the [0,1] relevance_score.

Fts5KeywordIndex

Bases: KeywordIndex

Keyword index implemented with SQLite FTS5 + jieba tokenization + bm25 ranking.

Before ingestion, jieba splits Chinese into words, joined by spaces and stored in the indexed tokens column; the original text goes into the UNINDEXED content column. This way FTS5's default unicode61 tokenizer only needs to split on spaces to get jieba's real words (unicode61 does not tokenize raw Chinese). Queries likewise jieba-tokenize first, OR-ing between words. Connection / transaction / lock are delegated to SqliteBackend; this class only handles FTS5-specific tokenization and table creation / querying.

__init__(db_path=':memory:', *, table='kw_items', connection=None, lock=None, metadata_columns=())

metadata_columns: declares which metadata fields are filterable (materialized as md_ UNINDEXED columns); defaults to none, sharing the same declaration as the vector store.

add(ids, contents, *, scope=None, metadatas=None)

Batch upsert: delete old rows by (id, scope) first, then insert; ids must be unique within a batch. The tokens column stores jieba tokens, the content column stores the original text.

metadatas (optional, same length as ids): each item's metadata dict; only declared fields are stored (metadata_columns, text-ified), the rest ignored.

delete(ids, *, scope=None)

Batch delete by id (within the scope's range, B semantics). Idempotent. An empty scope deletes across the whole database; the destructive guardrail is at the HybridRetriever layer.

search(query, *, top_k=5, scope=None, filters=None)

jieba-tokenize -> OR between words -> FTS5 + bm25 retrieval (narrowed by scope + optional metadata filters); smaller bm25 = more relevant, negated to satisfy the "larger = more relevant" convention.

close()

Close the database connection (a shared connection is closed by the orchestrator).

OpenAIEmbedder

Bases: Embedder

OpenAI-compatible embedding implementation (defaults to text-embedding-3-small).

Consistent with the LLM clients in agentmaker.core: reuses the openai package, builds the client lazily, and reads the OPENAI_* environment variables by default.

dim property

Vector dimension.

model_id property

Model identifier (goes into the index fingerprint, see base.Embedder.model_id): the model name; when the dimension is shortened explicitly, append the dimension suffix to distinguish it.

__init__(model='text-embedding-3-small', api_key=None, base_url=None, *, dimensions=None, timeout=30.0, max_batch=256)

Resolve key / base URL / dimensions and validate them; no network request here (the client is built on the first embed call).

Parameters:

Name	Type	Description	Default
model	str	Model name, defaults to text-embedding-3-small.	'text-embedding-3-small'
api_key	Optional[str]	API key; if omitted, read from the OPENAI_API_KEY environment variable.	None
base_url	Optional[str]	Service base URL; if omitted, read OPENAI_BASE_URL, then fall back to the official endpoint.	None
dimensions	Optional[int]	Optional, shortens the output dimension (supported by the OpenAI 3 series); if omitted, use the model's default dimension.	None
timeout	float	Timeout in seconds.	30.0
max_batch	int	Maximum number of texts embedded per API request; anything beyond is split into serial batches automatically (default 256; OpenAI has dual per-request limits, roughly 2048 texts + ~300k total tokens, so lower it when chunks are especially large).	256

embed(texts)

Embed a batch of texts; anything over max_batch is split into serial calls, then concatenated in input order, with exceptions normalized to RetrievalError.

OpenAI embeddings have dual per-request limits (roughly 2048 texts + ~300k total tokens); sending several thousand chunks from a large document at once would fail the whole batch with a 400. So we split by max_batch (default 256), validating count / dimension independently per batch.

Parameters:

Name	Type	Description	Default
texts	List[str]	List of texts (an empty list returns empty directly).	required

Returns:

Type	Description
List[List[float]]	List[List[float]]: Vectors strictly aligned to the input order.

SqliteHybridRetriever

Bases: HybridRetriever

Hybrid retriever sharing a single connection: add / delete collect both indexes' writes into a single cross-index transaction for atomic commit.

Stronger than the base HybridRetriever's best-effort compensation: any failed step rolls back the whole thing, with no half-write, and a failed update does not lose the old value either (vec0 and FTS5 share a connection and a transaction, committing / rolling back together). Constructed by build_sqlite_hybrid (both indexes share one connection + one lock).

add(ids, contents, *, scope=None, metadatas=None)

Both indexes' writes + one commit wrapped in the same transaction; any failed step rolls back the whole thing (a failed overwrite of an existing id also does not lose the old value).

replace(old_ids, new_ids, contents, *, scope=None, metadatas=None)

Ingesting new chunks + deleting old chunks (those not in new) collected into the same transaction for atomic commit: retrieval sees either all-old or all-new, eliminating the concurrency window where "old and new coexist and are hit together (duplicate hits)" and any leftover from a failed old-chunk delete. On failure, roll back the whole thing and keep the old version (no data loss).

delete(ids, *, scope=None, all_scopes=False)

Both indexes' deletes wrapped in the same transaction for atomic commit. Empty-scope guardrail same as the base class (see scope.py).

asearch(query, **kwargs) async

asearch wraps the sync search in a single to_thread (two-path gather has no gain under the shared connection lock).

asearch_many(queries, **kwargs) async

asearch_many wraps the sync search_many in a single to_thread.

close()

Close both stores (a no-op under a shared connection), then close the shared connection uniformly.

SqliteVecStore

Bases: VectorStore

Vector store implemented with sqlite-vec's vec0 virtual table; data lives in a single SQLite file (or in memory).

Table structure: embedding vector column + id + one column per scope dimension + declared metadata columns (md_*, filterable in KNN) + content auxiliary column (stored but not filtered). Distance uses vec0's default L2; with already-normalized vectors like OpenAI's, L2 and cosine rank identically. The dimension is fixed at table creation; a model with a different dimension needs a new table / new database file. Connection / transaction / lock are delegated to SqliteBackend; this class only handles vec0-specific table creation and KNN.

__init__(dim, db_path=':memory:', *, table='vec_items', connection=None, lock=None, metadata_columns=())

metadata_columns: declares which metadata fields are filterable (fixed at table-creation time, materialized as md_ columns); defaults to none.

add(ids, vectors, contents, *, scope=None, metadatas=None)

Batch upsert: delete old rows by (id, scope) first, then insert; ids must be unique within a batch, and rewriting the same (id, scope) across batches just overwrites. Vectors are passed as JSON array text.

metadatas (optional, same length as ids): each item's metadata dict; only declared fields are stored (metadata_columns, text-ified), the rest ignored.

delete(ids, *, scope=None)

Batch delete by id (within the scope's range, B semantics). Idempotent. An empty scope deletes across the whole database; the destructive guardrail is at the HybridRetriever layer.

delete_exact(ids, *, scope=None)

Delete by write footprint exactly (all-dimension match, _delete_exact): used by add's compensating undo of just-written vectors, without wrongly deleting same-id sibling rows across scopes.

search(query_vector, *, top_k=5, scope=None, filters=None)

KNN retrieval: vec0's MATCH + k + scope (+ optional metadata filters); smaller distance = nearer, converted into a "larger = more relevant" score.

The same query uses vec_to_json(embedding) to fetch the matched items' vectors back too (no extra query), filling embedding for upstream MMR reuse to avoid recomputation.

close()

Close the database connection (a shared connection is closed by the orchestrator).

Embedder

Bases: ABC

Abstract base class for turning text into vectors. Subclasses implement embed() and dim.

dim abstractmethod property

Vector dimension. Used to declare the column width float[dim] when building the vector store.

model_id property

Model identifier (e.g. "text-embedding-3-small"), used for the index<->embedder fingerprint check. Vectors from different models live in incomparable spaces; swapping to "a different model of the same dimension" without a fingerprint would silently mix vectors and quietly degrade retrieval. Defaults to None (unknown, in which case the check compares dimension only); subclasses should override.

embed(texts) abstractmethod

Turn a batch of texts into a batch of vectors (batched to save round trips).

Parameters:

Name	Type	Description	Default
texts	List[str]	List of texts.	required

Returns:

Type	Description
List[List[float]]	List[List[float]]: Vectors, same length and order as the input.

aembed(texts) async

Async version of embed (defaults to to_thread).

FusionStrategy

Bases: ABC

Abstract base class for the strategy that fuses multi-way retrieval results (the fifth port).

Fuses several result lists (each already sorted by relevance) into a single ranking. The default battery RRFFusion (reciprocal-rank scoring, tuning-free) lives in hybrid.py; when you have an eval set and need to tune the relative weights of the two paths, implement your own (e.g. alpha-weighted / RSF normalized weighting) and inject it via HybridRetriever(fusion=).

fuse(result_lists, *, top_k) abstractmethod

Fuse multiple result lists, returning the top_k items.

Convention: HybridRetriever passes [dense path, keyword path]; rag query expansion (MQE/HyDE) passes one list per query. Fusion aligns by id (same id counts as the same item).

Parameters:

Name	Type	Description	Default
result_lists	List[List[RetrievalResult]]	Multiple result lists, each already sorted by relevance highest first.	required
top_k	int	Number of items to return after fusion.	required

Returns:

Type	Description
List[RetrievalResult]	List[RetrievalResult]: The top_k items after fusion and re-ranking.

KeywordIndex

Bases: ABC

Abstract base class for keyword retrieval. Subclasses implement add() and search(). metadata filtering follows the same convention as VectorStore.

add(ids, contents, *, scope=None, metadatas=None) abstractmethod

Batch write: each item = id + original text (+ optional metadata, same convention as VectorStore.add). Lists are equal length and aligned.

search(query, *, top_k=5, scope=None, filters=None) abstractmethod

Keyword retrieval, returning the top_k most BM25-relevant items (highest first) within the range limited by scope (+ optional filters).

delete(ids, *, scope=None) abstractmethod

Batch delete by id (within the range limited by scope, B semantics).

aadd(ids, contents, *, scope=None, metadatas=None) async

Async version of add (defaults to to_thread).

asearch(query, *, top_k=5, scope=None, filters=None) async

Async version of search (defaults to to_thread).

adelete(ids, *, scope=None) async

Async version of delete (defaults to to_thread).

close()

Release underlying resources (such as database connections). No-op by default; subclasses override as needed.

Reranker

Bases: ABC

Abstract base class for reranking. Subclasses implement rerank().

rerank(query, results, *, top_k=5) abstractmethod

Precisely re-order the candidate results by relevance to query, returning the top_k most relevant (score is the rerank score).

Parameters:

Name	Type	Description	Default
query	str	The query text.	required
results	List[RetrievalResult]	The candidates to rerank (typically from hybrid retrieval).	required
top_k	int	Number of items to return.	5

Returns:

Type	Description
List[RetrievalResult]	List[RetrievalResult]: The top_k items after reranking.

arerank(query, results, *, top_k=5) async

Async version of rerank (defaults to to_thread).

VectorStore

Bases: ABC

Abstract base class for vector storage + nearest-neighbor retrieval.

scope isolates different data (memory / rag, different user / agent, etc.): one shared foundation, mutually non-interfering. metadata filtering: the implementation decides which fields are filterable via its own declaration mechanism (the SQLite battery uses the constructor parameter metadata_columns=); add stores the declared fields, search narrows by filters. Filtering an undeclared field should fail loud.

add(ids, vectors, contents, *, scope=None, metadatas=None) abstractmethod

Batch write: each item = id + vector + original text (+ optional metadata). All lists are equal length and aligned.

Parameters:

Name	Type	Description	Default
ids	List[str]	Unique identifier per item.	required
vectors	List[List[float]]	Vector per item; dimension must match the one used to build the store.	required
contents	List[str]	Original text per item.	required
scope	Optional[Scope]	Ownership label; defaults to Scope() (all dimensions empty, i.e. unrestricted, B semantics). Writing the same (id, scope) again is an upsert (overwrite).	None
metadatas	Optional[List[dict]]	Optional, one metadata dict per item; only fields declared when building the index are stored as filterable columns, the rest are ignored (the full metadata still lives in each subsystem's source-of-truth store). Not passing it = all declared columns stored empty.	None

search(query_vector, *, top_k=5, scope=None, filters=None) abstractmethod

Given a query vector, return the top_k nearest items within the range limited by scope (+ optional filters).

Parameters:

Name	Type	Description	Default
query_vector	List[float]	The query vector.	required
top_k	int	Number of items to return.	5
scope	Optional[Scope]	Ownership filter; defaults to Scope() (B semantics: only filters non-empty dimensions).	None
filters	Optional[List[MetadataFilter]]	Optional metadata filter conditions (AND semantics, see MetadataFilter in types.py); filtering an undeclared field raises RetrievalError.	None

Returns:

Type	Description
List[RetrievalResult]	List[RetrievalResult]: Sorted by relevance, highest first.

delete(ids, *, scope=None) abstractmethod

Batch delete by id (within the range limited by scope, B semantics).

delete_exact(ids, *, scope=None)

Delete by exact write footprint (all-dimension match, deleting only rows whose footprint matches exactly, leaving sibling rows with the same id but different scope untouched). Defaults to falling back to delete (B semantics, backward compatible, does not break existing backends); backends that are exact across all dimensions (such as SqliteVecStore) should override. Used by HybridRetriever.add's compensating path: when rolling back a just-written vector, the delete range must equal the write footprint, otherwise it would wrongly delete same-id rows under other scopes.

aadd(ids, vectors, contents, *, scope=None, metadatas=None) async

Async version of add (defaults to to_thread).

asearch(query_vector, *, top_k=5, scope=None, filters=None) async

Async version of search (defaults to to_thread).

adelete(ids, *, scope=None) async

Async version of delete (defaults to to_thread).

adelete_exact(ids, *, scope=None) async

Async version of delete_exact (defaults to to_thread).

close()

Release underlying resources (such as database connections). No-op by default; subclasses override as needed.

HybridRetriever

Hybrid retriever (storage-agnostic): coordinates Embedder + VectorStore + KeywordIndex (+ optional Reranker / FusionStrategy), exposing add() / search().

This is the unified entry point the retrieval foundation gives the upper layers; both memory and rag use it, isolating their data by scope. Each dependency is injected, so the vector store, keyword, rerank, and fusion backends can all be replaced independently. add is best-effort compensating (the generic fallback when the two indexes use separate connections): if the keyword write fails, the vector just written is rolled back. For "single-transaction atomic writes across both indexes" (shared connection, old values not lost on update failure), use SqliteHybridRetriever in sqlite.py (constructed via build_sqlite_hybrid).

__init__(embedder, vector_store, keyword_index, reranker=None, *, config=None, fusion=None)

Parameters:

Name	Type	Description	Default
embedder	Embedder	Turns text into vectors.	required
vector_store	VectorStore	Vector storage and retrieval (dense).	required
keyword_index	KeywordIndex	Keyword retrieval (sparse / BM25).	required
reranker	Optional[Reranker]	Optional reranker; if omitted, only "vector + keyword + fusion" runs, and if given it refines after fusion.	None
config	Optional[RetrievalConfig]	Optional retrieval knobs (top_k / candidate_pool / rrf_k); defaults to RetrievalConfig() if omitted. Parsed once at construction into self.config; search's per-call kwargs may override it (three-level resolution).	None
fusion	Optional[FusionStrategy]	Optional fusion strategy (FusionStrategy, see base.py); defaults to RRFFusion(config.rrf_k) if omitted. Inject your own (alpha weighting / RSF etc.) when you have an evaluation set and want to tune the relative weights of the two paths.	None

add(ids, contents, *, scope=None, metadatas=None)

Write in one call (upsert), keeping the vector store and keyword index in sync (embedding is done internally).

Best-effort compensation: write the vector first, then the keyword; if the latter fails, delete the vector just written back out (to avoid a "vector present, keyword missing" half-write), and the original exception is re-raised as usual. Compensation rolls back cleanly for a failed insert; for a failed update-overwrite it only deletes without restoring the old value. To avoid this entirely, use SqliteHybridRetriever (shared connection, single transaction).

Parameters:

Name	Type	Description	Default
ids	List[str]	A unique identifier per entry.	required
contents	List[str]	The original text per entry.	required
scope	Optional[Scope]	Ownership label; defaults to Scope().	None
metadatas	Optional[List[dict]]	Optional, a metadata dict per entry (same length as ids); only fields the index has declared are stored as filterable columns (see base.py).	None

replace(old_ids, new_ids, contents, *, scope=None, metadatas=None)

Replace a document's old chunks with new-version chunks: first ingest the new chunks (add), then delete the old chunks outside of new. Used for RAG document re-ingestion dedup.

The base class is compensating "add first, then delete" (the two indexes use separate connections and cannot share a single transaction): old chunks are deleted only after all new chunks are ingested successfully; on failure it does not delete the old and does not lose data. For "atomic replace" (no window of concurrent double hits, no residue even if deleting the old fails), use SqliteHybridRetriever (shared connection, single transaction, see its replace override). Old and new chunk ids differ, so deleting the old does not touch the new.

delete(ids, *, scope=None, all_scopes=False)

Delete by id from the vector store and keyword index in sync (within the scope range).

Both deletes are idempotent (deleting a nonexistent id is a no-op), so if one side fails you can simply retry the whole delete to reach consistency. When the scope is entirely empty (a delete across the whole store), it is rejected unless all_scopes=True is passed explicitly (guard, see scope.py).

close()

Close the underlying vector store and keyword index resources.

search(query, *, top_k=None, candidate_pool=None, scope=None, all_scopes=False, filters=None)

Take candidate_pool entries each from vector and keyword -> fuse (RRF by default) -> (if a reranker) refine -> return top_k.

Parameters:

Name	Type	Description	Default
query	str	The query text.	required
top_k	Optional[int]	Final number of results; defaults to config.top_k if omitted (None uses the default, without swallowing 0).	None
candidate_pool	Optional[int]	How many entries each path (vector / keyword) takes into fusion / rerank; defaults to config.candidate_pool if omitted.	None
scope	Optional[Scope]	The ownership range to restrict retrieval to (B semantics: only filters non-empty dimensions); defaults to Scope().	None
all_scopes	bool	Guard switch; when the scope is entirely empty (searching the whole store) this must be set to True explicitly, otherwise it is rejected (see scope.py).	False
filters	Optional[List[MetadataFilter]]	Optional metadata filter (AND semantics, see MetadataFilter in types.py); both retrieval paths narrow candidates by it first (pre-filtering).	None

Returns:

Type	Description
List[RetrievalResult]	List[RetrievalResult]: The final top_k entries.

search_many(queries, *, top_k=None, candidate_pool=None, scope=None, all_scopes=False, filters=None)

Batch-retrieve for multiple queries: embed all queries at once, then run vector + keyword retrieval (+ optional rerank) for each.

Returns a list the same length as queries (one top_k ranking per query). Compared with searching one by one, this saves N-1 embedding network round-trips; rag's query expansion (MQE / HyDE) uses it for multiple queries, batching N embeddings into one. Parameters are the same as search (queries is the list of queries).

asearch(query, *, top_k=None, candidate_pool=None, scope=None, all_scopes=False, filters=None) async

Async version of search: aembed gets the query vector, then _afuse_one runs the vector path / keyword path concurrently via gather.

asearch_many(queries, *, top_k=None, candidate_pool=None, scope=None, all_scopes=False, filters=None) async

Async version of search_many: aembed all queries at once, then each query runs the two paths concurrently via _afuse_one.

aadd(ids, contents, **kwargs) async

Async version of add (wraps its own sync atomic version in to_thread; the write path is not hot and preserves cross-index single-transaction semantics).

areplace(old_ids, new_ids, contents, **kwargs) async

Async version of replace (wraps the sync atomic version in to_thread).

adelete(ids, **kwargs) async

Async version of delete (to_thread).

RRFFusion

Bases: FusionStrategy

The default battery for the fusion strategy: wraps reciprocal_rank_fusion (a tuning-free baseline).

__init__(k=_RRF_K)

k: the RRF smoothing constant (default 60).

fuse(result_lists, *, top_k)

Fuse via RRF (see reciprocal_rank_fusion).

IndexSync

Bases: ABC

The derived-index sync seam: a subsystem's write path propagates changes to the retrieval index and reconciles through it.

Subclasses decide sync / async and whether to persist (the default SyncIndexSync writes through synchronously; production may switch to outbox + worker).

index(ids, contents, *, scope=None, metadatas=None) abstractmethod

Upsert several entries into the index (idempotent: unchanged content is skipped). Best-effort: does not raise on failure, marks pending.

metadatas (optional, same length as ids): a metadata dict per entry, passed through to the retrieval foundation to be stored as filterable columns (see MetadataFilter in retrieval/types.py).

replace(old_ids, new_ids, contents, *, scope=None, metadatas=None) abstractmethod

Atomically replace old->new (whole batch, by-doc, with no "old and new coexisting" window). Raises on failure, so the caller rolls back its source-of-truth side and keeps the old version.

drop(ids, *, scope=None) abstractmethod

Delete several entries from the index. Best-effort: does not raise on failure, marks pending, and leaves it to reconcile / read-time self-heal.

reconcile(items, *, scope=None, batch_size=256) abstractmethod

Make the index converge with items (a source-of-truth snapshot; elements must have .id / .content) as the authority: drop orphans + force-reingest the source. Returns the number reingested.

pending(*, scope=None) abstractmethod

The currently pending ids (recent best-effort write failures not yet converged by reconcile). For apps to monitor / trigger reconciliation.

tracked_ids(*, scope=None)

All ids currently indexed (fingerprint registered) in this scope, for consistency cross-checks (e.g. IngestionPipeline.verify).

Optional capability: the default implementation SyncIndexSync delegates to bookkeeping; a custom IndexSync that cannot enumerate may leave it unoverridden (the verifier skips it on NotImplementedError). Non-abstract, so it does not force existing subclasses to implement it and does not break compatibility.

close()

Release resources held by this seam (e.g. the bookkeeping DB connection / worker). Default is a no-op: resource-free implementations need not override (non-abstract, does not break existing subclasses).

InMemoryBookkeeping

Bases: SyncBookkeeping

The default bookkeeping: an in-process dict (zero dependency, zero overhead). Cleared on restart, so the idempotent-skip and pending set are lost with it, but no data is lost: reconcile / rebuild can still fully rebuild from the source-of-truth store, only the first write after restart does one extra full embedding.

SqliteBookkeeping

Bases: SyncBookkeeping

Persistent bookkeeping (aligned with the de-facto standard form of LangChain's RecordManager): one SQLite table storing (scope, id, fingerprint, timestamp).

Benefits: (1) idempotent skipping takes effect cross-process (unchanged content is no longer re-embedded after restart, mainly benefiting the index path of Memory's write path); (2) the pending set is not lost on restart (entries whose index write failed can still be converged by reconcile after restart); (3) the pending_since timestamp lets an app monitor "how long the oldest has been stuck". This table is naturally the seed of a future async outbox.

__init__(db_path, *, table='index_sync_bookkeeping')

db_path: the SQLite file path (co-locating with the source-of-truth store is fine, since bookkeeping is companion state of the ingestion pipeline).

close()

Close the database connection.

SyncBookkeeping

Bases: ABC

The bookkeeping-storage seam for SyncIndexSync: content fingerprints + pending set. Default in-process, switchable to SQLite (cross-process persistence).

get_hash(scope, id) abstractmethod

Read one registered content fingerprint; returns None if absent.

set_hashes(scope, pairs) abstractmethod

Register (id, fingerprint) in batch; overwrites for the same (id, scope).

delete_hashes(scope, ids) abstractmethod

Remove fingerprint registrations in batch (after entries are deleted from the index). Idempotent.

tracked_ids(scope) abstractmethod

All ids with a registered fingerprint in this scope (used for reconcile's orphan detection).

mark_pending(scope, ids) abstractmethod

Mark several ids as pending (on best-effort write failure); repeated marking keeps the earliest time.

clear_pending(scope, ids) abstractmethod

Remove several ids from the pending set after a successful write. Idempotent.

pending_ids(scope) abstractmethod

The set of currently pending ids in this scope (a copy).

close()

Release resources held by bookkeeping (e.g. the SQLite connection). Default is a no-op: the in-process implementation (InMemoryBookkeeping) need not override.

SyncIndexSync

Bases: IndexSync

The default implementation: write through synchronously to the retrieval foundation + bookkeeping (content fingerprint for idempotency, pending set for self-heal).

Bookkeeping defaults to in-process (InMemoryBookkeeping, cleared on restart, rebuildable by reconcile from the source-of-truth store); pass bookkeeping=SqliteBookkeeping(db_path) for cross-process persistence (from_config installs it by default). For fully async / distributed (outbox + worker), switch to a different IndexSync implementation without touching each subsystem's write path.

__init__(retriever, *, bookkeeping=None, tracer=None)

retriever: the retrieval foundation (HybridRetriever, needs add / replace / delete, signatures in hybrid.py). bookkeeping: optional bookkeeping storage; defaults to the in-process default if omitted (behaves as before). tracer: optional tracer (duck-typed emit); once attached, write failures (marked pending) and completed reconciliations emit index_sync events. Zero overhead when not attached.

index(ids, contents, *, scope=None, metadatas=None)

Idempotent upsert: skip unchanged entries by content fingerprint; write to the foundation, and on failure mark this batch of ids pending, without raising or rolling back the source-of-truth store.

replace(old_ids, new_ids, contents, *, scope=None, metadatas=None)

Atomically replace old->new (delegated to retriever.replace, with atomicity guaranteed by the foundation); raises on failure, does not touch bookkeeping, and the caller rolls back its source-of-truth side.

On success, update bookkeeping: register the new chunks' fingerprints and remove the replaced-out old chunks (the old outside of new).

drop(ids, *, scope=None)

Delete from the index; on failure mark this batch of ids pending (leaving orphans, with reconcile / read-time self-heal as backstop), without raising.

reconcile(items, *, scope=None, batch_size=256)

Reconcile with the source-of-truth snapshot items as authority: first drop orphans (in bookkeeping but no longer in the source), then force-reingest the source in batches (ignoring fingerprints, because the index may be entirely lost / swapped to an empty backend, in which case fingerprints would wrongly judge "already indexed"). Returns the number reingested. Unregistered orphans are backstopped by read-time self-heal.

pending(*, scope=None)

A copy of the pending ids (read-only snapshot).

tracked_ids(*, scope=None)

Delegate to bookkeeping: all ids with a registered fingerprint (already indexed) in this scope.

close()

Close bookkeeping resources (delegates to bookkeeping.close; e.g. the SqliteBookkeeping DB connection).

Scope dataclass

The ownership label for data. Every dimension is optional; omitting one means "do not restrict this dimension" (B semantics).

Fields

base: Subsystem distinction (memory / rag, etc.); leave empty to not restrict. user: User identifier (the key to multi-user isolation, the minimal security boundary). agent: Agent identifier (in a multi-agent system each agent keeps its own records). session: Session identifier (= run_id, the transient context of a single conversation). app: Application / organization identifier (shared context).

MetadataFilter dataclass

One metadata filter condition: field key, comparison value, and operator op (eq for equality / in for one-of-many).

Multiple conditions are AND'd (a hit must satisfy all of them). Filterable fields must be declared when building the index (the SQLite battery uses metadata_columns=); filtering an undeclared field fails loud, since a silent empty result is harder to diagnose than an error.

Example

MetadataFilter("doc_id", "abc123") -> doc_id = 'abc123' MetadataFilter("tag", ["faq", "policy"], op="in") -> tag IN ('faq', 'policy')

RetrievalConfig dataclass

Tunable knobs for the hybrid retrieval foundation (shared by memory / rag; inject a separate one per subsystem if you want them heterogeneous). All frozen; scalars are inherently immutable.

Fields

top_k: Final number of items to return. candidate_pool: How many items each path (dense / keyword) fetches into fusion / rerank (must be >= top_k). rrf_k: The smoothing constant for RRF fusion.

validate()

Range check: top_k >= 1, candidate_pool >= top_k, rrf_k >= 1.

RetrievalResult dataclass

One retrieval result: the hit content + relevance score + source + identifier + metadata.

Fields

content: The hit text body. score: Relevance score, by convention "higher is more relevant"; the actual value is produced by each retrieval implementation, used only for ordering, and not guaranteed comparable across implementations. source: Source identifier, such as "memory" / "rag" / a document name, to help the upper layers distinguish and trace. id: The item's unique identifier within its source, usable for going back to the original text / dedup; empty string if absent. embedding: The item's content vector; vector retrieval can carry it back for free (vec0's vec_to_json), for reuse by context engineering's MMR to avoid recomputation. metadata: Attached information (time, type, raw distance, etc.), defaults to an empty dict.

__str__()

Make print(result) show "[score] body (truncated if too long)", for easy debugging.

Returns:

Name	Type	Description
str	str	Something like "[0.812] I can't eat nuts".

build_sqlite_hybrid(embedder, *, db_path=':memory:', reranker=None, vec_table='vec_items', kw_table='kw_items', config=None, metadata_columns=(), fusion=None)

Convenience constructor: the vector store + keyword index share one SQLite connection -> add / delete is atomic across both indexes in a single transaction.

The recommended local construction: compared to two stores each with their own exclusive connection (where writes can only be best-effort compensated), a shared connection makes both indexes' writes either succeed together or roll back together, with no half-write, and a failed update does not lose the old value. On open it does an embedder fingerprint check (index_meta table): swapping in a mismatched embedding model errors out directly, preventing "silent mixed writes, degraded retrieval".

Parameters:

Name	Type	Description	Default
embedder	Embedder	Text-to-vector (embedder.dim builds the vector table; model_id goes into the fingerprint).	required
db_path	str	SQLite file path; defaults to ":memory:" for self-tests only, use a file path in production to persist.	':memory:'
reranker	Optional[Reranker]	Optional reranker.	None
vec_table / kw_table		The two indexes' table names (same database, different tables).	required
metadata_columns		Declares which metadata fields are filterable (e.g. ("doc_id", "tag")), shared by both indexes; defaults to none. Fixed at table-creation time; adding fields later requires a new table rebuild.	()
fusion		Optional fusion strategy (FusionStrategy); if omitted, use the RRF default.	None

Returns:

Name	Type	Description
SqliteHybridRetriever	SqliteHybridRetriever	A hybrid retriever wired to a shared connection with atomic add / delete.

reciprocal_rank_fusion(result_lists, *, k=_RRF_K, top_k=10)

Fuse multiple "each already sorted" result lists into one ranking via RRF.

A result ranked r-th (starting at 1) in a given list contributes 1/(k+r) points; the scores for the same id across multiple lists are summed, and finally the top_k are taken by descending total score. Entries hit by multiple paths at once naturally score higher.

Fusion aligns by id (in HybridRetriever, two paths sharing an id are the same entry, and merging by id to add scores is exactly the point of RRF); when the id is empty it degrades to aggregating by content, to avoid wrongly merging distinct empty-id results. Note: if you use this function to fuse results from different corpora that happen to share ids, the caller must first make the ids globally unique.

Parameters:

Name	Type	Description	Default
result_lists	List[List[RetrievalResult]]	Multiple result lists, each already sorted internally from most to least relevant.	required
k	int	Smoothing constant, default 60.	_RRF_K
top_k	int	Number of results to return after fusion.	10

Returns:

Type	Description
List[RetrievalResult]	List[RetrievalResult]: The top_k results after fusion re-ranking, with score being the RRF total.

require_valid_top_k(top_k, *, candidate_pool=None)

Validate the number of results to return: top_k must be at least 1, and if candidate_pool is given it must be >= top_k (otherwise the candidate pool is smaller than the final requirement).

This unifies how backends handle an invalid top_k; otherwise FTS5's LIMIT -1 would "return everything" and vec0's k=-1 would raise outright, producing inconsistent behavior.

require_explicit_scope(scope, all_scopes, action)

Guardrail for destructive / global operations: reject when scope is fully empty and all_scopes=True was not passed explicitly, to avoid accidentally acting on the whole database.

The mechanism provides the guardrail, the caller provides the rule: by default it forbids slip-ups like a bare Scope() that deletes / searches across the whole database, and if global scope is truly needed the caller must explicitly opt in. The framework's built-in memory / rag always carry a (non-empty) base and are unaffected: this only blocks a fully-unrestricted Scope().

Parameters:

Name	Type	Description	Default
scope	Scope	The ownership range to check.	required
all_scopes	bool	Whether the caller explicitly declares "yes, this should apply to all scopes".	required
action	str	The operation name, used in the error message (e.g. "delete" / "search").	required

scope_is_empty(scope)

True if no dimension has a value (i.e. no dimension is restricted, so the operation applies to the whole database). Scope fields are the source of truth for the dimensions.

scope_column_names()

The list of scope's underlying column names (used when creating tables / on INSERT).

scope_exact_where(scope)

The AND fragment and params that exactly match all dimensions (including empty ones, compared as empty string), for appending after an existing WHERE.

Unlike scope_where (B semantics, filters only non-empty dimensions), this brings all five dimensions into the equality match, comparing empty dimensions against their stored value "" (see scope_store_values), so it only hits rows whose scope footprint matches exactly. The pre-write upsert (delete-old-then-insert by (id, scope)) uses this to avoid wrongly deleting sibling rows with the same id but a different scope.

Returns:

Type	Description
(sql, params)	sql looks like " AND base = ? AND sc_user = ? AND ... (all five dimensions)", with params
List[str]	aligned to the column order.

scope_exact_where_clause(scope)

The full WHERE clause and params that exactly match all dimensions (including empty ones, compared as empty string), for queries with no other filter conditions.

This is the "own WHERE" version of scope_exact_where (as scope_where_clause is to scope_where): it brings all five dimensions into the equality match, so it only hits rows whose scope footprint matches exactly. Storage that does "point-access by exact scope", such as sessions / checkpoint, uses this: an empty scope (Scope()) only hits the all-empty bucket, not the whole table (avoiding wrongly reading / deleting all sessions when scope is not passed). It always carries a WHERE (all five dimensions, matching against empty strings even when no dimension is non-empty), and unlike B semantics does not degrade to an empty string.

Returns:

Type	Description
(sql, params)	sql looks like " WHERE base = ? AND sc_user = ? AND ... (all five dimensions)", with params
List[str]	aligned to the column order.

scope_store_values(scope)

The list of values for storage; None -> empty string, to make column filtering easy.

Shares the same source as scope_column_names (both iterate _SCOPE_COLS) -> strictly aligned: editing _SCOPE_COLS in one place guarantees that, within the code, stored values and column names do not drift (drift would write values into the wrong column and leak data across scopes without error). Note: this only holds for newly-created databases. _SCOPE_COLS is the persistence schema contract for every SQLite table; adding or removing a dimension is a whole-database schema change, and existing old databases must be migrated (each store's open-time self-check will block incompatible old databases). See scope.py and doc/retrieval/scope.md.

scope_where(scope)

The AND fragment and params for queries (B semantics: only non-empty dimensions), for appending after an existing WHERE.

Returns:

Type	Description
(sql, params)	sql looks like " AND base = ? AND sc_user = ?" (with a leading space, ready to concatenate);
List[str]	returns ("", []) when there is no non-empty dimension.

scope_where_clause(scope)

The full WHERE clause and params (B semantics), for queries with no other filter conditions.

Returns:

Type	Description
(sql, params)	sql looks like " WHERE base = ? AND sc_user = ?" (with a leading space);
List[str]	returns ("", []) when there is no non-empty dimension.