RandomZ / app /vectorstore /base.py
StormShadow308's picture
feat: async pipeline, job queue, generation hardening, and docs
732b14f
Raw
History Blame Contribute Delete
3.99 kB
"""Abstract base class for LangChain-backed vector store backends.
In the LangChain experiment branch the vector store interface changes:
- ``add_documents()`` replaces ``add(chunks, vectors)`` β€” the store handles
embedding internally via the injected LangChain ``Embeddings`` object.
- ``search()`` now takes a **text query** instead of a pre-computed vector β€”
again the store embeds internally.
This enables a much cleaner pipeline:
file β†’ loader β†’ splitter β†’ ``add_documents()`` β†’ done
query β†’ ``search(text)`` β†’ SearchResult list β†’ reranker β†’ LLM
"""
from abc import ABC, abstractmethod
from langchain_core.documents import Document
from app.models.schemas import SearchResult
class VectorStore(ABC):
"""Interface for a LangChain-backed pooled vector store.
All implementations must enforce ``tenant_id`` isolation: a search call
for tenant A must **never** return chunks belonging to tenant B.
The embedding step is delegated to the LangChain ``Embeddings`` object
injected at construction time β€” callers never compute vectors manually.
Example::
from app.vectorstore.factory import get_vectorstore
vs = get_vectorstore()
vs.add_documents(docs)
results = vs.search("Victorian terrace", tenant_id="t-abc", k=10)
"""
@abstractmethod
def add_documents(self, documents: list[Document]) -> None:
"""Embed and insert ``documents`` into the store.
Each document must carry ``chunk_id``, ``doc_id``, and ``tenant_id``
fields in its ``metadata`` dict.
Args:
documents: LangChain :class:`Document` objects with metadata.
"""
...
@abstractmethod
def search(
self,
query: str,
tenant_id: str,
k: int = 10,
*,
hierarchy_level: str | None = None,
doc_id_in: frozenset[str] | None = None,
) -> list[SearchResult]:
"""Retrieve top-k chunks for ``tenant_id`` nearest to ``query``.
The store embeds the query internally β€” no pre-computed vector needed.
Args:
query: Plain-text search query.
tenant_id: Only return chunks belonging to this tenant.
k: Maximum number of results.
hierarchy_level: When set, only rows with this ``hierarchy_level`` metadata.
doc_id_in: When set, only rows whose ``doc_id`` is in this set.
Returns:
List of :class:`~app.models.schemas.SearchResult` ordered by
descending similarity score.
"""
...
@abstractmethod
def delete_document(self, doc_id: str) -> None:
"""Remove all chunks belonging to ``doc_id`` from the store.
Args:
doc_id: Document identifier whose chunks should be deleted.
"""
...
@abstractmethod
def count(self, tenant_id: str) -> int:
"""Return total chunk count for ``tenant_id``.
Args:
tenant_id: Tenant identifier to count.
Returns:
Integer chunk count.
"""
...
async def search_async(
self,
query: str,
tenant_id: str,
k: int = 10,
*,
hierarchy_level: str | None = None,
doc_id_in: frozenset[str] | None = None,
) -> list[SearchResult]:
"""Async search; default runs sync ``search`` in a thread pool."""
from app.async_executor import run_sync_in_executor
return await run_sync_in_executor(
self.search,
query,
tenant_id,
k=k,
hierarchy_level=hierarchy_level,
doc_id_in=doc_id_in,
)
@abstractmethod
def count_for_doc(self, doc_id: str) -> int:
"""Return chunk count for a specific document.
Args:
doc_id: Document identifier to count.
Returns:
Integer chunk count for that document only.
"""
...