Source code for langchain_core.chat_history

"""**Chat message history** stores a history of the message interactions in a chat.


**Class hierarchy:**

.. code-block::

    BaseChatMessageHistory --> <name>ChatMessageHistory  # Examples: FileChatMessageHistory, PostgresChatMessageHistory

**Main helpers:**

.. code-block::

    AIMessage, HumanMessage, BaseMessage

"""  # noqa: E501
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import List, Sequence, Union

from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
    get_buffer_string,
)
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import run_in_executor


[docs]class BaseChatMessageHistory(ABC): """Abstract base class for storing chat message history. Implementations guidelines: Implementations are expected to over-ride all or some of the following methods: * add_messages: sync variant for bulk addition of messages * aadd_messages: async variant for bulk addition of messages * messages: sync variant for getting messages * aget_messages: async variant for getting messages * clear: sync variant for clearing messages * aclear: async variant for clearing messages add_messages contains a default implementation that calls add_message for each message in the sequence. This is provided for backwards compatibility with existing implementations which only had add_message. Async variants all have default implementations that call the sync variants. Implementers can choose to over-ride the async implementations to provide truly async implementations. Usage guidelines: When used for updating history, users should favor usage of `add_messages` over `add_message` or other variants like `add_user_message` and `add_ai_message` to avoid unnecessary round-trips to the underlying persistence layer. Example: Shows a default implementation. .. code-block:: python class FileChatMessageHistory(BaseChatMessageHistory): storage_path: str session_id: str @property def messages(self): with open(os.path.join(storage_path, session_id), 'r:utf-8') as f: messages = json.loads(f.read()) return messages_from_dict(messages) def add_messages(self, messages: Sequence[BaseMessage]) -> None: all_messages = list(self.messages) # Existing messages all_messages.extend(messages) # Add new messages serialized = [message_to_dict(message) for message in all_messages] # Can be further optimized by only writing new messages # using append mode. with open(os.path.join(storage_path, session_id), 'w') as f: json.dump(f, messages) def clear(self): with open(os.path.join(storage_path, session_id), 'w') as f: f.write("[]") """ messages: List[BaseMessage] """A property or attribute that returns a list of messages. In general, getting the messages may involve IO to the underlying persistence layer, so this operation is expected to incur some latency. """
[docs] async def aget_messages(self) -> List[BaseMessage]: """Async version of getting messages. Can over-ride this method to provide an efficient async implementation. In general, fetching messages may involve IO to the underlying persistence layer. """ return await run_in_executor(None, lambda: self.messages)
[docs] def add_user_message(self, message: Union[HumanMessage, str]) -> None: """Convenience method for adding a human message string to the store. Please note that this is a convenience method. Code should favor the bulk add_messages interface instead to save on round-trips to the underlying persistence layer. This method may be deprecated in a future release. Args: message: The human message to add """ if isinstance(message, HumanMessage): self.add_message(message) else: self.add_message(HumanMessage(content=message))
[docs] def add_ai_message(self, message: Union[AIMessage, str]) -> None: """Convenience method for adding an AI message string to the store. Please note that this is a convenience method. Code should favor the bulk add_messages interface instead to save on round-trips to the underlying persistence layer. This method may be deprecated in a future release. Args: message: The AI message to add. """ if isinstance(message, AIMessage): self.add_message(message) else: self.add_message(AIMessage(content=message))
[docs] def add_message(self, message: BaseMessage) -> None: """Add a Message object to the store. Args: message: A BaseMessage object to store. """ if type(self).add_messages != BaseChatMessageHistory.add_messages: # This means that the sub-class has implemented an efficient add_messages # method, so we should use it. self.add_messages([message]) else: raise NotImplementedError( "add_message is not implemented for this class. " "Please implement add_message or add_messages." )
[docs] def add_messages(self, messages: Sequence[BaseMessage]) -> None: """Add a list of messages. Implementations should over-ride this method to handle bulk addition of messages in an efficient manner to avoid unnecessary round-trips to the underlying store. Args: messages: A list of BaseMessage objects to store. """ for message in messages: self.add_message(message)
[docs] async def aadd_messages(self, messages: Sequence[BaseMessage]) -> None: """Add a list of messages. Args: messages: A list of BaseMessage objects to store. """ await run_in_executor(None, self.add_messages, messages)
[docs] @abstractmethod def clear(self) -> None: """Remove all messages from the store"""
[docs] async def aclear(self) -> None: """Remove all messages from the store""" await run_in_executor(None, self.clear)
def __str__(self) -> str: """Return a string representation of the chat history.""" return get_buffer_string(self.messages)
[docs]class InMemoryChatMessageHistory(BaseChatMessageHistory, BaseModel): """In memory implementation of chat message history. Stores messages in an in memory list. """ messages: List[BaseMessage] = Field(default_factory=list)
[docs] async def aget_messages(self) -> List[BaseMessage]: return self.messages
[docs] def add_message(self, message: BaseMessage) -> None: """Add a self-created message to the store""" self.messages.append(message)
[docs] async def aadd_messages(self, messages: Sequence[BaseMessage]) -> None: """Add messages to the store""" self.add_messages(messages)
[docs] def clear(self) -> None: self.messages = []
[docs] async def aclear(self) -> None: self.clear()