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.
Similarity is a linear scan computed in NumPy: every stored vector is loaded and dotted against the query. 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.