Retrievers

Bases: Protocol

Returns ranked RetrievalResults for a query. namespace is mandatory.

Source code in src/cenote/retrievers/base.py

class Retriever(Protocol):
    """Returns ranked RetrievalResults for a query. `namespace` is mandatory."""

    async def retrieve(
        self,
        query: str,
        namespace: str,
        limit: int = 10,
        filter: dict[str, Any] | None = None,
    ) -> list[RetrievalResult]: ...

Embed the query, then search the store.

Source code in src/cenote/retrievers/vector.py

class VectorRetriever:
    """Embed the query, then search the store."""

    def __init__(self, embedder: Embedder, store: VectorStore) -> None:
        self._embedder = embedder
        self._store = store

    async def retrieve(
        self,
        query: str,
        namespace: str,
        limit: int = 10,
        filter: dict[str, Any] | None = None,
    ) -> list[RetrievalResult]:
        logger.debug(
            "VectorRetriever: query='%s' namespace=%s limit=%d", query[:60], namespace, limit
        )
        vector = await self._embedder.embed_query(query)
        results = await self._store.search(vector, namespace=namespace, limit=limit, filter=filter)
        logger.debug("VectorRetriever: returned %d results", len(results))
        return [RetrievalResult(chunk=r.chunk, score=r.score, retriever="vector") for r in results]