Embeddings¶
Every name on this page is importable from sci_etl_core.embeddings.
AsyncOpenAIEmbedder(
api_key: str | SecretStr,
base_url: str,
model: str,
batch_size: int = 128,
max_retries: int = 3,
backoff_factor: float = 2.0,
sleep: Any = sleep,
max_retry_after: float = 60.0,
rate_limiter: RateLimiting | None = None,
)
Bases: AsyncEmbedder
Embed text through any OpenAI-compatible /embeddings endpoint.
Configure the embedder.
This embedder is the only retry layer: the OpenAI SDK's own retries are
turned off, so max_retries is the total number of attempts per
batch. Between attempts it waits backoff_factor ** attempt seconds,
or longer when the server's retry-after-ms or Retry-After header
asks for it, up to max_retry_after seconds.
Every attempt first enters rate_limiter, an
:class:~sci_etl_core.rate_limiter.AsyncRateLimiter or a
:class:~sci_etl_core.rate_limiter.HostRateLimiter matched against
base_url, and releases it once the response arrives. Share one
limiter between a chat client and an embedder that call the same
provider to keep both inside one budget.
Raises:
| Type | Description |
|---|---|
ValueError
|
|
Bases: AsyncEmbedder
Embed text locally with the optional sentence-transformers package.
Runs with no network access, trading a heavier install for zero per-request
cost. The blocking encode call is offloaded to a worker thread so the
event loop is never stalled. A preloaded model may be injected to keep the
hard dependency out of import-time and test paths.
Bases: TextChunker
Fixed-size word windows with overlap between neighbors.
Word counts approximate token counts (English runs ~1.3 tokens per word), so the default keeps each window comfortably under the 512-token ceiling of the common small embedding models. The overlap preserves context that would otherwise be severed at a window boundary.
When overlap_words is left unset it is derived from chunk_words so a
small custom window never collides with the standard overlap; an explicit
overlap is always validated strictly.
dataclass
¶
EmbeddingChunk(
record_id: str,
chunk_index: int,
text: str,
vector: list[float],
metadata: dict[str, Any] = dict(),
)
A single passage of an article, paired with its embedding vector.
dataclass
¶
SearchHit(
record_id: str,
chunk_index: int,
text: str,
score: float,
metadata: dict[str, Any] = dict(),
)
A stored chunk returned from a similarity query, with its score.
dataclass
¶
The passages a vector memory holds for one record, without their vectors.
passages are the chunk texts in chunk_index order, and metadata
is the metadata of the record's first chunk, which for chunks stored by
:class:~sci_etl_core.embeddings.ingest_async.AsyncChunkIngestor holds the
record's title and source_url.
Bases: ABC
An accumulating vector memory of article chunks, searchable by meaning.
abstractmethod
async
¶
add(chunks: Sequence[EmbeddingChunk]) -> None
Persist chunk embeddings, replacing any with the same id and index.
abstractmethod
async
¶
delete_record(record_id: str) -> None
Remove every stored chunk belonging to record_id.
async
¶
replace_record(
record_id: str, chunks: Sequence[EmbeddingChunk]
) -> None
Make chunks the only stored chunks of record_id.
Adding alone would leave a record's old higher-index chunks behind when its text now yields fewer passages. The default deletes and then adds; a backend that can do both atomically should override it.
abstractmethod
async
¶
query(
vector: Sequence[float],
top_k: int = 5,
min_score: float = -1.0,
exclude_record_id: str | None = None,
) -> list[SearchHit]
Return the nearest stored chunks to vector by cosine similarity.
exclude_record_id drops a record's own chunks, so an article can be
compared against every other article without matching itself. A stored
vector whose score is not finite, such as one holding NaN, never matches.
iter_records(
batch_size: int = 100,
) -> AsyncIterator[StoredRecord]
Yield every stored record's passages, in record_id order, without loading vectors.
batch_size is how many records are read at a time. A backend that
cannot enumerate its chunks keeps this default, which raises; the
bundled stores implement it.
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
The store cannot enumerate its records. |
Bases: AsyncEmbeddingStore
Non-persistent store backed by a dict. Vectors are unit-normalized.
Chunks are keyed by (record_id, chunk_index) so that re-adding an
article replaces its earlier chunks instead of accumulating duplicates.
That mirrors the primary key of the durable SQLite store, keeping the two
backends interchangeable: re-ingesting a record must not silently give it
extra weight in one of them.
async
¶
query(
vector: Sequence[float],
top_k: int = 5,
min_score: float = -1.0,
exclude_record_id: str | None = None,
) -> list[SearchHit]
async
¶
iter_records(
batch_size: int = 100,
) -> AsyncIterator[StoredRecord]
Yield every stored record's passages, from a snapshot taken when iteration starts.
Raises:
| Type | Description |
|---|---|
ValueError
|
|
Bases: AsyncEmbeddingStore
Durable vector memory persisting unit-normalized float32 chunk vectors.
Backed by the standard-library sqlite3 driver run off the event loop, so
it carries no third-party dependency. The single connection must never be
used by two worker threads at once, so every operation holds an
:class:asyncio.Lock until its thread finishes, even if the awaiting task
is cancelled meanwhile. Each write is one transaction, rolled back on
failure, and every SQLite failure (including a file that is not a
database) surfaces as :class:EmbeddingStoreError, as does a file written
by a newer sci-etl-core. The file records its schema version in
PRAGMA user_version.
Similarity is a linear scan computed in NumPy: every stored vector is loaded and scored against the query in one matrix product, and only the best chunks' text and metadata are read. The decoded vectors stay in memory between queries and are reloaded only after a write, from this store or any other connection to the file, so repeated queries skip the read. This is exact and dependency-light, and fits corpora up to the low millions of chunks; swap in an ANN index behind this same interface if the memory outgrows a full scan.
The file is opened, and its schema created, on first use. Use each instance from one event loop.
async
¶
replace_record(
record_id: str, chunks: Sequence[EmbeddingChunk]
) -> None
Delete the record's chunks and insert chunks in one transaction.
async
¶
query(
vector: Sequence[float],
top_k: int = 5,
min_score: float = -1.0,
exclude_record_id: str | None = None,
) -> list[SearchHit]
async
¶
iter_records(
batch_size: int = 100,
) -> AsyncIterator[StoredRecord]
Yield every stored record's passages, reading batch_size records at a time.
Each batch is read in one transaction and continues after the last
record_id of the batch before, so records written during iteration
are yielded when their id sorts after that point, and a record never
repeats. No vector is loaded.
Raises:
| Type | Description |
|---|---|
ValueError
|
|
EmbeddingStoreError
|
The store cannot be read, or a chunk's metadata is not a JSON object. |
AsyncChunkIngestor(
chunker: TextChunker,
embedder: AsyncEmbedder,
store: AsyncEmbeddingStore,
)
Chunk an article's full text, embed each passage, and add it to memory.
Intended as the second stage after an abstract-level relevance gate: only
articles that already cleared the gate reach here, so the cost of embedding
a whole body is spent only on records worth remembering. It satisfies
:class:~sci_etl_core.ingest_protocol.MemoryIngestor.
async
¶
Replace the record's stored passages with those of text.
Every earlier chunk of the record is removed, so re-ingesting text that
yields fewer passages leaves no stale chunks behind, and text with no
passages clears the record from memory. Each chunk's metadata is the
record's title and source_url only; record.metadata is not
stored in the vector memory.
Returns:
| Type | Description |
|---|---|
int
|
The number of chunks stored. |
Raises:
| Type | Description |
|---|---|
EmbeddingError
|
The embedder returned a different number of vectors than it was given passages. |
EmbeddingStoreError
|
The store could not be written. |
AsyncSimilarArticleFinder(
embedder: AsyncEmbedder, store: AsyncEmbeddingStore
)
Search the accumulated memory for passages resembling a piece of text.
async
¶
find_similar_chunks(
text: str,
top_k: int = 5,
min_score: float = 0.0,
exclude_record_id: str | None = None,
) -> list[SearchHit]
Return up to top_k stored chunks most similar to text, best first.
text is embedded in one call, and the other arguments are those of
:meth:~sci_etl_core.embeddings.store_base.AsyncEmbeddingStore.query,
except that min_score defaults to 0. An embedder that returns no
vector gives no hits.
async
¶
find_similar_articles(
text: str,
top_k: int = 5,
min_score: float = 0.0,
exclude_record_id: str | None = None,
chunk_pool: int = 50,
) -> list[tuple[str, float, dict[str, Any]]]
Rank whole articles, scoring each by its single best-matching chunk.
The chunk_pool best chunks are collapsed to one entry per record, so
at most chunk_pool records come back, and often far fewer when long
articles own many of those chunks; raise chunk_pool along with
top_k. Each entry is (record_id, score, metadata), where
metadata is the best chunk's; :meth:find_best_chunks also returns
the chunk's text.
async
¶
find_best_chunks(
text: str,
top_k: int = 5,
min_score: float = 0.0,
exclude_record_id: str | None = None,
chunk_pool: int = 50,
) -> list[SearchHit]
Rank whole articles as :meth:find_similar_articles does, returning each one's best chunk.
Each hit is the chunk that scored its record, text included, so a result can show the passage that made it similar. When two chunks of a record score the same, the one ranked first by the store is kept.