Source code for langchain_community.embeddings.solar

from __future__ import annotations

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

import requests
from langchain_core._api import deprecated
from langchain_core.embeddings import Embeddings
from langchain_core.pydantic_v1 import BaseModel, Extra, SecretStr, root_validator
from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env
from tenacity import (
    before_sleep_log,
    retry,
    stop_after_attempt,
    wait_exponential,
)

logger = logging.getLogger(__name__)


def _create_retry_decorator() -> Callable[[Any], Any]:
    """返回一个用于重试的坚韧性装饰器。"""

    multiplier = 1
    min_seconds = 1
    max_seconds = 4
    max_retries = 6

    return retry(
        reraise=True,
        stop=stop_after_attempt(max_retries),
        wait=wait_exponential(multiplier=multiplier, min=min_seconds, max=max_seconds),
        before_sleep=before_sleep_log(logger, logging.WARNING),
    )


def embed_with_retry(embeddings: SolarEmbeddings, *args: Any, **kwargs: Any) -> Any:
    """使用tenacity来重试完成调用。"""
    retry_decorator = _create_retry_decorator()

    @retry_decorator
    def _embed_with_retry(*args: Any, **kwargs: Any) -> Any:
        return embeddings.embed(*args, **kwargs)

    return _embed_with_retry(*args, **kwargs)


@deprecated(
    since="0.0.34", removal="0.3.0", alternative_import="langchain_upstage.ChatUpstage"
)
class SolarEmbeddings(BaseModel, Embeddings):
    """Solar的嵌入服务。

    要使用，请确保环境变量``SOLAR_API_KEY``已设置为您的API令牌，或者将其作为构造函数的命名参数传递。

    示例:
        .. code-block:: python

            from langchain_community.embeddings import SolarEmbeddings
            embeddings = SolarEmbeddings()

            query_text = "这是一个测试查询。"
            query_result = embeddings.embed_query(query_text)

            document_text = "这是一个测试文档。"
            document_result = embeddings.embed_documents([document_text])
"""

    endpoint_url: str = "https://api.upstage.ai/v1/solar/embeddings"
    """要使用的端点URL。"""
    model: str = "solar-1-mini-embedding-query"
    """嵌入模型的名称。"""
    solar_api_key: Optional[SecretStr] = None
    """Solar API的API密钥。"""

    class Config:
        """此pydantic对象的配置。"""

        extra = Extra.forbid

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """验证API密钥是否存在于环境中。"""
        solar_api_key = convert_to_secret_str(
            get_from_dict_or_env(values, "solar_api_key", "SOLAR_API_KEY")
        )
        values["solar_api_key"] = solar_api_key
        return values

    def embed(
        self,
        text: str,
    ) -> List[List[float]]:
        payload = {
            "model": self.model,
            "input": text,
        }

        # HTTP headers for authorization
        headers = {
            "Authorization": f"Bearer {self.solar_api_key.get_secret_value()}",  # type: ignore[union-attr]
            "Content-Type": "application/json",
        }

        # send request
        response = requests.post(self.endpoint_url, headers=headers, json=payload)
        parsed_response = response.json()

        # check for errors
        if len(parsed_response["data"]) == 0:
            raise ValueError(
                f"Solar API returned an error: {parsed_response['base_resp']}"
            )

        embedding = parsed_response["data"][0]["embedding"]

        return embedding

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """使用Solar嵌入端点嵌入文档。

参数：
    texts：要嵌入的文本列表。

返回：
    每个文本的嵌入列表。
"""
        embeddings = [embed_with_retry(self, text=text) for text in texts]
        return embeddings

    def embed_query(self, text: str) -> List[float]:
        """在使用Solar嵌入端点嵌入查询。

参数：
    text：要嵌入的文本。

返回：
    文本的嵌入。
"""
        embedding = embed_with_retry(self, text=text)
        return embedding