"""封装了Minimax API的函数。"""
from __future__ import annotations
import logging
from typing import (
Any,
Dict,
List,
Optional,
)
import requests
from langchain_core.callbacks import (
CallbackManagerForLLMRun,
)
from langchain_core.language_models.llms import LLM
from langchain_core.pydantic_v1 import BaseModel, Field, SecretStr, root_validator
from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env
from langchain_community.llms.utils import enforce_stop_tokens
logger = logging.getLogger(__name__)
class _MinimaxEndpointClient(BaseModel):
"""用于Minimax LLM端点的API客户端。"""
host: str
group_id: str
api_key: SecretStr
api_url: str
@root_validator(pre=True, allow_reuse=True)
def set_api_url(cls, values: Dict[str, Any]) -> Dict[str, Any]:
if "api_url" not in values:
host = values["host"]
group_id = values["group_id"]
api_url = f"{host}/v1/text/chatcompletion?GroupId={group_id}"
values["api_url"] = api_url
return values
def post(self, request: Any) -> Any:
headers = {"Authorization": f"Bearer {self.api_key.get_secret_value()}"}
response = requests.post(self.api_url, headers=headers, json=request)
# TODO: error handling and automatic retries
if not response.ok:
raise ValueError(f"HTTP {response.status_code} error: {response.text}")
if response.json()["base_resp"]["status_code"] > 0:
raise ValueError(
f"API {response.json()['base_resp']['status_code']}"
f" error: {response.json()['base_resp']['status_msg']}"
)
return response.json()["reply"]
[docs]class MinimaxCommon(BaseModel):
"""Minimax大型语言模型的常见参数。"""
_client: _MinimaxEndpointClient
model: str = "abab5.5-chat"
"""要使用的模型名称。"""
max_tokens: int = 256
"""表示每代要预测的令牌数量。"""
temperature: float = 0.7
"""一个非负浮点数,用于调整生成过程中的随机程度。"""
top_p: float = 0.95
"""每一步需要考虑的标记的总概率质量。"""
model_kwargs: Dict[str, Any] = Field(default_factory=dict)
"""保存任何在`create`调用中有效但未明确指定的模型参数。"""
minimax_api_host: Optional[str] = None
minimax_group_id: Optional[str] = None
minimax_api_key: Optional[SecretStr] = None
@root_validator()
def validate_environment(cls, values: Dict) -> Dict:
"""验证环境中是否存在API密钥和Python包。"""
values["minimax_api_key"] = convert_to_secret_str(
get_from_dict_or_env(values, "minimax_api_key", "MINIMAX_API_KEY")
)
values["minimax_group_id"] = get_from_dict_or_env(
values, "minimax_group_id", "MINIMAX_GROUP_ID"
)
# Get custom api url from environment.
values["minimax_api_host"] = get_from_dict_or_env(
values,
"minimax_api_host",
"MINIMAX_API_HOST",
default="https://api.minimax.chat",
)
values["_client"] = _MinimaxEndpointClient( # type: ignore[call-arg]
host=values["minimax_api_host"],
api_key=values["minimax_api_key"],
group_id=values["minimax_group_id"],
)
return values
@property
def _default_params(self) -> Dict[str, Any]:
"""获取调用OpenAI API的默认参数。"""
return {
"model": self.model,
"tokens_to_generate": self.max_tokens,
"temperature": self.temperature,
"top_p": self.top_p,
**self.model_kwargs,
}
@property
def _identifying_params(self) -> Dict[str, Any]:
"""获取识别参数。"""
return {**{"model": self.model}, **self._default_params}
@property
def _llm_type(self) -> str:
"""llm的返回类型。"""
return "minimax"
[docs]class Minimax(MinimaxCommon, LLM):
"""大型语言模型的极小化。
要使用,您应该设置环境变量``MINIMAX_API_KEY``和``MINIMAX_GROUP_ID``,并使用您的API密钥,
或将它们作为命名参数传递给构造函数。
示例:
. code-block:: python
from langchain_community.llms.minimax import Minimax
minimax = Minimax(model="<model_name>", minimax_api_key="my-api-key",
minimax_group_id="my-group-id")
"""
def _call(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> str:
r"""调用Minimax的完成端点进行聊天
参数:
prompt: 传递给模型的提示。
返回:
模型生成的字符串。
示例:
.. code-block:: python
response = minimax("告诉我一个笑话。")
"""
request = self._default_params
request["messages"] = [{"sender_type": "USER", "text": prompt}]
request.update(kwargs)
text = self._client.post(request)
if stop is not None:
# This is required since the stop tokens
# are not enforced by the model parameters
text = enforce_stop_tokens(text, stop)
return text