codekingpro/portable-devtools

1import warnings2from typing import Any, Dict, List, Optional3 4import requests5from langchain_core._api import deprecated, warn_deprecated6from langchain_core.embeddings import Embeddings7from pydantic import BaseModel, ConfigDict, Field, SecretStr8 9DEFAULT_MODEL_NAME = "sentence-transformers/all-mpnet-base-v2"10DEFAULT_INSTRUCT_MODEL = "hkunlp/instructor-large"11DEFAULT_BGE_MODEL = "BAAI/bge-large-en"12DEFAULT_EMBED_INSTRUCTION = "Represent the document for retrieval: "13DEFAULT_QUERY_INSTRUCTION = (14    "Represent the question for retrieving supporting documents: "15)16DEFAULT_QUERY_BGE_INSTRUCTION_EN = (17    "Represent this question for searching relevant passages: "18)19DEFAULT_QUERY_BGE_INSTRUCTION_ZH = "为这个句子生成表示以用于检索相关文章："20 21 22@deprecated(23    since="0.2.2",24    removal="1.0",25    alternative_import="langchain_huggingface.HuggingFaceEmbeddings",26)27class HuggingFaceEmbeddings(BaseModel, Embeddings):28    """HuggingFace sentence_transformers embedding models.29 30    To use, you should have the ``sentence_transformers`` python package installed.31 32    Example:33        .. code-block:: python34 35            from langchain_community.embeddings import HuggingFaceEmbeddings36 37            model_name = "sentence-transformers/all-mpnet-base-v2"38            model_kwargs = {'device': 'cpu'}39            encode_kwargs = {'normalize_embeddings': False}40            hf = HuggingFaceEmbeddings(41                model_name=model_name,42                model_kwargs=model_kwargs,43                encode_kwargs=encode_kwargs44            )45    """46 47    client: Any = None  #: :meta private:48    model_name: str = DEFAULT_MODEL_NAME49    """Model name to use."""50    cache_folder: Optional[str] = None51    """Path to store models. 52    Can be also set by SENTENCE_TRANSFORMERS_HOME environment variable."""53    model_kwargs: Dict[str, Any] = Field(default_factory=dict)54    """Keyword arguments to pass to the Sentence Transformer model, such as `device`,55    `prompts`, `default_prompt_name`, `revision`, `trust_remote_code`, or `token`.56    See also the Sentence Transformer documentation: https://sbert.net/docs/package_reference/SentenceTransformer.html#sentence_transformers.SentenceTransformer"""57    encode_kwargs: Dict[str, Any] = Field(default_factory=dict)58    """Keyword arguments to pass when calling the `encode` method of the Sentence59    Transformer model, such as `prompt_name`, `prompt`, `batch_size`, `precision`,60    `normalize_embeddings`, and more.61    See also the Sentence Transformer documentation: https://sbert.net/docs/package_reference/SentenceTransformer.html#sentence_transformers.SentenceTransformer.encode"""62    multi_process: bool = False63    """Run encode() on multiple GPUs."""64    show_progress: bool = False65    """Whether to show a progress bar."""66 67    def __init__(self, **kwargs: Any):68        """Initialize the sentence_transformer."""69        super().__init__(**kwargs)70 71        if "model_name" not in kwargs:72            since = "0.2.16"73            removal = "0.4.0"74            warn_deprecated(75                since=since,76                removal=removal,77                message=f"Default values for {self.__class__.__name__}.model_name"78                + f" were deprecated in LangChain {since} and will be removed in"79                + f" {removal}. Explicitly pass a model_name to the"80                + f" {self.__class__.__name__} constructor instead.",81            )82 83        try:84            import sentence_transformers85 86        except ImportError as exc:87            raise ImportError(88                "Could not import sentence_transformers python package. "89                "Please install it with `pip install sentence-transformers`."90            ) from exc91 92        self.client = sentence_transformers.SentenceTransformer(93            self.model_name, cache_folder=self.cache_folder, **self.model_kwargs94        )95 96    model_config = ConfigDict(extra="forbid", protected_namespaces=())97 98    def embed_documents(self, texts: List[str]) -> List[List[float]]:99        """Compute doc embeddings using a HuggingFace transformer model.100 101        Args:102            texts: The list of texts to embed.103 104        Returns:105            List of embeddings, one for each text.106        """107        import sentence_transformers108 109        texts = list(map(lambda x: x.replace("\n", " "), texts))110        if self.multi_process:111            pool = self.client.start_multi_process_pool()112            embeddings = self.client.encode_multi_process(texts, pool)113            sentence_transformers.SentenceTransformer.stop_multi_process_pool(pool)114        else:115            embeddings = self.client.encode(116                texts, show_progress_bar=self.show_progress, **self.encode_kwargs117            )118 119        return embeddings.tolist()120 121    def embed_query(self, text: str) -> List[float]:122        """Compute query embeddings using a HuggingFace transformer model.123 124        Args:125            text: The text to embed.126 127        Returns:128            Embeddings for the text.129        """130        return self.embed_documents([text])[0]131 132 133@deprecated(134    since="0.2.2",135    removal="1.0",136    alternative_import="langchain_huggingface.HuggingFaceEmbeddings",137)138class HuggingFaceInstructEmbeddings(BaseModel, Embeddings):139    """Wrapper around sentence_transformers embedding models.140 141    To use, you should have the ``sentence_transformers``142    and ``InstructorEmbedding`` python packages installed.143 144    Example:145        .. code-block:: python146 147            from langchain_community.embeddings import HuggingFaceInstructEmbeddings148 149            model_name = "hkunlp/instructor-large"150            model_kwargs = {'device': 'cpu'}151            encode_kwargs = {'normalize_embeddings': True}152            hf = HuggingFaceInstructEmbeddings(153                model_name=model_name,154                model_kwargs=model_kwargs,155                encode_kwargs=encode_kwargs156            )157    """158 159    client: Any = None  #: :meta private:160    model_name: str = DEFAULT_INSTRUCT_MODEL161    """Model name to use."""162    cache_folder: Optional[str] = None163    """Path to store models. 164    Can be also set by SENTENCE_TRANSFORMERS_HOME environment variable."""165    model_kwargs: Dict[str, Any] = Field(default_factory=dict)166    """Keyword arguments to pass to the model."""167    encode_kwargs: Dict[str, Any] = Field(default_factory=dict)168    """Keyword arguments to pass when calling the `encode` method of the model."""169    embed_instruction: str = DEFAULT_EMBED_INSTRUCTION170    """Instruction to use for embedding documents."""171    query_instruction: str = DEFAULT_QUERY_INSTRUCTION172    """Instruction to use for embedding query."""173    show_progress: bool = False174    """Whether to show a progress bar."""175 176    def __init__(self, **kwargs: Any):177        """Initialize the sentence_transformer."""178        super().__init__(**kwargs)179 180        if "model_name" not in kwargs:181            since = "0.2.16"182            removal = "0.4.0"183            warn_deprecated(184                since=since,185                removal=removal,186                message=f"Default values for {self.__class__.__name__}.model_name"187                + f" were deprecated in LangChain {since} and will be removed in"188                + f" {removal}. Explicitly pass a model_name to the"189                + f" {self.__class__.__name__} constructor instead.",190            )191 192        try:193            from InstructorEmbedding import INSTRUCTOR194 195            self.client = INSTRUCTOR(196                self.model_name, cache_folder=self.cache_folder, **self.model_kwargs197            )198        except ImportError as e:199            raise ImportError("Dependencies for InstructorEmbedding not found.") from e200 201        if "show_progress_bar" in self.encode_kwargs:202            warn_deprecated(203                since="0.2.5",204                removal="1.0",205                name="encode_kwargs['show_progress_bar']",206                alternative=f"the show_progress method on {self.__class__.__name__}",207            )208            if self.show_progress:209                warnings.warn(210                    "Both encode_kwargs['show_progress_bar'] and show_progress are set;"211                    "encode_kwargs['show_progress_bar'] takes precedence"212                )213            self.show_progress = self.encode_kwargs.pop("show_progress_bar")214 215    model_config = ConfigDict(extra="forbid", protected_namespaces=())216 217    def embed_documents(self, texts: List[str]) -> List[List[float]]:218        """Compute doc embeddings using a HuggingFace instruct model.219 220        Args:221            texts: The list of texts to embed.222 223        Returns:224            List of embeddings, one for each text.225        """226        instruction_pairs = [[self.embed_instruction, text] for text in texts]227        embeddings = self.client.encode(228            instruction_pairs,229            show_progress_bar=self.show_progress,230            **self.encode_kwargs,231        )232        return embeddings.tolist()233 234    def embed_query(self, text: str) -> List[float]:235        """Compute query embeddings using a HuggingFace instruct model.236 237        Args:238            text: The text to embed.239 240        Returns:241            Embeddings for the text.242        """243        instruction_pair = [self.query_instruction, text]244        embedding = self.client.encode(245            [instruction_pair],246            show_progress_bar=self.show_progress,247            **self.encode_kwargs,248        )[0]249        return embedding.tolist()250 251 252@deprecated(253    since="0.2.2",254    removal="1.0",255    alternative_import="langchain_huggingface.HuggingFaceEmbeddings",256)257class HuggingFaceBgeEmbeddings(BaseModel, Embeddings):258    """HuggingFace sentence_transformers embedding models.259 260    To use, you should have the ``sentence_transformers`` python package installed.261    To use Nomic, make sure the version of ``sentence_transformers`` >= 2.3.0.262 263    Bge Example:264        .. code-block:: python265 266            from langchain_community.embeddings import HuggingFaceBgeEmbeddings267 268            model_name = "BAAI/bge-large-en-v1.5"269            model_kwargs = {'device': 'cpu'}270            encode_kwargs = {'normalize_embeddings': True}271            hf = HuggingFaceBgeEmbeddings(272                model_name=model_name,273                model_kwargs=model_kwargs,274                encode_kwargs=encode_kwargs275            )276     Nomic Example:277        .. code-block:: python278 279            from langchain_community.embeddings import HuggingFaceBgeEmbeddings280 281            model_name = "nomic-ai/nomic-embed-text-v1"282            model_kwargs = {283                'device': 'cpu',284                'trust_remote_code':True285                }286            encode_kwargs = {'normalize_embeddings': True}287            hf = HuggingFaceBgeEmbeddings(288                model_name=model_name,289                model_kwargs=model_kwargs,290                encode_kwargs=encode_kwargs,291                query_instruction = "search_query:",292                embed_instruction = "search_document:"293            )294    """295 296    client: Any = None  #: :meta private:297    model_name: str = DEFAULT_BGE_MODEL298    """Model name to use."""299    cache_folder: Optional[str] = None300    """Path to store models.301    Can be also set by SENTENCE_TRANSFORMERS_HOME environment variable."""302    model_kwargs: Dict[str, Any] = Field(default_factory=dict)303    """Keyword arguments to pass to the model."""304    encode_kwargs: Dict[str, Any] = Field(default_factory=dict)305    """Keyword arguments to pass when calling the `encode` method of the model."""306    query_instruction: str = DEFAULT_QUERY_BGE_INSTRUCTION_EN307    """Instruction to use for embedding query."""308    embed_instruction: str = ""309    """Instruction to use for embedding document."""310    show_progress: bool = False311    """Whether to show a progress bar."""312 313    def __init__(self, **kwargs: Any):314        """Initialize the sentence_transformer."""315        super().__init__(**kwargs)316 317        if "model_name" not in kwargs:318            since = "0.2.5"319            removal = "0.4.0"320            warn_deprecated(321                since=since,322                removal=removal,323                message=f"Default values for {self.__class__.__name__}.model_name"324                + f" were deprecated in LangChain {since} and will be removed in"325                + f" {removal}. Explicitly pass a model_name to the"326                + f" {self.__class__.__name__} constructor instead.",327            )328 329        try:330            import sentence_transformers331 332        except ImportError as exc:333            raise ImportError(334                "Could not import sentence_transformers python package. "335                "Please install it with `pip install sentence-transformers`."336            ) from exc337        extra_model_kwargs = [338            "torch_dtype",339            "attn_implementation",340            "provider",341            "file_name",342            "export",343        ]344        extra_model_kwargs_dict = {345            k: self.model_kwargs.pop(k)346            for k in extra_model_kwargs347            if k in self.model_kwargs348        }349        self.client = sentence_transformers.SentenceTransformer(350            self.model_name,351            cache_folder=self.cache_folder,352            **self.model_kwargs,353            model_kwargs=extra_model_kwargs_dict,354        )355 356        if "-zh" in self.model_name:357            self.query_instruction = DEFAULT_QUERY_BGE_INSTRUCTION_ZH358 359        if "show_progress_bar" in self.encode_kwargs:360            warn_deprecated(361                since="0.2.5",362                removal="1.0",363                name="encode_kwargs['show_progress_bar']",364                alternative=f"the show_progress method on {self.__class__.__name__}",365            )366            if self.show_progress:367                warnings.warn(368                    "Both encode_kwargs['show_progress_bar'] and show_progress are set;"369                    "encode_kwargs['show_progress_bar'] takes precedence"370                )371            self.show_progress = self.encode_kwargs.pop("show_progress_bar")372 373    model_config = ConfigDict(extra="forbid", protected_namespaces=())374 375    def embed_documents(self, texts: List[str]) -> List[List[float]]:376        """Compute doc embeddings using a HuggingFace transformer model.377 378        Args:379            texts: The list of texts to embed.380 381        Returns:382            List of embeddings, one for each text.383        """384        texts = [self.embed_instruction + t.replace("\n", " ") for t in texts]385        embeddings = self.client.encode(386            texts, show_progress_bar=self.show_progress, **self.encode_kwargs387        )388        return embeddings.tolist()389 390    def embed_query(self, text: str) -> List[float]:391        """Compute query embeddings using a HuggingFace transformer model.392 393        Args:394            text: The text to embed.395 396        Returns:397            Embeddings for the text.398        """399        text = text.replace("\n", " ")400        embedding = self.client.encode(401            self.query_instruction + text,402            show_progress_bar=self.show_progress,403            **self.encode_kwargs,404        )405        return embedding.tolist()406 407 408@deprecated(409    since="0.2.2",410    removal="1.0",411    alternative_import="langchain_huggingface.HuggingFaceEndpointEmbeddings",412)413class HuggingFaceInferenceAPIEmbeddings(BaseModel, Embeddings):414    """Embed texts using the HuggingFace API.415 416    Requires a HuggingFace Inference API key and a model name.417    """418 419    api_key: SecretStr420    """Your API key for the HuggingFace Inference API."""421    model_name: str = "sentence-transformers/all-MiniLM-L6-v2"422    """The name of the model to use for text embeddings."""423    api_url: Optional[str] = None424    """Custom inference endpoint url. None for using default public url."""425    additional_headers: Dict[str, str] = {}426    """Pass additional headers to the requests library if needed."""427 428    model_config = ConfigDict(extra="forbid", protected_namespaces=())429 430    @property431    def _api_url(self) -> str:432        return self.api_url or self._default_api_url433 434    @property435    def _default_api_url(self) -> str:436        return (437            "https://api-inference.huggingface.co"438            "/pipeline"439            "/feature-extraction"440            f"/{self.model_name}"441        )442 443    @property444    def _headers(self) -> dict:445        return {446            "Authorization": f"Bearer {self.api_key.get_secret_value()}",447            **self.additional_headers,448        }449 450    def embed_documents(self, texts: List[str]) -> List[List[float]]:451        """Get the embeddings for a list of texts.452 453        Args:454            texts (Documents): A list of texts to get embeddings for.455 456        Returns:457            Embedded texts as List[List[float]], where each inner List[float]458                corresponds to a single input text.459 460        Example:461            .. code-block:: python462 463                from langchain_community.embeddings import (464                    HuggingFaceInferenceAPIEmbeddings,465                )466 467                hf_embeddings = HuggingFaceInferenceAPIEmbeddings(468                    api_key="your_api_key",469                    model_name="sentence-transformers/all-MiniLM-l6-v2"470                )471                texts = ["Hello, world!", "How are you?"]472                hf_embeddings.embed_documents(texts)473        """  # noqa: E501474        response = requests.post(475            self._api_url,476            headers=self._headers,477            json={478                "inputs": texts,479                "options": {"wait_for_model": True, "use_cache": True},480            },481        )482        return response.json()483 484    def embed_query(self, text: str) -> List[float]:485        """Compute query embeddings using a HuggingFace transformer model.486 487        Args:488            text: The text to embed.489 490        Returns:491            Embeddings for the text.492        """493        return self.embed_documents([text])[0]494