redis

This module contains the RedisMessageHistory class, which is used to store chat message history in a Redis database.

Classes:

Name	Description
RedisMessageHistory	Chat message history that stores history in a Redis database.

Attributes:

Name	Type	Description
DEFAULT_TABLE_NAME		A table name for the Redis database.
DEFAULT_URL		A URL for the Redis database.

DEFAULT_TABLE_NAME module-attribute

DEFAULT_TABLE_NAME = 'message_store'

A table name for the Redis database.

DEFAULT_URL module-attribute

DEFAULT_URL = 'redis://localhost:6379/0'

A URL for the Redis database.

RedisMessageHistory

RedisMessageHistory(
    session_id: str,
    url: str = DEFAULT_URL,
    key_prefix: str = f"{DEFAULT_TABLE_NAME}:",
    ttl: Optional[int] = None,
)

Bases: BaseChatMessageHistory

Chat message history that stores history in a Redis database.

Parameters:

Name	Type	Description	Default
session_id	str	Arbitrary key that is used to store the messages for a single chat session.	required
url	str	URL to connect to the Redis server.	DEFAULT_URL
key_prefix	str	Prefix for the Redis key.	f'{DEFAULT_TABLE_NAME}:'
ttl	Optional[int]	Time-to-live for the message records.	None

Methods:

Name	Description
add_message	Append the message to the record in Redis
clear	Clear session memory from Redis

Attributes:

Name	Type	Description
history	List[Message]	Retrieve the messages from Redis
key	str	Construct the record key to use

Source code in src/declarai/memory/redis.py

def __init__(
    self,
    session_id: str,
    url: str = DEFAULT_URL,
    key_prefix: str = f"{DEFAULT_TABLE_NAME}:",
    ttl: Optional[int] = None,
):
    super().__init__()
    try:
        import redis  # pylint: disable=import-outside-toplevel
    except ImportError:
        raise ImportError(
            "Could not import redis python package. "
            "Please install it with `pip install redis`."
        )

    self.redis_client = redis.StrictRedis.from_url(url)
    self.session_id = session_id
    self.key_prefix = key_prefix
    self.ttl = ttl

history property

history: List[Message]

Retrieve the messages from Redis

key property

key: str

Construct the record key to use

add_message

add_message(message: Message) -> None

Append the message to the record in Redis

Source code in src/declarai/memory/redis.py

def add_message(self, message: Message) -> None:
    """Append the message to the record in Redis"""
    self.redis_client.lpush(self.key, json.dumps(message.dict()))
    if self.ttl:
        self.redis_client.expire(self.key, self.ttl)

clear

clear() -> None

Clear session memory from Redis

Source code in src/declarai/memory/redis.py

def clear(self) -> None:
    """Clear session memory from Redis"""
    self.redis_client.delete(self.key)