ZerdaZerda

PineconeVectorDatabase

On this page

Table of contents

Overview

A wrapper around the managed Pinecone service. Unlike the FAISS and Chroma backends, the source of truth for vectors lives remotely; this class additionally maintains an optional local cache (self.chunks, gated by cache_chunks) purely to serve get_chunk_by_id/get_chunks_by_doc_id without a network round trip, and to speed up delete_by_doc_id.

Class attributes

Python · 12 lines
SUPPORTED_METRICS = {'cosine', 'euclidean', 'l2', 'dotproduct', 'dot'}
METRIC_MAPPING = {
    'cosine': 'cosine', 'euclidean': 'euclidean', 'l2': 'euclidean',
    'dotproduct': 'dotproduct', 'dot': 'dotproduct', 'ip': 'dotproduct',
}
UPSERT_BATCH_SIZE = 100
DELETE_BATCH_SIZE = 1000
FETCH_BATCH_SIZE = 1000        # defined but not referenced by any method body in this file
DEFAULT_ENVIRONMENT = "us-east-1"
DEFAULT_CLOUD = "aws"
DEFAULT_TOP_K = 5
DEFAULT_INDEX_READY_TIMEOUT = 300

Note 'ip' is accepted by METRIC_MAPPING (mapped to 'dotproduct') but is not present in SUPPORTED_METRICS, so PineconeVectorDatabase(distance_metric='ip') raises ValueError from _validate_metric before METRIC_MAPPING would ever be consulted with 'ip'. Also note FETCH_BATCH_SIZE is declared but never used to chunk any request in this file — fetch_by_ids issues a single index.fetch() call regardless of list size.

Constructor

Python · 12 lines
PineconeVectorDatabase(
    embedder: Optional[Any] = None,
    index_name: str = "default-index",
    embedding_dim: Optional[int] = None,
    api_key: Optional[str] = None,
    environment: str = "us-east-1",
    distance_metric: str = "cosine",
    cloud: str = "aws",
    pod_type: Optional[str] = None,
    namespace: Optional[str] = None,
    cache_chunks: bool = True,
)
ParameterTypeDefaultNotes
embedderOptional[Any]None.encode(texts) -> np.ndarray interface.
index_namestr"default-index"Created if it doesn't already exist.
embedding_dimOptional[int]NoneRequired if embedder is None.
api_keyOptional[str]NoneFalls back to PINECONE_API_KEY env var; ValueError if neither is set.
environmentstr"us-east-1"Region for serverless, or environment for pod-based.
distance_metricstr"cosine"Must be in SUPPORTED_METRICS; mapped via METRIC_MAPPING to Pinecone's native metric name.
cloudstr"aws"Passed to ServerlessSpec when creating a new serverless index.
pod_typeOptional[str]NoneIf set, creates a pod-based index (PodSpec) instead of serverless.
namespaceOptional[str]None → ""Default namespace for operations that don't override it.
cache_chunksboolTrueEnables the local DocumentChunk cache.

Raises: ImportError if pinecone isn't installed; ValueError for bad metric / missing dimension / missing API key; ConnectionError if Pinecone client/index initialization fails (wraps the underlying exception).

Initialization workflow

  1. super().__init__(embedder).
  2. Validate metric, resolve embedding_dim, map to pinecone_metric, resolve API key.
  3. _initialize_pinecone():
    • Create the Pinecone client.
    • List existing indexes; if index_name isn't among them, _create_index() (chooses PodSpec or ServerlessSpec based on pod_type, then calls _wait_for_index_ready(timeout=DEFAULT_INDEX_READY_TIMEOUT), polling once per second).
    • If it already exists, _verify_index_config() — checks the existing index's dimension matches (ValueError if not) and warns (does not raise) on a metric mismatch.
    • Get an Index handle via self._pc.Index(index_name).

[!WARNING] New serverless/pod index creation blocks for up to 5 minutes _wait_for_index_ready polls describe_index(...).status once per second and raises TimeoutError after DEFAULT_INDEX_READY_TIMEOUT = 300 seconds if the index never reports ready. Constructing PineconeVectorDatabase against a brand-new index name is therefore a blocking, potentially multi-minute call, not merely a lightweight client instantiation.

Thread safety

Not addressed in source. The local cache (self.chunks, the two lookup dicts) is mutated without locking in add, delete_by_ids, delete_by_doc_id, and delete_all.

Property: index

Python · 2 lines
@property
def index(self):

Returns self._index, raising RuntimeError if it's None (i.e., before/if initialization never completed).

Add operations

add_chunk

Python
def add_chunk(self, chunk: DocumentChunk) -> None

Computes chunk.embedding via self.embedder.encode([chunk.text])[0] if missing (raising ValueError if there's no embedder and no embedding), then delegates to add(chunks=[chunk], embeddings=...).

add

Python · 6 lines
def add(
    self,
    chunks: List[DocumentChunk],
    embeddings: Optional[np.ndarray] = None,
    namespace: Optional[str] = None,
) -> None

Workflow:

  1. Empty chunks → warn and return.
  2. _validate_chunks(chunks) — raises ValueError if any item isn't a DocumentChunk, or has an empty/non-string chunk_id/doc_id.
  3. Compute embeddings via the embedder if not supplied (ValueError if no embedder; RuntimeError if the embedder itself fails).
  4. Cast to float32, validate dimension, validate len(chunks) == len(embeddings).
  5. Resolve target_namespace = namespace if namespace is not None else self.namespace.
  6. Build a Pinecone vector dict per chunk: {'id': chunk_id, 'values': [...], 'metadata': {...}} where metadata always includes chunk_id, doc_id, text, plus sanitize_metadata(chunk.metadata) merged in if present.
  7. If cache_chunks, update or insert into the local cache and lookup dicts.
  8. Upsert to Pinecone in batches of UPSERT_BATCH_SIZE = 100 via self.index.upsert(vectors=batch, namespace=target_namespace).

Raises: ValueError for invalid input; RuntimeError wrapping upsert failures.

[!NOTE] Metadata always stores full chunk text in Pinecone Because text is embedded directly into every vector's metadata (not just an ID reference), large chunk texts count against Pinecone's per-vector metadata size limits. This wrapper does not truncate or externalize text — that's a constraint of how the source builds the metadata dict, not an omission in this documentation.

Search operations

Python · 10 lines
def search(
    self,
    query: Union[str, np.ndarray],
    top_k: int = 5,
    score_threshold: Optional[float] = None,
    filter_dict: Optional[Dict[str, Any]] = None,
    namespace: Optional[str] = None,
    include_metadata: bool = True,
    include_values: bool = False,
) -> List[Tuple[DocumentChunk, float]]
ParameterTypeDefaultNotes
querystr | np.ndarray—Text requires self.embedder.
top_kint5 (DEFAULT_TOP_K)Must be > 0.
score_thresholdOptional[float]NoneApplied client-side after the query returns.
filter_dictOptional[Dict]NonePassed straight through as Pinecone's filter expression.
namespaceOptional[str]None → self.namespaceOverrides the instance default for this call only.
include_metadataboolTruePassed to Pinecone's query.
include_valuesboolFalseIf True, embedding vectors are attached to returned chunks.

Workflow: validate top_k; resolve/validate the query embedding's dimension against self.embedding_dim; call self.index.query(vector=..., top_k=..., filter=filter_dict, namespace=..., include_metadata=..., include_values=...); for each match, apply score_threshold, reconstruct a DocumentChunk from match['metadata'] (falling back to the Pinecone match ID for chunk_id and '' for doc_id if absent from metadata), and attach .embedding if include_values was requested and present.

Raises: ValueError for bad top_k / missing embedder / dimension mismatch; RuntimeError wrapping query failures.

search_by_id

Python
def search_by_id(self, chunk_id: str, top_k: int = 5, exclude_self: bool = True, namespace: Optional[str] = None) -> List[Tuple[DocumentChunk, float]]

Fetches the named vector's embedding from Pinecone, then calls search() using that embedding as the query (requesting top_k + 1 when excluding self, then filtering the self-match out of the results). Returns [] with a warning if chunk_id isn't found on the index.

Delete operations

delete_by_ids

Python
def delete_by_ids(self, chunk_ids: List[str], namespace: Optional[str] = None) -> int

Empty list → warn, return 0. Otherwise deletes in batches of DELETE_BATCH_SIZE = 1000, updates the local cache (marks cached entries as None in place rather than compacting the list, to keep existing indices valid), and returns len(chunk_ids) (the requested count, not a confirmed-deleted count from Pinecone — Pinecone's delete API doesn't return one).

delete_by_doc_id

Python
def delete_by_doc_id(self, doc_id: str, namespace: Optional[str] = None) -> int

If the local cache has entries for doc_id, resolves their chunk_ids and delegates to delete_by_ids (returning that call's count). Otherwise, falls back to a metadata filter delete: self.index.delete(filter={'doc_id': doc_id}, namespace=...), cleans up any matching local-cache entries, and returns 0 — per the inline comment, Pinecone doesn't report a count for filter-based deletes, so the return value in this fallback path is not a reliable count of chunks actually removed.

delete_by_filter

Python
def delete_by_filter(self, filter_dict: Dict[str, Any], namespace: Optional[str] = None) -> None

Empty filter → warn, return. Otherwise self.index.delete(filter=filter_dict, namespace=...). Returns None — note this method's return type is None, unlike the other delete methods on this class, and it does not touch the local cache at all (any cached chunks matching the filter remain in self.chunks even after the remote delete succeeds).

delete_all

Python
def delete_all(self, namespace: Optional[str] = None) -> None

self.index.delete(delete_all=True, namespace=...). If cache_chunks, clears the entire local cache regardless of namespace (i.e., even if you only deleted one namespace remotely, the local cache — which isn't namespace-partitioned — is wiped completely). Cannot be undone (per the source's own docstring warning).

Fetch operations

fetch_by_ids

Python
def fetch_by_ids(self, chunk_ids: List[str], namespace: Optional[str] = None) -> List[DocumentChunk]

Empty list → returns []. Otherwise a single self.index.fetch(ids=chunk_ids, namespace=...) call (see the FETCH_BATCH_SIZE note above — large chunk_ids lists are not chunked here), reconstructing a DocumentChunk per returned vector (with .embedding set from values if present).

Raises: RuntimeError wrapping fetch failures.

Statistics & utilities

get_stats

Python
def get_stats(self, namespace: Optional[str] = None) -> Dict[str, Any]
Python · 13 lines
{
    'index_name': str,
    'total_vector_count': int,
    'embedding_dim': int,
    'distance_metric': str,
    'namespace': str,
    'namespace_vector_count': int,
    'namespaces': List[str],
    'has_embedder': bool,
    'cache_enabled': bool,
    'cached_chunks': int,
    'unique_docs_cached': int,   # only present if cache_chunks is True
}

[!NOTE] Failure mode differs from other backends Unlike FAISS/Chroma's get_stats, this method catches all exceptions and returns {'error': str(e)} instead of propagating. Code that expects get_stats() to always contain keys like total_vector_count must account for this alternate shape.

list_namespaces

Python
def list_namespaces(self) -> List[str]

Returns [] (not a raised exception) if describe_index_stats() fails.

get_chunk_by_id / get_chunks_by_doc_id

Local-cache-only lookups; return None/[] respectively if cache_chunks is False or the ID/doc isn't cached — these do not query Pinecone remotely, unlike fetch_by_ids.

Dunder methods

Python · 3 lines
def __len__(self) -> int          # index.describe_index_stats().total_vector_count, or 0 on any exception
def __repr__(self) -> str         # falls back to "unknown" count on exception
def __contains__(self, chunk_id) -> bool   # local cache only, and only if cache_chunks is True

__contains__ checks the local cache, not the remote index — chunk_id in db can be False even for a chunk_id that genuinely exists in Pinecone, if it was added by a different process/session or if cache_chunks=False.

Unified interface aliases

Python · 6 lines
def remove_by_doc_id(self, doc_id: str) -> int:
    return self.delete_by_doc_id(doc_id)

def clear(self) -> int:
    self.delete_all()
    return 0   # Pinecone doesn't return a count for full deletes

clear() always returns 0 regardless of how many vectors actually existed — this is a fixed return value in the source, not a computed count.

Context manager support

batch_operation

Python · 2 lines
@contextmanager
def batch_operation(self, namespace: Optional[str] = None)

Temporarily overrides self.namespace for the duration of the with block (restoring the original value in a finally), logging start/success/failure. Like Chroma's equivalent, it does not defer or buffer writes — it's a namespace-scoping and logging convenience, not a performance batching mechanism.

Async API

Python · 3 lines
async def asearch(self, query, top_k=10, **kwargs)
async def aadd(self, chunks, **kwargs)
async def __aenter__(self) / async def __aexit__(self, *_)

Same asyncio.to_thread(...) pattern as the other two backends.

Common mistakes

  • Assuming delete_by_doc_id's return value is always an accurate count — it only is when the local cache has the relevant entries; the metadata-filter fallback path always returns 0.
  • Relying on get_stats() always exposing total_vector_count/etc. — on any internal failure it instead returns {'error': ...}.
  • Using chunk_id in db (__contains__) as a proxy for "does this vector exist in Pinecone" — it only reflects the local cache.
  • Passing distance_metric='ip' expecting it to map to dot-product — it's rejected by _validate_metric before the mapping is consulted; use 'dotproduct' or 'dot' instead.
  • Passing large chunk_ids lists to fetch_by_ids expecting automatic batching — there isn't any in this method, despite FETCH_BATCH_SIZE being defined on the class.

Related pages