windlass.providers.vectordb.pinecone

pinecone

Pinecone vector store.

Pinecone is the managed option: no index to host, scales past what a single machine holds, and supports namespaces for multi-tenant isolation. The trade-off is network latency on every query and eventual consistency on writes — a chunk you just upserted may not appear in the very next search.

Install with::

pip install "windlass[pinecone]"

Example

from windlass import Windlass # doctest: +SKIP store = Windlass.vectordb("pinecone", collection="docs", # doctest: +SKIP ... dimensions=1536) # doctest: +SKIP

PineconeVectorStore

PineconeVectorStore(
    *,
    collection: str = "windlass",
    dimensions: int | None = None,
    metric: str = "cosine",
    api_key: str | None = None,
    namespace: str = "",
    cloud: str = "aws",
    region: str = "us-east-1",
    create_if_missing: bool = True,
    **config: Any
)

Bases: VectorStore

Vector store backed by a Pinecone serverless index.

Parameters:

Name	Type	Description	Default
collection	str	Pinecone index name.	'windlass'
dimensions	int | None	Vector length. Required when the index must be created.	None
metric	str	cosine, dot or euclidean.	'cosine'
api_key	str | None	Credential. Falls back to PINECONE_API_KEY.	None
namespace	str	Namespace within the index — the clean way to isolate tenants while sharing one index.	''
cloud	str	Serverless cloud provider for index creation.	'aws'
region	str	Serverless region for index creation.	'us-east-1'
create_if_missing	bool	Create the index when it does not exist. Creation is not instant; the first write may wait for it to become ready.	True
**config	Any	Forwarded to :class:~windlass.interfaces.vectordb.VectorStore.	{}

Raises:

Type	Description
MissingDependencyError	When pinecone is not installed.
ConfigurationError	When the API key or dimensions are missing.
ProviderError	When the index cannot be reached or created.

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

def __init__(
    self,
    *,
    collection: str = "windlass",
    dimensions: int | None = None,
    metric: str = "cosine",
    api_key: str | None = None,
    namespace: str = "",
    cloud: str = "aws",
    region: str = "us-east-1",
    create_if_missing: bool = True,
    **config: Any,
) -> None:
    super().__init__(collection=collection, dimensions=dimensions, metric=metric, **config)
    sdk = require("pinecone", extra="pinecone", feature="The Pinecone vector store")
    key = api_key or settings().secret("pinecone_api_key")
    if not key:
        raise ConfigurationError(
            "No API key configured for Pinecone.",
            hint="Set PINECONE_API_KEY, or pass "
            "Windlass.vectordb('pinecone', api_key='...').",
        )
    self.namespace = namespace
    self._sdk = sdk
    try:
        self._client = sdk.Pinecone(api_key=key)
        existing = {idx["name"] for idx in self._client.list_indexes()}
        if collection not in existing:
            if not create_if_missing:
                raise ConfigurationError(
                    f"Pinecone index {collection!r} does not exist.",
                    hint="Create it in the console, or pass create_if_missing=True.",
                )
            if not dimensions:
                raise ConfigurationError(
                    "Creating a Pinecone index requires dimensions=...",
                    hint="Pass the embedding model's dimensionality.",
                )
            self._client.create_index(
                name=collection,
                dimension=int(dimensions),
                metric=_metric_name(metric),
                spec=sdk.ServerlessSpec(cloud=cloud, region=region),
            )
        self._index = self._client.Index(collection)
    except ConfigurationError:
        raise
    except Exception as exc:
        raise ProviderError(
            f"Could not open the Pinecone index {collection!r}: {exc}",
            provider="pinecone",
            original=exc,
        ) from exc

native

native() -> Any

Return the underlying Pinecone Index handle (Level 3 access).

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

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

aadd async

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

Upsert chunks in batches.

Parameters:

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

Returns:

Type	Description
int	How many chunks were sent.

Raises:

Type	Description
ConfigurationError	If a chunk has no embedding.
ProviderError	When the API rejects a batch.

Note

Pinecone is eventually consistent. A chunk written here may take a moment to become visible to :meth:asearch.

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

async def aadd(self, chunks: Sequence[Chunk]) -> int:
    """Upsert chunks in batches.

    Args:
        chunks: Chunks with embeddings set.

    Returns:
        How many chunks were sent.

    Raises:
        ConfigurationError: If a chunk has no embedding.
        ProviderError: When the API rejects a batch.

    Note:
        Pinecone is eventually consistent. A chunk written here may take a
        moment to become visible to :meth:`asearch`.
    """
    if not chunks:
        return 0
    self.validate_embeddings(chunks)

    vectors = [
        {
            "id": chunk.id,
            "values": list(chunk.embedding or []),
            "metadata": _encode_metadata(chunk),
        }
        for chunk in chunks
    ]

    def _upsert() -> int:
        for start in range(0, len(vectors), _UPSERT_BATCH):
            batch = vectors[start : start + _UPSERT_BATCH]
            self._index.upsert(vectors=batch, namespace=self.namespace or None)
        return len(vectors)

    try:
        return await to_thread(_upsert)
    except Exception as exc:
        raise ProviderError(
            f"Pinecone rejected a batch of {len(vectors)} vectors: {exc}",
            provider="pinecone",
            original=exc,
        ) from exc

adelete async

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

Delete by id or metadata filter.

Parameters:

Name	Type	Description	Default
ids	Sequence[str] | None	Chunk ids to remove.	None
filters	MetadataFilter | None	Metadata constraints, translated to Pinecone's filter syntax.	None

Returns:

Type	Description
int	The number of ids requested. Pinecone does not report how many rows
int	a filtered delete actually matched, so filtered deletes return 0.

Raises:

Type	Description
ValueError	When neither argument is supplied.

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

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

    Args:
        ids: Chunk ids to remove.
        filters: Metadata constraints, translated to Pinecone's filter syntax.

    Returns:
        The number of ids requested. Pinecone does not report how many rows
        a filtered delete actually matched, so filtered deletes return 0.

    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 namespace.")

    def _delete() -> int:
        if ids:
            self._index.delete(ids=list(ids), namespace=self.namespace or None)
            return len(list(ids))
        self._index.delete(
            filter=_to_pinecone_filter(filters or {}), namespace=self.namespace or None
        )
        return 0

    try:
        return await to_thread(_delete)
    except Exception as exc:
        raise ProviderError(
            f"Pinecone delete failed: {exc}", provider="pinecone", original=exc
        ) from exc

aclear async

aclear() -> None

Delete every vector in the namespace.

Clearing a namespace that does not exist yet succeeds. Pinecone creates namespaces lazily on first write and returns 404 for a delete-all against one that has never been written to, but "make this empty" is already satisfied in that case — raising would make teardown code fail precisely when there is nothing to tear down.

Raises:

Type	Description
ProviderError	For any failure other than an absent namespace.

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

async def aclear(self) -> None:
    """Delete every vector in the namespace.

    Clearing a namespace that does not exist yet succeeds. Pinecone creates
    namespaces lazily on first write and returns 404 for a delete-all
    against one that has never been written to, but "make this empty" is
    already satisfied in that case — raising would make teardown code fail
    precisely when there is nothing to tear down.

    Raises:
        ProviderError: For any failure other than an absent namespace.
    """

    def _clear() -> None:
        try:
            self._index.delete(delete_all=True, namespace=self.namespace or None)
        except Exception as exc:
            if _is_missing_namespace(exc):
                self._log.debug(
                    "Namespace %r does not exist; nothing to clear.", self.namespace
                )
                return
            raise ProviderError(
                f"Pinecone clear failed: {exc}", provider="pinecone", original=exc
            ) from exc

    await to_thread(_clear)

asearch async

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

Query 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, pushed down to Pinecone.	None
**kwargs	Any	Extra keyword arguments for index.query.	{}

Returns:

Type	Description
list[ScoredChunk]	Ranked hits.

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

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

    Args:
        vector: The query embedding.
        k: How many results to return.
        filters: Metadata constraints, pushed down to Pinecone.
        **kwargs: Extra keyword arguments for ``index.query``.

    Returns:
        Ranked hits.
    """

    def _query() -> Any:
        payload: dict[str, Any] = {
            "vector": list(vector),
            "top_k": max(1, k),
            "include_metadata": True,
            "namespace": self.namespace or None,
        }
        if filters:
            payload["filter"] = _to_pinecone_filter(filters)
        payload.update(kwargs)
        return self._index.query(**payload)

    try:
        response = await to_thread(_query)
    except Exception as exc:
        raise ProviderError(
            f"Pinecone query failed: {exc}", provider="pinecone", original=exc
        ) from exc

    hits = [
        ScoredChunk(
            chunk=_decode_chunk(match["id"], match.get("metadata") or {}),
            score=float(match.get("score", 0.0)),
            retriever=self.name,
        )
        for match in (response.get("matches") or [])
    ]
    return self.rank(hits)

aget async

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

Fetch chunks by id.

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

async def aget(self, ids: Sequence[str]) -> list[Chunk]:
    """Fetch chunks by id."""
    if not ids:
        return []

    def _fetch() -> Any:
        return self._index.fetch(ids=list(ids), namespace=self.namespace or None)

    response = await to_thread(_fetch)
    vectors = response.get("vectors") or {}
    return [
        _decode_chunk(cid, payload.get("metadata") or {}) for cid, payload in vectors.items()
    ]

acount async

acount() -> int

Return the namespace's vector count.

Note

Pinecone's statistics lag writes by a few seconds.

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

async def acount(self) -> int:
    """Return the namespace's vector count.

    Note:
        Pinecone's statistics lag writes by a few seconds.
    """

    def _stats() -> int:
        stats = self._index.describe_index_stats()
        if self.namespace:
            namespaces = stats.get("namespaces") or {}
            return int((namespaces.get(self.namespace) or {}).get("vector_count", 0))
        return int(stats.get("total_vector_count", 0))

    return await to_thread(_stats)