Source code for langchain_community.llms.arcee

from typing import Any, Dict, List, Optional, Union, cast

from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
from langchain_core.pydantic_v1 import Extra, SecretStr, root_validator
from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env

from langchain_community.utilities.arcee import ArceeWrapper, DALMFilter


class Arcee(LLM):
    """Arcee的领域自适应语言模型（DALMs）。

    要使用，请设置``ARCEE_API_KEY``环境变量为您的Arcee API密钥，
    或将``arcee_api_key``作为命名参数传递。

    示例：
        .. code-block:: python

            from langchain_community.llms import Arcee

            arcee = Arcee(
                model="DALM-PubMed",
                arcee_api_key="ARCEE-API-KEY"
            )

            response = arcee("基于AI的音乐疗法")
"""

    _client: Optional[ArceeWrapper] = None  #: :meta private:
    """Arcee _client。"""

    arcee_api_key: Union[SecretStr, str, None] = None
    """Arcee API密钥"""

    model: str
    """Arcee DALM 名称"""

    arcee_api_url: str = "https://api.arcee.ai"
    """Arcee API URL"""

    arcee_api_version: str = "v2"
    """Arcee API版本"""

    arcee_app_url: str = "https://app.arcee.ai"
    """Arcee 应用程序 URL"""

    model_id: str = ""
    """阿尔茜模型ID"""

    model_kwargs: Optional[Dict[str, Any]] = None
    """传递给模型的关键字参数。"""

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

        extra = Extra.forbid
        underscore_attrs_are_private = True

    @property
    def _llm_type(self) -> str:
        """llm的返回类型。"""
        return "arcee"

    def __init__(self, **data: Any) -> None:
        """初始化私有字段。"""

        super().__init__(**data)
        api_key = cast(SecretStr, self.arcee_api_key)
        self._client = ArceeWrapper(
            arcee_api_key=api_key,
            arcee_api_url=self.arcee_api_url,
            arcee_api_version=self.arcee_api_version,
            model_kwargs=self.model_kwargs,
            model_name=self.model,
        )

    @root_validator(pre=False)
    def validate_environments(cls, values: Dict) -> Dict:
        """验证Arcee环境变量。"""

        # validate env vars
        values["arcee_api_key"] = convert_to_secret_str(
            get_from_dict_or_env(
                values,
                "arcee_api_key",
                "ARCEE_API_KEY",
            )
        )

        values["arcee_api_url"] = get_from_dict_or_env(
            values,
            "arcee_api_url",
            "ARCEE_API_URL",
        )

        values["arcee_app_url"] = get_from_dict_or_env(
            values,
            "arcee_app_url",
            "ARCEE_APP_URL",
        )

        values["arcee_api_version"] = get_from_dict_or_env(
            values,
            "arcee_api_version",
            "ARCEE_API_VERSION",
        )

        # validate model kwargs
        if values.get("model_kwargs"):
            kw = values["model_kwargs"]

            # validate size
            if kw.get("size") is not None:
                if not kw.get("size") >= 0:
                    raise ValueError("`size` must be positive")

            # validate filters
            if kw.get("filters") is not None:
                if not isinstance(kw.get("filters"), List):
                    raise ValueError("`filters` must be a list")
                for f in kw.get("filters"):
                    DALMFilter(**f)
        return values

    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        """从Arcee DALM生成文本。

参数：
    prompt：要生成文本的提示。
    size：要检索的上下文结果的最大数量。
    默认为3。（如果提供了过滤器，则可能会更少）。
    filters：要应用于上下文数据集的过滤器。
"""

        try:
            if not self._client:
                raise ValueError("Client is not initialized.")
            return self._client.generate(prompt=prompt, **kwargs)
        except Exception as e:
            raise Exception(f"Failed to generate text: {e}") from e