codekingpro/portable-devtools

1from __future__ import annotations2 3from typing import Any, Dict, Optional4 5from langchain_community.chat_message_histories import ZepCloudChatMessageHistory6 7try:8    from langchain_classic.memory import ConversationBufferMemory9    from zep_cloud import MemoryGetRequestMemoryType10 11    class ZepCloudMemory(ConversationBufferMemory):12        """Persist your chain history to the Zep MemoryStore.13 14        Documentation: https://help.getzep.com15 16        Example:17            .. code-block:: python18 19            memory = ZepCloudMemory(20                        session_id=session_id,  # Identifies your user or a user's session21                        api_key=<your_api_key>, # Your Zep Project API key22                        memory_key="history",   # Ensure this matches the key used in23                                                # chain's prompt template24                        return_messages=True,   # Does your prompt template expect a string25                                                # or a list of Messages?26                    )27            chain = LLMChain(memory=memory,...) # Configure your chain to use the ZepMemory28                                                  instance29 30 31        Note:32            To persist metadata alongside your chat history, your will need to create a33        custom Chain class that overrides the `prep_outputs` method to include the metadata34        in the call to `self.memory.save_context`.35 36 37        Zep - Recall, understand, and extract data from chat histories. Power personalized AI experiences.38        =========39        Zep is a long-term memory service for AI Assistant apps. With Zep, you can provide AI assistants with the ability to recall past conversations,40        no matter how distant, while also reducing hallucinations, latency, and cost.41 42        For more information on the zep-python package, see:43        https://github.com/getzep/zep-python44 45        """  # noqa: E50146 47        chat_memory: ZepCloudChatMessageHistory48 49        def __init__(50            self,51            session_id: str,52            api_key: str,53            memory_type: Optional[MemoryGetRequestMemoryType] = None,54            lastn: Optional[int] = None,55            output_key: Optional[str] = None,56            input_key: Optional[str] = None,57            return_messages: bool = False,58            human_prefix: str = "Human",59            ai_prefix: str = "AI",60            memory_key: str = "history",61        ):62            """Initialize ZepMemory.63 64            Args:65                session_id (str): Identifies your user or a user's session66                api_key (str): Your Zep Project key.67                memory_type (Optional[MemoryGetRequestMemoryType], optional): Zep Memory Type, defaults to perpetual68                lastn (Optional[int], optional): Number of messages to retrieve. Will add the last summary generated prior to the nth oldest message. Defaults to 669                output_key (Optional[str], optional): The key to use for the output message.70                                                  Defaults to None.71                input_key (Optional[str], optional): The key to use for the input message.72                                                  Defaults to None.73                return_messages (bool, optional): Does your prompt template expect a string74                                                  or a list of Messages? Defaults to False75                                                  i.e. return a string.76                human_prefix (str, optional): The prefix to use for human messages.77                                              Defaults to "Human".78                ai_prefix (str, optional): The prefix to use for AI messages.79                                           Defaults to "AI".80                memory_key (str, optional): The key to use for the memory.81                                            Defaults to "history".82                                            Ensure that this matches the key used in83                                            chain's prompt template.84            """  # noqa: E50185            chat_message_history = ZepCloudChatMessageHistory(86                session_id=session_id,87                memory_type=memory_type,88                lastn=lastn,89                api_key=api_key,90            )91            super().__init__(92                chat_memory=chat_message_history,93                output_key=output_key,94                input_key=input_key,95                return_messages=return_messages,96                human_prefix=human_prefix,97                ai_prefix=ai_prefix,98                memory_key=memory_key,99            )100 101        def save_context(102            self,103            inputs: Dict[str, Any],104            outputs: Dict[str, str],105            metadata: Optional[Dict[str, Any]] = None,106        ) -> None:107            """Save context from this conversation to buffer.108 109            Args:110                inputs (Dict[str, Any]): The inputs to the chain.111                outputs (Dict[str, str]): The outputs from the chain.112                metadata (Optional[Dict[str, Any]], optional): Any metadata to save with113                                                           the context. Defaults to None114 115            Returns:116                None117            """118            input_str, output_str = self._get_input_output(inputs, outputs)119            self.chat_memory.add_user_message(input_str, metadata=metadata)120            self.chat_memory.add_ai_message(output_str, metadata=metadata)121 122except ImportError:123    # Placeholder object124    class ZepCloudMemory:  # type: ignore[no-redef]125        pass126