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 messagesaadd_messages: async variant for bulk addition of messagesmessages: sync variant for getting messagesaget_messages: async variant for getting messagesclear: sync variant for clearing messagesaclear: async variant for clearing messagesadd_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 override 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:
import json
import os
from langchain_core.messages import messages_from_dict, message_to_dict
class FileChatMessageHistory(BaseChatMessageHistory):
storage_path: str
session_id: str
@property
def messages(self) -> list[BaseMessage]:
try:
with open(
os.path.join(self.storage_path, self.session_id),
"r",
encoding="utf-8",
) as f:
messages_data = json.load(f)
return messages_from_dict(messages_data)
except FileNotFoundError:
return []
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]
file_path = os.path.join(self.storage_path, self.session_id)
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, "w", encoding="utf-8") as f:
json.dump(serialized, f)
def clear(self) -> None:
file_path = os.path.join(self.storage_path, self.session_id)
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, "w", encoding="utf-8") as f:
json.dump([], f)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.
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.
Convenience method for adding a human message string to the store.
This is a convenience method. Code should favor the bulk add_messages
interface instead to save on round-trips to the persistence layer.
This method may be deprecated in a future release.
Convenience method for adding an AIMessage string to the store.
This is a convenience method. Code should favor the bulk add_messages
interface instead to save on round-trips to the persistence layer.
This method may be deprecated in a future release.
Add a Message object to the store.
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.
Async add a list of messages.
Remove all messages from the store.
Async remove all messages from the store.