windlass.providers.vectordb.faiss

faiss

FAISS vector store.

FAISS gives you approximate nearest-neighbour search over millions of vectors on a single machine, with no server to run. Windlass keeps chunk payloads in a side-table (FAISS stores vectors only) and maps between FAISS's integer ids and Windlass chunk ids.

Install with::

pip install "windlass[faiss]"

Example

from windlass import Windlass # doctest: +SKIP store = Windlass.vectordb("faiss", dimensions=384, # doctest: +SKIP ... persist_path="./index") # doctest: +SKIP

FaissVectorStore

FaissVectorStore(
    *,
    collection: str = "windlass",
    dimensions: int | None = None,
    metric: str = "cosine",
    index_type: str = "flat",
    nlist: int = 100,
    nprobe: int = 10,
    persist_path: str | None = None,
    **config: Any
)

Bases: VectorStore

Vector store backed by a FAISS index.

Parameters:

Name	Type	Description	Default
collection	str	Logical name, used for the persistence filenames.	'windlass'
dimensions	int | None	Vector length. Required — FAISS allocates the index up front and cannot infer it.	None
metric	str	cosine (inner product over normalised vectors), dot (raw inner product) or euclidean.	'cosine'
index_type	str	"flat" for exact search, "ivf" for a coarse-quantised index (much faster on large corpora, needs training), or "hnsw" for a graph index (fast, higher memory, no training).	'flat'
nlist	int	Number of IVF clusters. Ignored for other index types.	100
nprobe	int	How many IVF clusters to scan at query time. Higher means better recall and slower search.	10
persist_path	str | None	Directory to save the index and payload side-table to.	None
**config	Any	Forwarded to :class:~windlass.interfaces.vectordb.VectorStore.	{}

Raises:

Type	Description
MissingDependencyError	When faiss-cpu is not installed.
ConfigurationError	When dimensions is omitted.

Performance

flat is exact and fast to about a million vectors. ivf needs at least 39 * nlist vectors to train and is trained automatically on the first write that has enough data. Searching an untrained IVF index falls back to brute force rather than failing.

Source code in src\windlass\providers\vectordb\faiss.py

def __init__(
    self,
    *,
    collection: str = "windlass",
    dimensions: int | None = None,
    metric: str = "cosine",
    index_type: str = "flat",
    nlist: int = 100,
    nprobe: int = 10,
    persist_path: str | None = None,
    **config: Any,
) -> None:
    super().__init__(
        collection=collection,
        dimensions=dimensions,
        metric=metric,
        persist_path=persist_path,
        **config,
    )
    if not dimensions:
        raise ConfigurationError(
            "FAISS needs to know the vector dimensionality up front.",
            hint="Pass dimensions=..., or let the RAG builder infer it from "
            "the embedding model.",
        )
    self._faiss = require("faiss", extra="faiss", feature="The FAISS vector store")
    self._np = require("numpy", extra="faiss", feature="The FAISS vector store")
    self.index_type = index_type
    self.nlist = nlist
    self.nprobe = nprobe
    self._lock = threading.RLock()
    self._payloads: dict[int, Chunk] = {}
    self._by_chunk_id: dict[str, int] = {}
    self._next_id = 0
    self._index = self._build_index()

    if persist_path and self._index_file().exists():
        self.load(persist_path)

native

native() -> Any

Return the underlying faiss.Index (Level 3 access).

Source code in src\windlass\providers\vectordb\faiss.py

def native(self) -> Any:
    """Return the underlying ``faiss.Index`` (Level 3 access)."""
    return self._index

aadd async

aadd(chunks: Sequence[Chunk]) -> int

Insert or update chunks.

Parameters:

Name	Type	Description	Default
chunks	Sequence[Chunk]	Chunks with embeddings set.	required

Returns:

Type	Description
int	How many chunks were written.

Raises:

Type	Description
ConfigurationError	If a chunk has no embedding.
ProviderError	On dimension mismatch or a FAISS failure.

Source code in src\windlass\providers\vectordb\faiss.py

async def aadd(self, chunks: Sequence[Chunk]) -> int:
    """Insert or update chunks.

    Args:
        chunks: Chunks with embeddings set.

    Returns:
        How many chunks were written.

    Raises:
        ConfigurationError: If a chunk has no embedding.
        ProviderError: On dimension mismatch or a FAISS failure.
    """
    if not chunks:
        return 0
    self.validate_embeddings(chunks)
    return await to_thread(self._add_sync, list(chunks))

adelete async

adelete(
    ids: Sequence[str] | None = None, *, filters: MetadataFilter | None = None
) -> int

Delete by chunk id or metadata filter.

Parameters:

Name	Type	Description	Default
ids	Sequence[str] | None	Chunk ids to remove.	None
filters	MetadataFilter | None	Metadata constraints.	None

Returns:

Type	Description
int	How many chunks were removed.

Raises:

Type	Description
ValueError	When neither argument is supplied.

Source code in src\windlass\providers\vectordb\faiss.py

async def adelete(
    self, ids: Sequence[str] | None = None, *, filters: MetadataFilter | None = None
) -> int:
    """Delete by chunk id or metadata filter.

    Args:
        ids: Chunk ids to remove.
        filters: Metadata constraints.

    Returns:
        How many chunks were removed.

    Raises:
        ValueError: When neither argument is supplied.
    """
    if ids is None and filters is None:
        raise ValueError("Pass ids or filters; use clear() to empty the index.")
    return await to_thread(self._delete_sync, list(ids or ()), filters)

aclear async

aclear() -> None

Drop the index and rebuild it empty.

Source code in src\windlass\providers\vectordb\faiss.py

async def aclear(self) -> None:
    """Drop the index and rebuild it empty."""
    with self._lock:
        self._index = self._build_index()
        self._payloads.clear()
        self._by_chunk_id.clear()
        self._next_id = 0

asearch async

asearch(
    vector: Sequence[float],
    k: int = 5,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any
) -> list[ScoredChunk]

Search the index.

Parameters:

Name	Type	Description	Default
vector	Sequence[float]	The query embedding.	required
k	int	How many results to return.	5
filters	MetadataFilter | None	Metadata constraints, applied after the ANN search. Windlass over-fetches by a factor of :data:_FILTER_OVERFETCH so a selective filter still returns k hits.	None
**kwargs	Any	Ignored.	{}

Returns:

Type	Description
list[ScoredChunk]	Ranked hits.

Source code in src\windlass\providers\vectordb\faiss.py

async def asearch(
    self,
    vector: Sequence[float],
    k: int = 5,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any,
) -> list[ScoredChunk]:
    """Search the index.

    Args:
        vector: The query embedding.
        k: How many results to return.
        filters: Metadata constraints, applied after the ANN search. Windlass
            over-fetches by a factor of :data:`_FILTER_OVERFETCH` so a
            selective filter still returns ``k`` hits.
        **kwargs: Ignored.

    Returns:
        Ranked hits.
    """
    return await to_thread(self._search_sync, list(vector), k, filters)

aget async

aget(ids: Sequence[str]) -> list[Chunk]

Fetch chunks by id.

Source code in src\windlass\providers\vectordb\faiss.py

async def aget(self, ids: Sequence[str]) -> list[Chunk]:
    """Fetch chunks by id."""
    with self._lock:
        return [self._payloads[self._by_chunk_id[i]] for i in ids if i in self._by_chunk_id]

acount async

acount() -> int

Return how many vectors the index holds.

Source code in src\windlass\providers\vectordb\faiss.py

async def acount(self) -> int:
    """Return how many vectors the index holds."""
    with self._lock:
        return int(self._index.ntotal)

all_chunks

all_chunks() -> list[Chunk]

Return every stored chunk, for building a companion lexical index.

Source code in src\windlass\providers\vectordb\faiss.py

def all_chunks(self) -> list[Chunk]:
    """Return every stored chunk, for building a companion lexical index."""
    with self._lock:
        return list(self._payloads.values())

apersist async

apersist() -> None

Save the index and payloads to :attr:persist_path.

Source code in src\windlass\providers\vectordb\faiss.py

async def apersist(self) -> None:
    """Save the index and payloads to :attr:`persist_path`."""
    if not self.persist_path:
        raise ProviderError(
            "This store has no persist_path configured.",
            provider="faiss",
            hint="Construct it with persist_path='./index'.",
        )
    await to_thread(self.save, self.persist_path)

save

save(directory: str | Path) -> Path

Write the FAISS index and its payload side-table to directory.

Parameters:

Name	Type	Description	Default
directory	str | Path	Destination directory, created if absent.	required

Returns:

Type	Description
Path	The directory written to.

Raises:

Type	Description
ProviderError	When the write fails.

Source code in src\windlass\providers\vectordb\faiss.py

def save(self, directory: str | Path) -> Path:
    """Write the FAISS index and its payload side-table to ``directory``.

    Args:
        directory: Destination directory, created if absent.

    Returns:
        The directory written to.

    Raises:
        ProviderError: When the write fails.
    """
    target = Path(directory)
    target.mkdir(parents=True, exist_ok=True)
    with self._lock:
        try:
            self._faiss.write_index(self._index, str(target / f"{self.collection}.faiss"))
            payload = {
                "version": 1,
                "dimensions": self.dimensions,
                "metric": self.metric,
                "index_type": self.index_type,
                "next_id": self._next_id,
                "payloads": {
                    str(fid): chunk.model_dump() for fid, chunk in self._payloads.items()
                },
            }
            (target / f"{self.collection}.meta.json").write_text(
                json.dumps(payload, ensure_ascii=False), encoding="utf-8"
            )
        except Exception as exc:
            raise ProviderError(
                f"Could not save the FAISS index to {target}: {exc}",
                provider="faiss",
                original=exc,
            ) from exc
    return target

load

load(directory: str | Path) -> FaissVectorStore

Load an index previously written by :meth:save.

Parameters:

Name	Type	Description	Default
directory	str | Path	Directory containing the index files.	required

Returns:

Type	Description
FaissVectorStore	self.

Raises:

Type	Description
ProviderError	When the files are missing or unreadable.

Source code in src\windlass\providers\vectordb\faiss.py

def load(self, directory: str | Path) -> FaissVectorStore:
    """Load an index previously written by :meth:`save`.

    Args:
        directory: Directory containing the index files.

    Returns:
        ``self``.

    Raises:
        ProviderError: When the files are missing or unreadable.
    """
    target = Path(directory)
    index_file = target / f"{self.collection}.faiss"
    meta_file = target / f"{self.collection}.meta.json"
    if not index_file.is_file() or not meta_file.is_file():
        raise ProviderError(
            f"No saved FAISS index for {self.collection!r} in {target}.", provider="faiss"
        )
    try:
        with self._lock:
            self._index = self._faiss.read_index(str(index_file))
            meta = json.loads(meta_file.read_text(encoding="utf-8"))
            self._payloads = {
                int(fid): Chunk.model_validate(data)
                for fid, data in meta.get("payloads", {}).items()
            }
            self._by_chunk_id = {c.id: fid for fid, c in self._payloads.items()}
            self._next_id = int(meta.get("next_id", len(self._payloads)))
            self.dimensions = meta.get("dimensions", self.dimensions)
            self.metric = meta.get("metric", self.metric)
    except Exception as exc:
        raise ProviderError(
            f"Could not load the FAISS index from {target}: {exc}",
            provider="faiss",
            original=exc,
        ) from exc
    return self