venv/lib/python3.12/site-packages/langchain/embeddings/base.py

import functools
from importlib import util
from typing import Any, List, Optional, Tuple, Union

from langchain_core._api import beta
from langchain_core.embeddings import Embeddings
from langchain_core.runnables import Runnable

_SUPPORTED_PROVIDERS = {
    "azure_openai": "langchain_openai",
    "bedrock": "langchain_aws",
    "cohere": "langchain_cohere",
    "google_vertexai": "langchain_google_vertexai",
    "huggingface": "langchain_huggingface",
    "mistralai": "langchain_mistralai",
    "openai": "langchain_openai",
}


def _get_provider_list() -> str:
    """Get formatted list of providers and their packages."""
    return "\n".join(
        f"  - {p}: {pkg.replace('_', '-')}" for p, pkg in _SUPPORTED_PROVIDERS.items()
    )


def _parse_model_string(model_name: str) -> Tuple[str, str]:
    """Parse a model string into provider and model name components.

    The model string should be in the format 'provider:model-name', where provider
    is one of the supported providers.

    Args:
        model_name: A model string in the format 'provider:model-name'

    Returns:
        A tuple of (provider, model_name)

    .. code-block:: python

        _parse_model_string("openai:text-embedding-3-small")
        # Returns: ("openai", "text-embedding-3-small")

        _parse_model_string("bedrock:amazon.titan-embed-text-v1")
        # Returns: ("bedrock", "amazon.titan-embed-text-v1")

    Raises:
        ValueError: If the model string is not in the correct format or
            the provider is unsupported
    """
    if ":" not in model_name:
        providers = _SUPPORTED_PROVIDERS
        raise ValueError(
            f"Invalid model format '{model_name}'.\n"
            f"Model name must be in format 'provider:model-name'\n"
            f"Example valid model strings:\n"
            f"  - openai:text-embedding-3-small\n"
            f"  - bedrock:amazon.titan-embed-text-v1\n"
            f"  - cohere:embed-english-v3.0\n"
            f"Supported providers: {providers}"
        )

    provider, model = model_name.split(":", 1)
    provider = provider.lower().strip()
    model = model.strip()

    if provider not in _SUPPORTED_PROVIDERS:
        raise ValueError(
            f"Provider '{provider}' is not supported.\n"
            f"Supported providers and their required packages:\n"
            f"{_get_provider_list()}"
        )
    if not model:
        raise ValueError("Model name cannot be empty")
    return provider, model


def _infer_model_and_provider(
    model: str, *, provider: Optional[str] = None
) -> Tuple[str, str]:
    if not model.strip():
        raise ValueError("Model name cannot be empty")
    if provider is None and ":" in model:
        provider, model_name = _parse_model_string(model)
    else:
        provider = provider
        model_name = model

    if not provider:
        providers = _SUPPORTED_PROVIDERS
        raise ValueError(
            "Must specify either:\n"
            "1. A model string in format 'provider:model-name'\n"
            "   Example: 'openai:text-embedding-3-small'\n"
            "2. Or explicitly set provider from: "
            f"{providers}"
        )

    if provider not in _SUPPORTED_PROVIDERS:
        raise ValueError(
            f"Provider '{provider}' is not supported.\n"
            f"Supported providers and their required packages:\n"
            f"{_get_provider_list()}"
        )
    return provider, model_name


@functools.lru_cache(maxsize=len(_SUPPORTED_PROVIDERS))
def _check_pkg(pkg: str) -> None:
    """Check if a package is installed."""
    if not util.find_spec(pkg):
        raise ImportError(
            f"Could not import {pkg} python package. "
            f"Please install it with `pip install {pkg}`"
        )


@beta()
def init_embeddings(
    model: str,
    *,
    provider: Optional[str] = None,
    **kwargs: Any,
) -> Union[Embeddings, Runnable[Any, List[float]]]:
    """Initialize an embeddings model from a model name and optional provider.

    **Note:** Must have the integration package corresponding to the model provider
    installed.

    Args:
        model: Name of the model to use. Can be either:
            - A model string like "openai:text-embedding-3-small"
            - Just the model name if provider is specified
        provider: Optional explicit provider name. If not specified,
            will attempt to parse from the model string. Supported providers
            and their required packages:

            {_get_provider_list()}

        **kwargs: Additional model-specific parameters passed to the embedding model.
            These vary by provider, see the provider-specific documentation for details.

    Returns:
        An Embeddings instance that can generate embeddings for text.

    Raises:
        ValueError: If the model provider is not supported or cannot be determined
        ImportError: If the required provider package is not installed

    .. dropdown:: Example Usage
        :open:

        .. code-block:: python

            # Using a model string
            model = init_embeddings("openai:text-embedding-3-small")
            model.embed_query("Hello, world!")

            # Using explicit provider
            model = init_embeddings(
                model="text-embedding-3-small",
                provider="openai"
            )
            model.embed_documents(["Hello, world!", "Goodbye, world!"])

            # With additional parameters
            model = init_embeddings(
                "openai:text-embedding-3-small",
                api_key="sk-..."
            )

    .. versionadded:: 0.3.9
    """
    if not model:
        providers = _SUPPORTED_PROVIDERS.keys()
        raise ValueError(
            "Must specify model name. "
            f"Supported providers are: {', '.join(providers)}"
        )

    provider, model_name = _infer_model_and_provider(model, provider=provider)
    pkg = _SUPPORTED_PROVIDERS[provider]
    _check_pkg(pkg)

    if provider == "openai":
        from langchain_openai import OpenAIEmbeddings

        return OpenAIEmbeddings(model=model_name, **kwargs)
    elif provider == "azure_openai":
        from langchain_openai import AzureOpenAIEmbeddings

        return AzureOpenAIEmbeddings(model=model_name, **kwargs)
    elif provider == "google_vertexai":
        from langchain_google_vertexai import VertexAIEmbeddings

        return VertexAIEmbeddings(model=model_name, **kwargs)
    elif provider == "bedrock":
        from langchain_aws import BedrockEmbeddings

        return BedrockEmbeddings(model_id=model_name, **kwargs)
    elif provider == "cohere":
        from langchain_cohere import CohereEmbeddings

        return CohereEmbeddings(model=model_name, **kwargs)
    elif provider == "mistralai":
        from langchain_mistralai import MistralAIEmbeddings

        return MistralAIEmbeddings(model=model_name, **kwargs)
    elif provider == "huggingface":
        from langchain_huggingface import HuggingFaceEmbeddings

        return HuggingFaceEmbeddings(model_name=model_name, **kwargs)
    else:
        raise ValueError(
            f"Provider '{provider}' is not supported.\n"
            f"Supported providers and their required packages:\n"
            f"{_get_provider_list()}"
        )


__all__ = [
    "init_embeddings",
    "Embeddings",  # This one is for backwards compatibility
]