codekingpro/portable-devtools

1import logging2from typing import Any, List, Mapping, Optional3 4import requests5from langchain_core.callbacks import CallbackManagerForLLMRun6from langchain_core.language_models.llms import LLM7 8from langchain_community.llms.utils import enforce_stop_tokens9 10logger = logging.getLogger(__name__)11 12 13class ChatGLM(LLM):14    """ChatGLM LLM service.15 16    Example:17        .. code-block:: python18 19            from langchain_community.llms import ChatGLM20            endpoint_url = (21                "http://127.0.0.1:8000"22            )23            ChatGLM_llm = ChatGLM(24                endpoint_url=endpoint_url25            )26    """27 28    endpoint_url: str = "http://127.0.0.1:8000/"29    """Endpoint URL to use."""30    model_kwargs: Optional[dict] = None31    """Keyword arguments to pass to the model."""32    max_token: int = 2000033    """Max token allowed to pass to the model."""34    temperature: float = 0.135    """LLM model temperature from 0 to 10."""36    history: List[List] = []37    """History of the conversation"""38    top_p: float = 0.739    """Top P for nucleus sampling from 0 to 1"""40    with_history: bool = False41    """Whether to use history or not"""42 43    @property44    def _llm_type(self) -> str:45        return "chat_glm"46 47    @property48    def _identifying_params(self) -> Mapping[str, Any]:49        """Get the identifying parameters."""50        _model_kwargs = self.model_kwargs or {}51        return {52            **{"endpoint_url": self.endpoint_url},53            **{"model_kwargs": _model_kwargs},54        }55 56    def _call(57        self,58        prompt: str,59        stop: Optional[List[str]] = None,60        run_manager: Optional[CallbackManagerForLLMRun] = None,61        **kwargs: Any,62    ) -> str:63        """Call out to a ChatGLM LLM inference endpoint.64 65        Args:66            prompt: The prompt to pass into the model.67            stop: Optional list of stop words to use when generating.68 69        Returns:70            The string generated by the model.71 72        Example:73            .. code-block:: python74 75                response = chatglm_llm.invoke("Who are you?")76        """77 78        _model_kwargs = self.model_kwargs or {}79 80        # HTTP headers for authorization81        headers = {"Content-Type": "application/json"}82 83        payload = {84            "prompt": prompt,85            "temperature": self.temperature,86            "history": self.history,87            "max_length": self.max_token,88            "top_p": self.top_p,89        }90        payload.update(_model_kwargs)91        payload.update(kwargs)92 93        logger.debug(f"ChatGLM payload: {payload}")94 95        # call api96        try:97            response = requests.post(self.endpoint_url, headers=headers, json=payload)98        except requests.exceptions.RequestException as e:99            raise ValueError(f"Error raised by inference endpoint: {e}")100 101        logger.debug(f"ChatGLM response: {response}")102 103        if response.status_code != 200:104            raise ValueError(f"Failed with response: {response}")105 106        try:107            parsed_response = response.json()108 109            # Check if response content does exists110            if isinstance(parsed_response, dict):111                content_keys = "response"112                if content_keys in parsed_response:113                    text = parsed_response[content_keys]114                else:115                    raise ValueError(f"No content in response : {parsed_response}")116            else:117                raise ValueError(f"Unexpected response type: {parsed_response}")118 119        except requests.exceptions.JSONDecodeError as e:120            raise ValueError(121                f"Error raised during decoding response from inference endpoint: {e}."122                f"\nResponse: {response.text}"123            )124 125        if stop is not None:126            text = enforce_stop_tokens(text, stop)127        if self.with_history:128            self.history = parsed_response["history"]129        return text130