Module zep_python.langchain.history

Classes

class ZepChatMessageHistory (session_id: str, zep_client: Optional[ZepClient] = None, api_url: Optional[str] = 'https://api.getzep.com', api_key: Optional[str] = None, memory_type: Optional[str] = None, ai_prefix: Optional[str] = None, human_prefix: Optional[str] = None, summary_instruction: Optional[str] = None)

LangChain Chat message history that uses Zep as a backend.

Attributes

session_id : str

The unique identifier of the session.

zep_client : ZepClient

The Zep client used for making API requests. Pass in this rather than the API key and URL.

api_url : str

The Zep API service URL. Not required if using Zep Cloud.

api_key : str

The Zep API key. Not required if using Zep Open Source.

memory_type : str

The type of memory to use. Can be "perpetual", "summary_retrieval", or "message_window". Defaults to "perpetual".

summary_instruction : Optional[str]

Additional instructions for generating dialog summaries.

Expand source code

class ZepChatMessageHistory(BaseChatMessageHistory):
    """
    LangChain Chat message history that uses Zep as a backend.

    Attributes
    ----------
    session_id : str
        The unique identifier of the session.
    zep_client : ZepClient
        The Zep client used for making API requests.
        Pass in this rather than the API key and URL.
    api_url : str
        The Zep API service URL. Not required if using Zep Cloud.
    api_key : str
        The Zep API key. Not required if using Zep Open Source.
    memory_type : str
        The type of memory to use. Can be "perpetual", "summary_retrieval",
        or "message_window". Defaults to "perpetual".
    summary_instruction : Optional[str]
        Additional instructions for generating dialog summaries.
    """

    def __init__(
        self,
        session_id: str,
        zep_client: Optional[ZepClient] = None,
        api_url: Optional[str] = API_URL,
        api_key: Optional[str] = None,
        memory_type: Optional[str] = None,
        ai_prefix: Optional[str] = None,
        human_prefix: Optional[str] = None,
        summary_instruction: Optional[str] = None,
    ) -> None:
        if zep_client is None:
            self._client = ZepClient(api_url=api_url, api_key=api_key)
        else:
            self._client = zep_client

        self.session_id = session_id
        self.memory_type = memory_type or "perpetual"

        self.ai_prefix = ai_prefix or "ai"
        self.human_prefix = human_prefix or "human"
        self.summary_instruction = summary_instruction

    @property
    def messages(self) -> List[BaseMessage]:  # type: ignore
        """Retrieve messages from Zep memory"""

        zep_memory: Optional[Memory] = self._get_memory()
        if not zep_memory:
            return []

        messages: List[BaseMessage] = []
        # Extract facts and summary, if present, and messages
        if zep_memory.facts:
            messages.append(SystemMessage(content="\n".join(zep_memory.facts)))

        if zep_memory.summary:
            if len(zep_memory.summary.content) > 0:
                messages.append(SystemMessage(content=zep_memory.summary.content))

        if zep_memory.messages:
            for msg in zep_memory.messages:
                metadata = {
                    "uuid": msg.uuid,
                    "created_at": msg.created_at,
                    "token_count": msg.token_count,
                    "metadata": msg.metadata,
                }
                message_class = AIMessage if msg.role == "ai" else HumanMessage
                messages.append(
                    message_class(content=msg.content, additional_kwargs=metadata)
                )

        return messages

    @property
    def zep_messages(self) -> Union[List[Message], None]:
        """Retrieve summary from Zep memory"""
        zep_memory: Optional[Memory] = self._get_memory()
        if not zep_memory:
            return []

        return zep_memory.messages

    @property
    def zep_summary(self) -> Optional[str]:
        """Retrieve summary from Zep memory"""
        zep_memory: Optional[Memory] = self._get_memory()
        if not zep_memory or not zep_memory.summary:
            return None

        return zep_memory.summary.content

    def _get_memory(self) -> Optional[Memory]:
        """Retrieve memory from Zep"""
        try:
            zep_memory: Memory = self._client.memory.get_memory(
                self.session_id, self.memory_type
            )
        except NotFoundError:
            logger.warning(
                f"Session {self.session_id} not found in Zep. Returning None"
            )
            return None
        return zep_memory

    def add_user_message(  # type: ignore
        self, message: str, metadata: Optional[Dict[str, Any]] = None
    ) -> None:
        """Convenience method for adding a human message string to the store.

        Args:
            message: The string contents of a human message.
            metadata: Optional metadata to attach to the message.
        """
        from langchain_core.messages import HumanMessage

        self.add_message(HumanMessage(content=message), metadata=metadata)

    def add_ai_message(  # type: ignore
        self, message: str, metadata: Optional[Dict[str, Any]] = None
    ) -> None:
        """Convenience method for adding an AI message string to the store.

        Args:
            message: The string contents of an AI message.
            metadata: Optional metadata to attach to the message.
        """
        from langchain_core.messages import AIMessage

        self.add_message(AIMessage(content=message), metadata=metadata)

    def add_message(
        self, message: BaseMessage, metadata: Optional[Dict[str, Any]] = None
    ) -> None:
        """Append the message to the Zep memory history"""

        if message.content is None:
            raise ValueError("Message content cannot be None")

        if isinstance(message.content, list):
            raise ValueError("Message content cannot be a list")

        if message.type == "ai":
            message.name = self.ai_prefix
        elif message.type == "human":
            message.name = self.human_prefix

        zep_message = Message(
            content=message.content,
            # If name is not set, use type as role
            role=message.name or message.type,
            role_type=get_zep_message_role_type(message.type),
            metadata=metadata,
        )
        zep_memory = Memory(
            messages=[zep_message], summary_instruction=self.summary_instruction
        )

        self._client.memory.add_memory(self.session_id, zep_memory)

    def clear(self) -> None:
        """Clear session memory from Zep. Note that Zep is long-term storage for memory
        and this is not advised unless you have specific data retention requirements.
        """
        try:
            self._client.memory.delete_memory(self.session_id)
        except NotFoundError:
            logger.warning(
                f"Session {self.session_id} not found in Zep. Skipping delete."
            )

Ancestors

• langchain_core.chat_history.BaseChatMessageHistory
• abc.ABC

Instance variables

var messages : List[langchain_core.messages.base.BaseMessage]

Retrieve messages from Zep memory

Expand source code

@property
def messages(self) -> List[BaseMessage]:  # type: ignore
    """Retrieve messages from Zep memory"""

    zep_memory: Optional[Memory] = self._get_memory()
    if not zep_memory:
        return []

    messages: List[BaseMessage] = []
    # Extract facts and summary, if present, and messages
    if zep_memory.facts:
        messages.append(SystemMessage(content="\n".join(zep_memory.facts)))

    if zep_memory.summary:
        if len(zep_memory.summary.content) > 0:
            messages.append(SystemMessage(content=zep_memory.summary.content))

    if zep_memory.messages:
        for msg in zep_memory.messages:
            metadata = {
                "uuid": msg.uuid,
                "created_at": msg.created_at,
                "token_count": msg.token_count,
                "metadata": msg.metadata,
            }
            message_class = AIMessage if msg.role == "ai" else HumanMessage
            messages.append(
                message_class(content=msg.content, additional_kwargs=metadata)
            )

    return messages

var zep_messages : Optional[List[Message]]

Retrieve summary from Zep memory

Expand source code

@property
def zep_messages(self) -> Union[List[Message], None]:
    """Retrieve summary from Zep memory"""
    zep_memory: Optional[Memory] = self._get_memory()
    if not zep_memory:
        return []

    return zep_memory.messages

var zep_summary : Optional[str]

Retrieve summary from Zep memory

Expand source code

@property
def zep_summary(self) -> Optional[str]:
    """Retrieve summary from Zep memory"""
    zep_memory: Optional[Memory] = self._get_memory()
    if not zep_memory or not zep_memory.summary:
        return None

    return zep_memory.summary.content

Methods

def add_ai_message(self, message: str, metadata: Optional[Dict[str, Any]] = None) ‑> None

Convenience method for adding an AI message string to the store.

Args

message

The string contents of an AI message.

metadata

Optional metadata to attach to the message.

Expand source code

def add_ai_message(  # type: ignore
    self, message: str, metadata: Optional[Dict[str, Any]] = None
) -> None:
    """Convenience method for adding an AI message string to the store.

    Args:
        message: The string contents of an AI message.
        metadata: Optional metadata to attach to the message.
    """
    from langchain_core.messages import AIMessage

    self.add_message(AIMessage(content=message), metadata=metadata)

def add_message(self, message: BaseMessage, metadata: Optional[Dict[str, Any]] = None) ‑> None

Append the message to the Zep memory history

Expand source code

def add_message(
    self, message: BaseMessage, metadata: Optional[Dict[str, Any]] = None
) -> None:
    """Append the message to the Zep memory history"""

    if message.content is None:
        raise ValueError("Message content cannot be None")

    if isinstance(message.content, list):
        raise ValueError("Message content cannot be a list")

    if message.type == "ai":
        message.name = self.ai_prefix
    elif message.type == "human":
        message.name = self.human_prefix

    zep_message = Message(
        content=message.content,
        # If name is not set, use type as role
        role=message.name or message.type,
        role_type=get_zep_message_role_type(message.type),
        metadata=metadata,
    )
    zep_memory = Memory(
        messages=[zep_message], summary_instruction=self.summary_instruction
    )

    self._client.memory.add_memory(self.session_id, zep_memory)

def add_user_message(self, message: str, metadata: Optional[Dict[str, Any]] = None) ‑> None

Convenience method for adding a human message string to the store.

Args

message

The string contents of a human message.

metadata

Optional metadata to attach to the message.

Expand source code

def add_user_message(  # type: ignore
    self, message: str, metadata: Optional[Dict[str, Any]] = None
) -> None:
    """Convenience method for adding a human message string to the store.

    Args:
        message: The string contents of a human message.
        metadata: Optional metadata to attach to the message.
    """
    from langchain_core.messages import HumanMessage

    self.add_message(HumanMessage(content=message), metadata=metadata)

def clear(self) ‑> None

Clear session memory from Zep. Note that Zep is long-term storage for memory and this is not advised unless you have specific data retention requirements.

Expand source code

def clear(self) -> None:
    """Clear session memory from Zep. Note that Zep is long-term storage for memory
    and this is not advised unless you have specific data retention requirements.
    """
    try:
        self._client.memory.delete_memory(self.session_id)
    except NotFoundError:
        logger.warning(
            f"Session {self.session_id} not found in Zep. Skipping delete."
        )