from typing import Any, Dict, List, Optional
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models import BaseLLM
from langchain_core.outputs import Generation, LLMResult
from langchain_core.pydantic_v1 import Field, root_validator
[docs]class Aphrodite(BaseLLM):
"""阿芙洛狄忒语言模型。"""
model: str = ""
"""HuggingFace Transformers 模型的名称或路径。"""
tensor_parallel_size: Optional[int] = 1
"""用于张量并行分布式执行的GPU数量。"""
trust_remote_code: Optional[bool] = False
"""信任远程代码（例如来自HuggingFace）在下载模型和分词器时。"""
n: int = 1
"""给定提示返回的输出序列数量。"""
best_of: Optional[int] = None
"""从提示生成的输出序列数量。
从这些`best_of`序列中，返回前`n`个序列。
`best_of`必须 >= `n`。当`use_beam_search`为True时，这被视为波束宽度。
默认情况下，`best_of`设置为`n`。"""
presence_penalty: float = 0.0
"""根据生成的文本迄今为止，对新标记进行惩罚的浮点数。值> 0鼓励模型生成新标记，而值< 0鼓励模型重复标记。"""
frequency_penalty: float = 0.0
"""根据生成文本中新标记的频率对其进行惩罚的浮点数。以对数形式添加到逻辑值中。"""
repetition_penalty: float = 1.0
"""根据生成文本中新标记的频率对其进行惩罚的浮点数。以对数的形式应用乘法。"""
temperature: float = 1.0
"""控制采样随机性的浮点数。较低的值使模型更确定，而较高的值使模型更随机。零等同于贪婪采样。"""
top_p: float = 1.0
"""控制要考虑的顶部标记的累积概率的浮点数。
必须在（0，1]范围内。设置为1.0以考虑所有标记。"""
top_k: int = -1
"""控制要考虑的顶部标记数量的整数。设置为-1以考虑所有标记（禁用）。"""
top_a: float = 0.0
"""控制Top-A采样截断的浮点数。确切的截断值为top_a*max_prob**2。必须在[0,inf]范围内，设为0则禁用。"""
min_p: float = 0.0
"""控制最小p采样截断的浮点数。确切的截断值为min_p*max_prob。必须在[0,1]范围内，为0时禁用。"""
tfs: float = 1.0
"""控制尾部自由采样中保留的累积近似曲率的浮点数。必须在(0, 1]范围内。设置为1.0以禁用。"""
eta_cutoff: float = 0.0
"""控制Eta采样截断阈值的浮点数（一种熵自适应截断采样）。阈值计算为`min(eta, sqrt(eta)*entropy(probs))`。以1e-4为单位指定。设置为0以禁用。"""
epsilon_cutoff: float = 0.0
"""控制Epsilon采样的截断阈值的浮点数（简单概率阈值截断）。以1e-4为单位指定。设置为0以禁用。"""
typical_p: float = 1.0
"""控制累积概率的浮点数，该概率表示与预期惊喜最接近的令牌的惊喜程度。必须在(0, 1]范围内。设置为1以禁用。"""
mirostat_mode: int = 0
"""要使用的mirostat模式。0表示没有mirostat，2表示mirostat v2。
不支持模式1。"""
mirostat_tau: float = 0.0
"""mirostat工作的目标'surprisal'。范围[0, 无穷)。"""
use_beam_search: bool = False
"""是否使用束搜索而不是采样。"""
length_penalty: float = 1.0
"""根据序列长度对序列进行惩罚的浮点数。仅在`use_beam_search`为True时使用。"""
early_stopping: bool = False
"""控制束搜索的停止条件。它接受以下值：`True`，在有`best_of`个完整候选项时停止生成；`False`，在生成时应用一种启发式方法，当很不可能找到更好的候选项时停止生成；`never`，其中束搜索过程仅在不能有更好的候选项时停止（标准束搜索算法）。"""
stop: Optional[List[str]] = None
"""生成时停止生成的字符串列表。
返回的输出将不包含停止标记。"""
stop_token_ids: Optional[List[int]] = None
"""生成时停止生成的令牌列表。
返回的输出将包含停止令牌，除非停止令牌是特殊令牌。"""
ignore_eos: bool = False
"""是否忽略EOS标记并在生成EOS标记后继续生成标记。"""
max_tokens: int = 512
"""每个输出序列生成的最大令牌数。"""
logprobs: Optional[int] = None
"""每个输出标记返回的对数概率数量。"""
prompt_logprobs: Optional[int] = None
"""每个提示标记返回的对数概率数量。"""
custom_token_bans: Optional[List[int]] = None
"""需要禁止生成的令牌ID列表。"""
skip_special_tokens: bool = True
"""是否跳过输出中的特殊标记。默认为True。"""
spaces_between_special_tokens: bool = True
"""是否在输出中的特殊标记之间添加空格。
默认为True。"""
logit_bias: Optional[Dict[str, float]] = None
"""在运行时更改标记预测概率的LogitsProcessors列表。"""
dtype: str = "auto"
"""模型权重和激活的数据类型。"""
download_dir: Optional[str] = None
"""下载和加载权重的目录。（默认为huggingface的默认缓存目录）"""
quantization: Optional[str] = None
"""量化模式要使用。可以是`awq`或`gptq`之一。"""
aphrodite_kwargs: Dict[str, Any] = Field(default_factory=dict)
"""保存任何未明确指定的`aphrodite.LLM`调用有效的模型参数。"""
client: Any #: :meta private:
@root_validator()
def validate_environment(cls, values: Dict) -> Dict:
"""验证Python包是否存在于环境中。"""
try:
from aphrodite import LLM as AphroditeModel
except ImportError:
raise ImportError(
"Could not import aphrodite-engine python package. "
"Please install it with `pip install aphrodite-engine`."
)
# aphrodite_kwargs = values["aphrodite_kwargs"]
# if values.get("quantization"):
# aphrodite_kwargs["quantization"] = values["quantization"]
values["client"] = AphroditeModel(
model=values["model"],
tensor_parallel_size=values["tensor_parallel_size"],
trust_remote_code=values["trust_remote_code"],
dtype=values["dtype"],
download_dir=values["download_dir"],
**values["aphrodite_kwargs"],
)
return values
@property
def _default_params(self) -> Dict[str, Any]:
"""获取调用Aphrodite时的默认参数。"""
return {
"n": self.n,
"best_of": self.best_of,
"max_tokens": self.max_tokens,
"top_k": self.top_k,
"top_p": self.top_p,
"top_a": self.top_a,
"min_p": self.min_p,
"temperature": self.temperature,
"presence_penalty": self.presence_penalty,
"frequency_penalty": self.frequency_penalty,
"repetition_penalty": self.repetition_penalty,
"tfs": self.tfs,
"eta_cutoff": self.eta_cutoff,
"epsilon_cutoff": self.epsilon_cutoff,
"typical_p": self.typical_p,
"mirostat_mode": self.mirostat_mode,
"mirostat_tau": self.mirostat_tau,
"length_penalty": self.length_penalty,
"early_stopping": self.early_stopping,
"use_beam_search": self.use_beam_search,
"stop": self.stop,
"ignore_eos": self.ignore_eos,
"logprobs": self.logprobs,
"prompt_logprobs": self.prompt_logprobs,
"custom_token_bans": self.custom_token_bans,
"skip_special_tokens": self.skip_special_tokens,
"spaces_between_special_tokens": self.spaces_between_special_tokens,
"logit_bias": self.logit_bias,
}
def _generate(
self,
prompts: List[str],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> LLMResult:
"""在给定的提示和输入上运行LLM。"""
from aphrodite import SamplingParams
# build sampling parameters
params = {**self._default_params, **kwargs, "stop": stop}
if "logit_bias" in params:
del params["logit_bias"]
sampling_params = SamplingParams(**params)
# call the model
outputs = self.client.generate(prompts, sampling_params)
generations = []
for output in outputs:
text = output.outputs[0].text
generations.append([Generation(text=text)])
return LLMResult(generations=generations)
@property
def _llm_type(self) -> str:
"""llm的返回类型。"""
return "aphrodite"