Skip to content

Redis conversation memory driver

RedisConversationMemoryDriver

Bases: BaseConversationMemoryDriver

A Conversation Memory Driver for Redis.

This driver interfaces with a Redis instance and utilizes the Redis hashes and RediSearch module to store, retrieve, and query conversations in a structured manner. Proper setup of the Redis instance and RediSearch is necessary for the driver to function correctly.

Attributes:

Name Type Description
host str

The host of the Redis instance.
port int

The port of the Redis instance.
db int

The database of the Redis instance.
password Optional[str]

The password of the Redis instance.
index str

The name of the index to use.
conversation_id str

The id of the conversation.
Source code in griptape/drivers/memory/conversation/redis_conversation_memory_driver.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@define
class RedisConversationMemoryDriver(BaseConversationMemoryDriver):
    """A Conversation Memory Driver for Redis.

    This driver interfaces with a Redis instance and utilizes the Redis hashes and RediSearch module to store,
    retrieve, and query conversations in a structured manner.
    Proper setup of the Redis instance and RediSearch is necessary for the driver to function correctly.

    Attributes:
        host: The host of the Redis instance.
        port: The port of the Redis instance.
        db: The database of the Redis instance.
        password: The password of the Redis instance.
        index: The name of the index to use.
        conversation_id: The id of the conversation.
    """

    host: str = field(kw_only=True, metadata={"serializable": True})
    port: int = field(kw_only=True, metadata={"serializable": True})
    db: int = field(kw_only=True, default=0, metadata={"serializable": True})
    password: Optional[str] = field(default=None, kw_only=True, metadata={"serializable": False})
    index: str = field(kw_only=True, metadata={"serializable": True})
    conversation_id: str = field(kw_only=True, default=uuid.uuid4().hex)

    client: Redis = field(
        default=Factory(
            lambda self: import_optional_dependency("redis").Redis(
                host=self.host, port=self.port, db=self.db, password=self.password, decode_responses=False
            ),
            takes_self=True,
        )
    )

    def store(self, memory: BaseConversationMemory) -> None:
        self.client.hset(self.index, self.conversation_id, memory.to_json())

    def load(self) -> Optional[BaseConversationMemory]:
        key = self.index
        memory_json = self.client.hget(key, self.conversation_id)
        if memory_json:
            memory = BaseConversationMemory.from_json(memory_json)
            memory.driver = self
            return memory
        return None

client: Redis = field(default=Factory(lambda self: import_optional_dependency('redis').Redis(host=self.host, port=self.port, db=self.db, password=self.password, decode_responses=False), takes_self=True)) class-attribute instance-attribute

conversation_id: str = field(kw_only=True, default=uuid.uuid4().hex) class-attribute instance-attribute

db: int = field(kw_only=True, default=0, metadata={'serializable': True}) class-attribute instance-attribute

host: str = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

index: str = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

password: Optional[str] = field(default=None, kw_only=True, metadata={'serializable': False}) class-attribute instance-attribute

port: int = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

load()

Source code in griptape/drivers/memory/conversation/redis_conversation_memory_driver.py 
49
50
51
52
53
54
55
56
def load(self) -> Optional[BaseConversationMemory]:
    key = self.index
    memory_json = self.client.hget(key, self.conversation_id)
    if memory_json:
        memory = BaseConversationMemory.from_json(memory_json)
        memory.driver = self
        return memory
    return None

store(memory)

Source code in griptape/drivers/memory/conversation/redis_conversation_memory_driver.py 
46
47
def store(self, memory: BaseConversationMemory) -> None:
    self.client.hset(self.index, self.conversation_id, memory.to_json())