codekingpro/portable-devtools

1from __future__ import annotations2 3from typing import Any, Dict, List, Optional, cast4 5import requests6from langchain_core._api.deprecation import deprecated7from langchain_core.embeddings import Embeddings8from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env9from pydantic import BaseModel, ConfigDict, SecretStr, model_validator10 11 12@deprecated(13    since="0.3.4",14    removal="1.0.0",15    alternative_import="langchain_community.ClovaXEmbeddings",16)17class ClovaEmbeddings(BaseModel, Embeddings):18    """19    Clova's embedding service.20 21    To use this service,22 23    you should have the following environment variables24    set with your API tokens and application ID,25    or pass them as named parameters to the constructor:26 27    - ``CLOVA_EMB_API_KEY``: API key for accessing Clova's embedding service.28    - ``CLOVA_EMB_APIGW_API_KEY``: API gateway key for enhanced security.29    - ``CLOVA_EMB_APP_ID``: Application ID for identifying your application.30 31    Example:32        .. code-block:: python33 34            from langchain_community.embeddings import ClovaEmbeddings35            embeddings = ClovaEmbeddings(36                clova_emb_api_key='your_clova_emb_api_key',37                clova_emb_apigw_api_key='your_clova_emb_apigw_api_key',38                app_id='your_app_id'39            )40 41            query_text = "This is a test query."42            query_result = embeddings.embed_query(query_text)43 44            document_text = "This is a test document."45            document_result = embeddings.embed_documents([document_text])46 47    """48 49    endpoint_url: str = (50        "https://clovastudio.apigw.ntruss.com/testapp/v1/api-tools/embedding"51    )52    """Endpoint URL to use."""53    model: str = "clir-emb-dolphin"54    """Embedding model name to use."""55    clova_emb_api_key: Optional[SecretStr] = None56    """API key for accessing Clova's embedding service."""57    clova_emb_apigw_api_key: Optional[SecretStr] = None58    """API gateway key for enhanced security."""59    app_id: Optional[SecretStr] = None60    """Application ID for identifying your application."""61 62    model_config = ConfigDict(63        extra="forbid",64    )65 66    @model_validator(mode="before")67    @classmethod68    def validate_environment(cls, values: Dict) -> Any:69        """Validate api key exists in environment."""70        values["clova_emb_api_key"] = convert_to_secret_str(71            get_from_dict_or_env(values, "clova_emb_api_key", "CLOVA_EMB_API_KEY")72        )73        values["clova_emb_apigw_api_key"] = convert_to_secret_str(74            get_from_dict_or_env(75                values, "clova_emb_apigw_api_key", "CLOVA_EMB_APIGW_API_KEY"76            )77        )78        values["app_id"] = convert_to_secret_str(79            get_from_dict_or_env(values, "app_id", "CLOVA_EMB_APP_ID")80        )81        return values82 83    def embed_documents(self, texts: List[str]) -> List[List[float]]:84        """85        Embed a list of texts and return their embeddings.86 87        Args:88            texts: The list of texts to embed.89 90        Returns:91            List of embeddings, one for each text.92        """93        embeddings = []94        for text in texts:95            embeddings.append(self._embed_text(text))96        return embeddings97 98    def embed_query(self, text: str) -> List[float]:99        """100        Embed a single query text and return its embedding.101 102        Args:103            text: The text to embed.104 105        Returns:106            Embeddings for the text.107        """108        return self._embed_text(text)109 110    def _embed_text(self, text: str) -> List[float]:111        """112        Internal method to call the embedding API and handle the response.113        """114        payload = {"text": text}115 116        # HTTP headers for authorization117        headers = {118            "X-NCP-CLOVASTUDIO-API-KEY": cast(119                SecretStr, self.clova_emb_api_key120            ).get_secret_value(),121            "X-NCP-APIGW-API-KEY": cast(122                SecretStr, self.clova_emb_apigw_api_key123            ).get_secret_value(),124            "Content-Type": "application/json",125        }126 127        # send request128        app_id = cast(SecretStr, self.app_id).get_secret_value()129        response = requests.post(130            f"{self.endpoint_url}/{self.model}/{app_id}",131            headers=headers,132            json=payload,133        )134 135        # check for errors136        if response.status_code == 200:137            response_data = response.json()138            if "result" in response_data and "embedding" in response_data["result"]:139                return response_data["result"]["embedding"]140        raise ValueError(141            f"API request failed with status {response.status_code}: {response.text}"142        )143