codekingpro/portable-devtools

1from typing import Any, Dict, List, Optional, Union2 3from aiohttp import ClientSession4from langchain_core.callbacks import (5    AsyncCallbackManagerForLLMRun,6    CallbackManagerForLLMRun,7)8from langchain_core.language_models.chat_models import (9    BaseChatModel,10)11from langchain_core.messages import (12    AIMessage,13    BaseMessage,14)15from langchain_core.outputs import ChatGeneration, ChatResult16from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env17from pydantic import ConfigDict, Field, SecretStr, model_validator18 19from langchain_community.utilities.requests import Requests20 21 22def _format_dappier_messages(23    messages: List[BaseMessage],24) -> List[Dict[str, Union[str, List[Union[str, Dict[Any, Any]]]]]]:25    formatted_messages = []26 27    for message in messages:28        if message.type == "human":29            formatted_messages.append({"role": "user", "content": message.content})30        elif message.type == "system":31            formatted_messages.append({"role": "system", "content": message.content})32 33    return formatted_messages34 35 36class ChatDappierAI(BaseChatModel):37    """`Dappier` chat large language models.38 39    `Dappier` is a platform enabling access to diverse, real-time data models.40    Enhance your AI applications with Dappier's pre-trained, LLM-ready data models41    and ensure accurate, current responses with reduced inaccuracies.42 43    To use one of our Dappier AI Data Models, you will need an API key.44    Please visit Dappier Platform (https://platform.dappier.com/) to log in45    and create an API key in your profile.46 47    Example:48        .. code-block:: python49 50            from langchain_community.chat_models import ChatDappierAI51            from langchain_core.messages import HumanMessage52 53            # Initialize `ChatDappierAI` with the desired configuration54            chat = ChatDappierAI(55                dappier_endpoint="https://api.dappier.com/app/datamodel/dm_01hpsxyfm2fwdt2zet9cg6fdxt",56                dappier_api_key="<YOUR_KEY>")57 58            # Create a list of messages to interact with the model59            messages = [HumanMessage(content="hello")]60 61            # Invoke the model with the provided messages62            chat.invoke(messages)63 64 65    you can find more details here : https://docs.dappier.com/introduction"""66 67    dappier_endpoint: str = "https://api.dappier.com/app/datamodelconversation"68 69    dappier_model: str = "dm_01hpsxyfm2fwdt2zet9cg6fdxt"70 71    dappier_api_key: Optional[SecretStr] = Field(None, description="Dappier API Token")72 73    model_config = ConfigDict(74        extra="forbid",75    )76 77    @model_validator(mode="before")78    @classmethod79    def validate_environment(cls, values: Dict) -> Any:80        """Validate that api key exists in environment."""81        values["dappier_api_key"] = convert_to_secret_str(82            get_from_dict_or_env(values, "dappier_api_key", "DAPPIER_API_KEY")83        )84        return values85 86    @staticmethod87    def get_user_agent() -> str:88        from langchain_community import __version__89 90        return f"langchain/{__version__}"91 92    @property93    def _llm_type(self) -> str:94        """Return type of chat model."""95        return "dappier-realtimesearch-chat"96 97    @property98    def _api_key(self) -> str:99        if self.dappier_api_key:100            return self.dappier_api_key.get_secret_value()101        return ""102 103    def _generate(104        self,105        messages: List[BaseMessage],106        stop: Optional[List[str]] = None,107        run_manager: Optional[CallbackManagerForLLMRun] = None,108        **kwargs: Any,109    ) -> ChatResult:110        url = f"{self.dappier_endpoint}"111        headers = {112            "Authorization": f"Bearer {self._api_key}",113            "User-Agent": self.get_user_agent(),114        }115        user_query = _format_dappier_messages(messages=messages)116        payload: Dict[str, Any] = {117            "model": self.dappier_model,118            "conversation": user_query,119        }120 121        request = Requests(headers=headers)122        response = request.post(url=url, data=payload)123        response.raise_for_status()124 125        data = response.json()126 127        message_response = data["message"]128 129        return ChatResult(130            generations=[ChatGeneration(message=AIMessage(content=message_response))]131        )132 133    async def _agenerate(134        self,135        messages: List[BaseMessage],136        stop: Optional[List[str]] = None,137        run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,138        **kwargs: Any,139    ) -> ChatResult:140        url = f"{self.dappier_endpoint}"141        headers = {142            "Authorization": f"Bearer {self._api_key}",143            "User-Agent": self.get_user_agent(),144        }145        user_query = _format_dappier_messages(messages=messages)146        payload: Dict[str, Any] = {147            "model": self.dappier_model,148            "conversation": user_query,149        }150 151        async with ClientSession() as session:152            async with session.post(url, json=payload, headers=headers) as response:153                response.raise_for_status()154                data = await response.json()155                message_response = data["message"]156 157                return ChatResult(158                    generations=[159                        ChatGeneration(message=AIMessage(content=message_response))160                    ]161                )162