codekingpro/portable-devtools

1from __future__ import annotations2 3from typing import Iterator, List4from urllib.parse import urlparse5 6from langchain_core._api import deprecated7 8from langchain_community.embeddings.mlflow import MlflowEmbeddings9 10 11def _chunk(texts: List[str], size: int) -> Iterator[List[str]]:12    for i in range(0, len(texts), size):13        yield texts[i : i + size]14 15 16@deprecated(17    since="0.3.3",18    removal="1.0",19    alternative_import="databricks_langchain.DatabricksEmbeddings",20)21class DatabricksEmbeddings(MlflowEmbeddings):22    """Databricks embeddings.23 24    To use, you should have the ``mlflow`` python package installed.25    For more information, see https://mlflow.org/docs/latest/llms/deployments.26 27    Example:28        .. code-block:: python29 30            from langchain_community.embeddings import DatabricksEmbeddings31 32            embeddings = DatabricksEmbeddings(33                target_uri="databricks",34                endpoint="embeddings",35            )36    """37 38    target_uri: str = "databricks"39    """The target URI to use. Defaults to ``databricks``."""40 41    @property42    def _mlflow_extras(self) -> str:43        return ""44 45    def _validate_uri(self) -> None:46        if self.target_uri == "databricks":47            return48 49        if urlparse(self.target_uri).scheme != "databricks":50            raise ValueError(51                "Invalid target URI. The target URI must be a valid databricks URI."52            )53