langchain_community.llms.mlflow — 🦜🔗 LangChain 0.2.0

from __future__ import annotations

from typing import Any, Dict, List, Mapping, Optional
from urllib.parse import urlparse

from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models import LLM
from langchain_core.pydantic_v1 import Field, PrivateAttr


[docs]class Mlflow(LLM):
    """MLflow LLM 服务。

    要使用，您应该安装 `mlflow[genai]` python 包。
    更多信息，请参见 https://mlflow.org/docs/latest/llms/deployments.

    示例:
        .. code-block:: python

            from langchain_community.llms import Mlflow

            completions = Mlflow(
                target_uri="http://localhost:5000",
                endpoint="test",
                temperature=0.1,
            )"""

    endpoint: str
    """要使用的端点。"""
    target_uri: str
    """要使用的目标URI。"""
    temperature: float = 0.0
    """采样温度。"""
    n: int = 1
    """生成完成选项的数量。"""
    stop: Optional[List[str]] = None
    """停止序列。"""
    max_tokens: Optional[int] = None
    """生成的最大令牌数量。"""
    extra_params: Dict[str, Any] = Field(default_factory=dict)
    """传递给端点的任何额外参数。"""

    """额外参数，比如`temperature`。"""
    _client: Any = PrivateAttr()

    def __init__(self, **kwargs: Any):
        super().__init__(**kwargs)
        self._validate_uri()
        try:
            from mlflow.deployments import get_deploy_client

            self._client = get_deploy_client(self.target_uri)
        except ImportError as e:
            raise ImportError(
                "Failed to create the client. "
                "Please run `pip install mlflow[genai]` to install "
                "required dependencies."
            ) from e

    def _validate_uri(self) -> None:
        if self.target_uri == "databricks":
            return
        allowed = ["http", "https", "databricks"]
        if urlparse(self.target_uri).scheme not in allowed:
            raise ValueError(
                f"Invalid target URI: {self.target_uri}. "
                f"The scheme must be one of {allowed}."
            )

    @property
    def _default_params(self) -> Dict[str, Any]:
        return {
            "target_uri": self.target_uri,
            "endpoint": self.endpoint,
            "temperature": self.temperature,
            "n": self.n,
            "stop": self.stop,
            "max_tokens": self.max_tokens,
            "extra_params": self.extra_params,
        }

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        return self._default_params

    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        data: Dict[str, Any] = {
            "prompt": prompt,
            "temperature": self.temperature,
            "n": self.n,
            **self.extra_params,
            **kwargs,
        }
        if stop := self.stop or stop:
            data["stop"] = stop
        if self.max_tokens is not None:
            data["max_tokens"] = self.max_tokens

        resp = self._client.predict(endpoint=self.endpoint, inputs=data)
        return resp["choices"][0]["text"]

    @property
    def _llm_type(self) -> str:
        return "mlflow"
Source code for langchain_community.llms.mlflow
