Source code for langchain_pinecone.embeddings

import logging
from typing import Any, Dict, Iterable, List, Optional

from langchain_core.embeddings import Embeddings
from langchain_core.utils import secret_from_env
from pinecone import Pinecone as PineconeClient  # type: ignore[import-untyped]
from pinecone import (
    PineconeAsyncio as PineconeAsyncioClient,  # type: ignore[import-untyped]
)
from pinecone import SparseValues

# Conditional import for EmbeddingsList based on Pinecone version
try:
    from pinecone.core.openapi.inference.model.embeddings_list import EmbeddingsList
except ImportError:
    # Fallback for pinecone versions < 7.0.0
    from pinecone.data.features.inference.inference import EmbeddingsList

from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    PrivateAttr,
    SecretStr,
    model_validator,
)
from typing_extensions import Self

logger = logging.getLogger(__name__)

DEFAULT_BATCH_SIZE = 64



[docs]
class PineconeEmbeddings(BaseModel, Embeddings):
    """PineconeEmbeddings embedding model.

    Example:
        .. code-block:: python

            from langchain_pinecone import PineconeEmbeddings
            from langchain_pinecone import PineconeVectorStore
            from langchain_core.documents import Document

            # Initialize embeddings with a specific model
            embeddings = PineconeEmbeddings(model="multilingual-e5-large")

            # Embed a single query
            query_embedding = embeddings.embed_query("What is machine learning?")

            # Embed multiple documents
            docs = ["Document 1 content", "Document 2 content"]
            doc_embeddings = embeddings.embed_documents(docs)

            # Use with PineconeVectorStore
            from pinecone import Pinecone

            pc = Pinecone(api_key="your-api-key")
            index = pc.Index("your-index-name")

            vectorstore = PineconeVectorStore(
                index=index,
                embedding=embeddings
            )

            # Add documents to vector store
            vectorstore.add_documents([
                Document(page_content="Hello, world!"),
                Document(page_content="This is a test.")
            ])

            # Search for similar documents
            results = vectorstore.similarity_search("hello", k=2)
    """

    # Clients
    _client: PineconeClient = PrivateAttr(default=None)
    _async_client: Optional[PineconeAsyncioClient] = PrivateAttr(default=None)
    model: str
    """Model to use for example 'multilingual-e5-large'."""
    # Config
    batch_size: Optional[int] = None
    """Batch size for embedding documents."""
    query_params: Dict = Field(default_factory=dict)
    """Parameters for embedding query."""
    document_params: Dict = Field(default_factory=dict)
    """Parameters for embedding document"""
    #
    dimension: Optional[int] = None
    #
    show_progress_bar: bool = False
    pinecone_api_key: SecretStr = Field(
        default_factory=secret_from_env(
            "PINECONE_API_KEY",
            error_message="Pinecone API key not found. Please set the PINECONE_API_KEY "
            "environment variable or pass it via `pinecone_api_key`.",
        ),
        alias="api_key",
    )
    """Pinecone API key. 
    
    If not provided, will look for the PINECONE_API_KEY environment variable."""

    model_config = ConfigDict(
        extra="forbid",
        populate_by_name=True,
        protected_namespaces=(),
    )

    @property
    def async_client(self) -> PineconeAsyncioClient:
        """Lazily initialize the async client."""
        return PineconeAsyncioClient(
            api_key=self.pinecone_api_key.get_secret_value(), source_tag="langchain"
        )

    @model_validator(mode="before")
    @classmethod
    def set_default_config(cls, values: dict) -> Any:
        """Set default configuration based on model."""
        default_config_map = {
            "multilingual-e5-large": {
                "batch_size": 96,
                "query_params": {"input_type": "query", "truncate": "END"},
                "document_params": {"input_type": "passage", "truncate": "END"},
                "dimension": 1024,
            },
        }
        model = values.get("model")
        if model in default_config_map:
            config = default_config_map[model]
            for key, value in config.items():
                if key not in values:
                    values[key] = value
        return values

    @model_validator(mode="after")
    def validate_environment(self) -> Self:
        """Validate that Pinecone version and credentials exist in environment."""
        api_key_str = self.pinecone_api_key.get_secret_value()
        client = PineconeClient(api_key=api_key_str, source_tag="langchain")
        self._client = client

        # Ensure async_client is lazily initialized
        return self

    def _get_batch_iterator(self, texts: List[str]) -> tuple[Iterable, int]:
        if self.batch_size is None:
            batch_size = DEFAULT_BATCH_SIZE
        else:
            batch_size = self.batch_size

        if self.show_progress_bar:
            try:
                from tqdm.auto import tqdm  # type: ignore
            except ImportError as e:
                raise ImportError(
                    "Must have tqdm installed if `show_progress_bar` is set to True. "
                    "Please install with `pip install tqdm`."
                ) from e

            _iter = tqdm(range(0, len(texts), batch_size))
        else:
            _iter = range(0, len(texts), batch_size)

        return _iter, batch_size


[docs]
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """Embed search docs."""
        embeddings: List[List[float]] = []

        _iter, batch_size = self._get_batch_iterator(texts)
        for i in _iter:
            response = self._embed_texts(
                model=self.model,
                parameters=self.document_params,
                texts=texts[i : i + batch_size],
            )
            embeddings.extend([r["values"] for r in response])

        return embeddings



[docs]
    async def aembed_documents(self, texts: List[str]) -> List[List[float]]:
        embeddings: List[List[float]] = []
        _iter, batch_size = self._get_batch_iterator(texts)
        for i in _iter:
            response = await self._aembed_texts(
                model=self.model,
                parameters=self.document_params,
                texts=texts[i : i + batch_size],
            )
            embeddings.extend([r["values"] for r in response])
        return embeddings



[docs]
    def embed_query(self, text: str) -> List[float]:
        """Embed query text."""
        return self._embed_texts(
            model=self.model, parameters=self.query_params, texts=[text]
        )[0]["values"]



[docs]
    async def aembed_query(self, text: str) -> List[float]:
        """Asynchronously embed query text."""
        embeddings = await self._aembed_texts(
            model=self.model,
            parameters=self.document_params,
            texts=[text],
        )
        return embeddings[0]["values"]


    def _embed_texts(
        self, texts: List[str], model: str, parameters: dict
    ) -> EmbeddingsList:
        return self._client.inference.embed(
            model=model, inputs=texts, parameters=parameters
        )

    async def _aembed_texts(
        self, texts: List[str], model: str, parameters: dict
    ) -> EmbeddingsList:
        async with self.async_client as aclient:
            embeddings: EmbeddingsList = await aclient.inference.embed(
                model=model, inputs=texts, parameters=parameters
            )
            return embeddings




[docs]
class PineconeSparseEmbeddings(PineconeEmbeddings):
    """PineconeSparseEmbeddings embedding model.

    Example:
        .. code-block:: python

            from langchain_pinecone import PineconeSparseEmbeddings
            from langchain_pinecone import PineconeVectorStore
            from langchain_core.documents import Document

            # Initialize sparse embeddings
            sparse_embeddings = PineconeSparseEmbeddings(model="pinecone-sparse-english-v0")

            # Embed a single query (returns SparseValues)
            query_embedding = sparse_embeddings.embed_query("What is machine learning?")
            # query_embedding contains SparseValues with indices and values

            # Embed multiple documents
            docs = ["Document 1 content", "Document 2 content"]
            doc_embeddings = sparse_embeddings.embed_documents(docs)

            # Use with an index configured for sparse vectors
            from pinecone import Pinecone

            pc = Pinecone(api_key="your-api-key")

            # Create index with sparse embeddings support
            if not pc.has_index("sparse-index"):
                pc.create_index_for_model(
                    name="sparse-index",
                    cloud="aws",
                    region="us-east-1",
                    embed={
                        "model": "pinecone-sparse-english-v0",
                        "field_map": {"text": "chunk_text"},
                        "metric": "dotproduct",
                        "read_parameters": {},
                        "write_parameters": {}
                    }
                )

            index = pc.Index("sparse-index")

            # IMPORTANT: Use PineconeSparseVectorStore for sparse vectors
            # The regular PineconeVectorStore won't work with sparse embeddings
            from langchain_pinecone.vectorstores_sparse import PineconeSparseVectorStore

            # Initialize sparse vector store with sparse embeddings
            vector_store = PineconeSparseVectorStore(
                index=index,
                embedding=sparse_embeddings
            )

            # Add documents
            from uuid import uuid4

            documents = [
                Document(page_content="Machine learning is awesome", metadata={"source": "article"}),
                Document(page_content="Neural networks power modern AI", metadata={"source": "book"})
            ]

            # Generate unique IDs for each document
            uuids = [str(uuid4()) for _ in range(len(documents))]

            # Add documents to the vector store
            vector_store.add_documents(documents=documents, ids=uuids)

            # Search for similar documents
            results = vector_store.similarity_search("machine learning", k=2)
    """

    @model_validator(mode="before")
    @classmethod
    def set_default_config(cls, values: dict) -> Any:
        """Set default configuration based on model."""
        default_config_map = {
            "pinecone-sparse-english-v0": {
                "batch_size": 96,
                "query_params": {"input_type": "query", "truncate": "END"},
                "document_params": {"input_type": "passage", "truncate": "END"},
                "dimension": None,
            },
        }
        model = values.get("model")
        if model in default_config_map:
            config = default_config_map[model]
            for key, value in config.items():
                if key not in values:
                    values[key] = value
        return values


[docs]
    def embed_documents(self, texts: List[str]) -> List[SparseValues]:
        """Embed search docs with sparse embeddings."""
        embeddings: List[SparseValues] = []

        _iter, batch_size = self._get_batch_iterator(texts)
        for i in _iter:
            response = self._embed_texts(
                model=self.model,
                parameters=self.document_params,
                texts=texts[i : i + batch_size],
            )
            for r in response:
                embeddings.append(
                    SparseValues(indices=r["sparse_indices"], values=r["sparse_values"])
                )

        return embeddings



[docs]
    async def aembed_documents(self, texts: List[str]) -> List[SparseValues]:
        """Asynchronously embed search docs with sparse embeddings."""
        embeddings: List[SparseValues] = []
        _iter, batch_size = self._get_batch_iterator(texts)
        for i in _iter:
            response = await self._aembed_texts(
                model=self.model,
                parameters=self.document_params,
                texts=texts[i : i + batch_size],
            )
            for r in response:
                embeddings.append(
                    SparseValues(indices=r["sparse_indices"], values=r["sparse_values"])
                )
        return embeddings



[docs]
    def embed_query(self, text: str) -> SparseValues:
        """Embed query text with sparse embeddings."""
        response = self._embed_texts(
            model=self.model, parameters=self.query_params, texts=[text]
        )[0]
        return SparseValues(
            indices=response["sparse_indices"], values=response["sparse_values"]
        )



[docs]
    async def aembed_query(self, text: str) -> SparseValues:
        """Asynchronously embed query text with sparse embeddings."""
        embeddings = await self._aembed_texts(
            model=self.model,
            parameters=self.query_params,
            texts=[text],
        )
        response = embeddings[0]
        return SparseValues(
            indices=response["sparse_indices"], values=response["sparse_values"]
        )