BaseChatMessageHistory Class — langchain Architecture

Source Code

libs/core/langchain_core/chat_history.py lines 22–199

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 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:
        ```python
        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)
        ```
    """

    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.
    """

    async def aget_messages(self) -> list[BaseMessage]:
        """Async version of getting messages.

        Can over-ride this method to provide an efficient async implementation.