jeevesagent.vectorstore

Vector stores for semantic search over Chunk / Document objects.

Unified async interface (modeled on LangChain’s VectorStore but properly async-first and typed against our Chunk / Document from jeevesagent.loader):

• add() — embed + store chunks; returns their ids.

• delete() — remove by id.

• search() — top-k by cosine similarity + metadata filter.

• search_by_vector() — same with a precomputed query vector.

Implementations:

• InMemoryVectorStore — default; zero-deps; cosine over a Python list. Great for dev / tests / small corpora.

• ChromaVectorStore — wraps chromadb for persistent on-disk or hosted Chroma. Lazy import.

• PostgresVectorStore — wraps pgvector via asyncpg. Production durable. Lazy import.

• FAISSVectorStore — wraps faiss-cpu for fast in-memory ANN search over large corpora. Lazy import.

One-liner usage:

from jeevesagent import HashEmbedder
from jeevesagent.vectorstore import InMemoryVectorStore
from jeevesagent.loader import load, MarkdownChunker

vs = InMemoryVectorStore(embedder=HashEmbedder())

doc = load("research.pdf")
chunks = MarkdownChunker().split(doc.content, source=str(doc.metadata["source"]))
await vs.add(chunks)

results = await vs.search("what is RAG?", k=5)
for r in results:
    print(f"{r.score:.3f}: {r.chunk.content[:100]}")

Optional dependencies:

pip install 'jeevesagent[vectorstore-chroma]'
pip install 'jeevesagent[vectorstore-postgres]'
pip install 'jeevesagent[vectorstore-faiss]'
pip install 'jeevesagent[vectorstore]'              # all of the above

Submodules

• jeevesagent.vectorstore.base
• jeevesagent.vectorstore.chroma
• jeevesagent.vectorstore.faiss
• jeevesagent.vectorstore.inmemory
• jeevesagent.vectorstore.postgres

Classes

ChromaVectorStore	Vector store backed by chromadb.
FAISSVectorStore	Vector store backed by faiss-cpu.
InMemoryVectorStore	In-process vector store backed by a Python list.
PostgresVectorStore	Vector store backed by Postgres + pgvector.
SearchResult	One hit from VectorStore.search().
VectorStore	Async protocol for vector stores.

Package Contents

class jeevesagent.vectorstore.ChromaVectorStore(embedder: jeevesagent.core.protocols.Embedder, *, collection_name: str = 'jeeves_vectors', persist_directory: str | None = None, client: Any = None)

Vector store backed by chromadb.

async add(chunks: list[jeevesagent.loader.base.Chunk], ids: list[str] | None = None) → list[str]

async count() → int

async delete(ids: list[str]) → None

classmethod from_chunks(chunks: list[jeevesagent.loader.base.Chunk], *, embedder: jeevesagent.core.protocols.Embedder, ids: list[str] | None = None, collection_name: str = 'jeeves_vectors', persist_directory: str | None = None, client: Any = None) → ChromaVectorStore

Async:

One-shot: construct a ChromaVectorStore + add chunks.

classmethod from_texts(texts: list[str], *, embedder: jeevesagent.core.protocols.Embedder, metadatas: list[dict[str, Any]] | None = None, ids: list[str] | None = None, collection_name: str = 'jeeves_vectors', persist_directory: str | None = None, client: Any = None) → ChromaVectorStore

Async:

One-shot: construct a ChromaVectorStore from raw text strings (each becomes a Chunk with the matching metadata dict, or empty if metadatas is None).

async get_by_ids(ids: list[str]) → list[jeevesagent.loader.base.Chunk]

async search(query: str, *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[jeevesagent.vectorstore.base.SearchResult]

async search_by_vector(vector: list[float], *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[jeevesagent.vectorstore.base.SearchResult]

property embedder: jeevesagent.core.protocols.Embedder

name = 'chroma'

class jeevesagent.vectorstore.FAISSVectorStore(embedder: jeevesagent.core.protocols.Embedder, *, dimension: int | None = None, index_factory_string: str = 'HNSW32', metric: str = 'ip')

Vector store backed by faiss-cpu.

async add(chunks: list[jeevesagent.loader.base.Chunk], ids: list[str] | None = None) → list[str]

async count() → int

async delete(ids: list[str]) → None

classmethod from_chunks(chunks: list[jeevesagent.loader.base.Chunk], *, embedder: jeevesagent.core.protocols.Embedder, ids: list[str] | None = None, dimension: int | None = None, index_factory_string: str = 'HNSW32', metric: str = 'ip') → FAISSVectorStore

Async:

One-shot: construct a FAISSVectorStore + add chunks.

classmethod from_texts(texts: list[str], *, embedder: jeevesagent.core.protocols.Embedder, metadatas: list[dict[str, Any]] | None = None, ids: list[str] | None = None, dimension: int | None = None, index_factory_string: str = 'HNSW32', metric: str = 'ip') → FAISSVectorStore

Async:

One-shot: construct a FAISSVectorStore from raw text strings (each becomes a Chunk with the matching metadata dict, or empty if metadatas is None).

async get_by_ids(ids: list[str]) → list[jeevesagent.loader.base.Chunk]

async search(query: str, *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[jeevesagent.vectorstore.base.SearchResult]

async search_by_vector(vector: list[float], *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[jeevesagent.vectorstore.base.SearchResult]

property embedder: jeevesagent.core.protocols.Embedder

name = 'faiss'

class jeevesagent.vectorstore.InMemoryVectorStore(embedder: jeevesagent.core.protocols.Embedder)

In-process vector store backed by a Python list.

async add(chunks: list[jeevesagent.loader.base.Chunk], ids: list[str] | None = None) → list[str]

async count() → int

async delete(ids: list[str]) → None

classmethod from_chunks(chunks: list[jeevesagent.loader.base.Chunk], *, embedder: jeevesagent.core.protocols.Embedder, ids: list[str] | None = None) → InMemoryVectorStore

Async:

One-shot: construct an InMemoryVectorStore + add chunks.

classmethod from_texts(texts: list[str], *, embedder: jeevesagent.core.protocols.Embedder, metadatas: list[dict[str, Any]] | None = None, ids: list[str] | None = None) → InMemoryVectorStore

Async:

One-shot: construct an InMemoryVectorStore from raw text strings (each becomes a Chunk with the matching metadata dict, or empty if metadatas is None).

async get_by_ids(ids: list[str]) → list[jeevesagent.loader.base.Chunk]

classmethod load(path: str | pathlib.Path, *, embedder: jeevesagent.core.protocols.Embedder) → InMemoryVectorStore

Async:

Restore a store previously save()-d. Pass the same embedder kind/dimensions or queries will produce nonsense scores.

async save(path: str | pathlib.Path) → None

Write the full store (chunks + vectors + ids) to a JSON file. The embedder is NOT serialized — supply the same embedder when calling load().

async search(query: str, *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[jeevesagent.vectorstore.base.SearchResult]

async search_by_vector(vector: list[float], *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[jeevesagent.vectorstore.base.SearchResult]

async search_hybrid(query: str, *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, alpha: float = 0.5) → list[jeevesagent.vectorstore.base.SearchResult]

Hybrid lexical (BM25) + vector search via RRF.

alpha is in [0, 1]: 0 = pure BM25, 1 = pure vector, 0.5 = even weighting (RRF default). Both rankings are computed independently and fused by Reciprocal Rank Fusion, then the top-k survivors are returned.

Embeddings catch semantic similarity (“automobile” ↔ “car”), BM25 catches exact-term hits (model names, error codes, person names) — together they outperform either alone on most retrieval benchmarks.

property embedder: jeevesagent.core.protocols.Embedder

name = 'in-memory'

class jeevesagent.vectorstore.PostgresVectorStore(embedder: jeevesagent.core.protocols.Embedder, *, dsn: str, table: str = 'jeeves_vectors', dimension: int | None = None)

Vector store backed by Postgres + pgvector.

async add(chunks: list[jeevesagent.loader.base.Chunk], ids: list[str] | None = None) → list[str]

async count() → int

async delete(ids: list[str]) → None

classmethod from_chunks(chunks: list[jeevesagent.loader.base.Chunk], *, embedder: jeevesagent.core.protocols.Embedder, ids: list[str] | None = None, dsn: str, table: str = 'jeeves_vectors', dimension: int | None = None) → PostgresVectorStore

Async:

One-shot: construct a PostgresVectorStore + add chunks.

classmethod from_texts(texts: list[str], *, embedder: jeevesagent.core.protocols.Embedder, metadatas: list[dict[str, Any]] | None = None, ids: list[str] | None = None, dsn: str, table: str = 'jeeves_vectors', dimension: int | None = None) → PostgresVectorStore

Async:

One-shot: construct a PostgresVectorStore from raw text strings (each becomes a Chunk with the matching metadata dict, or empty if metadatas is None).

async get_by_ids(ids: list[str]) → list[jeevesagent.loader.base.Chunk]

async init_schema(dimension: int) → None

Create the table + HNSW index. Idempotent.

async search(query: str, *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[jeevesagent.vectorstore.base.SearchResult]

async search_by_vector(vector: list[float], *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[jeevesagent.vectorstore.base.SearchResult]

property embedder: jeevesagent.core.protocols.Embedder

name = 'postgres'

class jeevesagent.vectorstore.SearchResult

One hit from VectorStore.search().

• chunk — the matched chunk (with its full metadata).

• score — similarity in [-1, 1] for cosine; backend- specific for other distance metrics. Higher = more similar.

• id — the store-assigned id (so callers can delete() or get_by_ids() later).

chunk: jeevesagent.loader.base.Chunk

id: str

score: float

class jeevesagent.vectorstore.VectorStore

Bases: Protocol

Async protocol for vector stores.

Six methods cover the lifecycle: add (embed + store), delete, search (by query string), search_by_vector (precomputed), count, get_by_ids.

Backends that aren’t natively async (FAISS, Chroma) wrap their sync calls in anyio.to_thread.run_sync() so they don’t block the event loop.

async add(chunks: list[jeevesagent.loader.base.Chunk], ids: list[str] | None = None) → list[str]

Embed + store chunks. Returns the assigned ids (caller-provided or generated).

async count() → int

Number of chunks currently in the store.

async delete(ids: list[str]) → None

Remove the named chunks. Unknown ids are silently skipped (idempotent).

async get_by_ids(ids: list[str]) → list[jeevesagent.loader.base.Chunk]

Fetch chunks by id, in the same order as ids. Unknown ids are skipped (the result may be shorter than the input).

async search(query: str, *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[SearchResult]

Embed query and return the top-k chunks ranked by similarity. filter (optional) restricts candidates by metadata. diversity (optional, 0..1) enables MMR reranking for varied results.

async search_by_vector(vector: list[float], *, k: int = 4, filter: collections.abc.Mapping[str, Any] | None = None, diversity: float | None = None) → list[SearchResult]

Same as search() but with a precomputed query vector.