Skip to main content
Version: 2.19

ChatMessage Store

Module haystack_experimental.chat_message_stores.in_memory​

InMemoryChatMessageStore​

Stores chat messages in-memory.

The chat_history_id parameter is used as a unique identifier for each conversation or chat session. It acts as a namespace that isolates messages from different sessions. Each chat_history_id value corresponds to a separate list of ChatMessage objects stored in memory.

Typical usage involves providing a unique chat_history_id (for example, a session ID or conversation ID) whenever you write, read, or delete messages. This ensures that chat messages from different conversations do not overlap.

Usage example:

python
from haystack.dataclasses import ChatMessage
from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore

message_store = InMemoryChatMessageStore()

messages = [
    ChatMessage.from_assistant("Hello, how can I help you?"),
    ChatMessage.from_user("Hi, I have a question about Python. What is a Protocol?"),
]
message_store.write_messages(chat_history_id="user_456_session_123", messages=messages)
retrieved_messages = message_store.retrieve_messages(chat_history_id="user_456_session_123")

print(retrieved_messages)

InMemoryChatMessageStore.__init__​

python
def __init__(skip_system_messages: bool = True,
             last_k: Optional[int] = 10) -> None

Create an InMemoryChatMessageStore.

Arguments:

  • skip_system_messages: Whether to skip storing system messages. Defaults to True.
  • last_k: The number of last messages to retrieve. Defaults to 10 messages if not specified.

InMemoryChatMessageStore.to_dict​

python
def to_dict() -> dict[str, Any]

Serializes the component to a dictionary.

Returns:

Dictionary with serialized data.

InMemoryChatMessageStore.from_dict​

python
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "InMemoryChatMessageStore"

Deserializes the component from a dictionary.

Arguments:

  • data: The dictionary to deserialize from.

Returns:

The deserialized component.

InMemoryChatMessageStore.count_messages​

python
def count_messages(chat_history_id: str) -> int

Returns the number of chat messages stored in this store.

Arguments:

  • chat_history_id: The chat history id for which to count messages.

Returns:

The number of messages.

InMemoryChatMessageStore.write_messages​

python
def write_messages(chat_history_id: str, messages: list[ChatMessage]) -> int

Writes chat messages to the ChatMessageStore.

Arguments:

  • chat_history_id: The chat history id under which to store the messages.
  • messages: A list of ChatMessages to write.

Raises:

  • ValueError: If messages is not a list of ChatMessages.

Returns:

The number of messages written.

InMemoryChatMessageStore.retrieve_messages​

python
def retrieve_messages(chat_history_id: str,
                      last_k: Optional[int] = None) -> list[ChatMessage]

Retrieves all stored chat messages.

Arguments:

  • chat_history_id: The chat history id from which to retrieve messages.
  • last_k: The number of last messages to retrieve. If unspecified, the last_k parameter passed to the constructor will be used.

Raises:

  • ValueError: If last_k is not None and is less than 0.

Returns:

A list of chat messages.

InMemoryChatMessageStore.delete_messages​

python
def delete_messages(chat_history_id: str) -> None

Deletes all stored chat messages.

Arguments:

  • chat_history_id: The chat history id from which to delete messages.

InMemoryChatMessageStore.delete_all_messages​

python
def delete_all_messages() -> None

Deletes all stored chat messages from all chat history ids.