codekingpro/portable-devtools

1"""Wrapper around Minimax APIs."""2 3from __future__ import annotations4 5import logging6from typing import (7    Any,8    Dict,9    List,10    Optional,11)12 13import requests14from langchain_core.callbacks import (15    CallbackManagerForLLMRun,16)17from langchain_core.language_models.llms import LLM18from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env, pre_init19from pydantic import BaseModel, ConfigDict, Field, SecretStr, model_validator20 21from langchain_community.llms.utils import enforce_stop_tokens22 23logger = logging.getLogger(__name__)24 25 26class _MinimaxEndpointClient(BaseModel):27    """API client for the Minimax LLM endpoint."""28 29    host: str30    group_id: str31    api_key: SecretStr32    api_url: str33 34    @model_validator(mode="before")35    @classmethod36    def set_api_url(cls, values: Dict[str, Any]) -> Any:37        if "api_url" not in values:38            host = values["host"]39            group_id = values["group_id"]40            api_url = f"{host}/v1/text/chatcompletion?GroupId={group_id}"41            values["api_url"] = api_url42        return values43 44    def post(self, request: Any) -> Any:45        headers = {"Authorization": f"Bearer {self.api_key.get_secret_value()}"}46        response = requests.post(self.api_url, headers=headers, json=request)47        # TODO: error handling and automatic retries48        if not response.ok:49            raise ValueError(f"HTTP {response.status_code} error: {response.text}")50        if response.json()["base_resp"]["status_code"] > 0:51            raise ValueError(52                f"API {response.json()['base_resp']['status_code']}"53                f" error: {response.json()['base_resp']['status_msg']}"54            )55        return response.json()["reply"]56 57 58class MinimaxCommon(BaseModel):59    """Common parameters for Minimax large language models."""60 61    model_config = ConfigDict(protected_namespaces=())62 63    _client: _MinimaxEndpointClient64    model: str = "abab5.5-chat"65    """Model name to use."""66    max_tokens: int = 25667    """Denotes the number of tokens to predict per generation."""68    temperature: float = 0.769    """A non-negative float that tunes the degree of randomness in generation."""70    top_p: float = 0.9571    """Total probability mass of tokens to consider at each step."""72    model_kwargs: Dict[str, Any] = Field(default_factory=dict)73    """Holds any model parameters valid for `create` call not explicitly specified."""74    minimax_api_host: Optional[str] = None75    minimax_group_id: Optional[str] = None76    minimax_api_key: Optional[SecretStr] = None77 78    @pre_init79    def validate_environment(cls, values: Dict) -> Dict:80        """Validate that api key and python package exists in environment."""81        values["minimax_api_key"] = convert_to_secret_str(82            get_from_dict_or_env(values, "minimax_api_key", "MINIMAX_API_KEY")83        )84        values["minimax_group_id"] = get_from_dict_or_env(85            values, "minimax_group_id", "MINIMAX_GROUP_ID"86        )87        # Get custom api url from environment.88        values["minimax_api_host"] = get_from_dict_or_env(89            values,90            "minimax_api_host",91            "MINIMAX_API_HOST",92            default="https://api.minimax.chat",93        )94        values["_client"] = _MinimaxEndpointClient(  # type: ignore[call-arg]95            host=values["minimax_api_host"],96            api_key=values["minimax_api_key"],97            group_id=values["minimax_group_id"],98        )99        return values100 101    @property102    def _default_params(self) -> Dict[str, Any]:103        """Get the default parameters for calling OpenAI API."""104        return {105            "model": self.model,106            "tokens_to_generate": self.max_tokens,107            "temperature": self.temperature,108            "top_p": self.top_p,109            **self.model_kwargs,110        }111 112    @property113    def _identifying_params(self) -> Dict[str, Any]:114        """Get the identifying parameters."""115        return {**{"model": self.model}, **self._default_params}116 117    @property118    def _llm_type(self) -> str:119        """Return type of llm."""120        return "minimax"121 122 123class Minimax(MinimaxCommon, LLM):124    """Minimax large language models.125 126    To use, you should have the environment variable127    ``MINIMAX_API_KEY`` and ``MINIMAX_GROUP_ID`` set with your API key,128    or pass them as a named parameter to the constructor.129    Example:130     . code-block:: python131         from langchain_community.llms.minimax import Minimax132         minimax = Minimax(model="<model_name>", minimax_api_key="my-api-key",133          minimax_group_id="my-group-id")134    """135 136    def _call(137        self,138        prompt: str,139        stop: Optional[List[str]] = None,140        run_manager: Optional[CallbackManagerForLLMRun] = None,141        **kwargs: Any,142    ) -> str:143        r"""Call out to Minimax's completion endpoint to chat144        Args:145            prompt: The prompt to pass into the model.146        Returns:147            The string generated by the model.148        Example:149            .. code-block:: python150                response = minimax("Tell me a joke.")151        """152        request = self._default_params153        request["messages"] = [{"sender_type": "USER", "text": prompt}]154        request.update(kwargs)155        text = self._client.post(request)156        if stop is not None:157            # This is required since the stop tokens158            # are not enforced by the model parameters159            text = enforce_stop_tokens(text, stop)160 161        return text162