Skip to content

redis

__all__ = ['RedisConversationMemoryDriver'] module-attribute

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.
username str

The username 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 
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
57
58
59
60
61
62
63
64
65
@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.
        username: The username 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})
    username: str = field(kw_only=True, default="default", metadata={"serializable": False})
    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,
                username=self.username,
                password=self.password,
                decode_responses=False,
            ),
            takes_self=True,
        ),
    )

    def store(self, runs: list[Run], metadata: dict[str, Any]) -> None:
        self.client.hset(self.index, self.conversation_id, json.dumps(self._to_params_dict(runs, metadata)))

    def load(self) -> tuple[list[Run], dict[str, Any]]:
        memory_json = self.client.hget(self.index, self.conversation_id)
        if memory_json is not None:
            return self._from_params_dict(json.loads(memory_json))  # pyright: ignore[reportArgumentType] https://github.com/redis/redis-py/issues/2399
        return [], {}

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

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

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

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

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

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

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

username = field(kw_only=True, default='default', metadata={'serializable': False}) class-attribute instance-attribute

load()

Source code in griptape/drivers/memory/conversation/redis_conversation_memory_driver.py 
61
62
63
64
65
def load(self) -> tuple[list[Run], dict[str, Any]]:
    memory_json = self.client.hget(self.index, self.conversation_id)
    if memory_json is not None:
        return self._from_params_dict(json.loads(memory_json))  # pyright: ignore[reportArgumentType] https://github.com/redis/redis-py/issues/2399
    return [], {}

store(runs, metadata)

Source code in griptape/drivers/memory/conversation/redis_conversation_memory_driver.py 
58
59
def store(self, runs: list[Run], metadata: dict[str, Any]) -> None:
    self.client.hset(self.index, self.conversation_id, json.dumps(self._to_params_dict(runs, metadata)))