redis

redis#

  • __all__ = ['RedisConversationMemoryDriver'] module-attribute

Bases: BaseConversationMemoryDriver “)

Attributes#

Name

Type

Description

<a href="redis.mdx#griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.host" title="host = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute (griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.host)"><code class="language-python">host</code></a>

str

The host of the Redis instance.<br/>

<a href="redis.mdx#griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.port" title="port = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute (griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.port)"><code class="language-python">port</code></a>

int

The port of the Redis instance.<br/>

<a href="redis.mdx#griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.db" title="db = field(kw_only=True, default=0, metadata={'serializable': True}) class-attribute instance-attribute (griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.db)"><code class="language-python">db</code></a>

int

The database of the Redis instance.<br/>

<a href="redis.mdx#griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.username" title="username = field(kw_only=True, default='default', metadata={'serializable': False}) class-attribute instance-attribute (griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.username)"><code class="language-python">username</code></a>

str

The username of the Redis instance.<br/>

<a href="redis.mdx#griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.password" title="password = field(default=None, kw_only=True, metadata={'serializable': False}) class-attribute instance-attribute (griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.password)"><code class="language-python">password</code></a>

Optional[str]

The password of the Redis instance.<br/>

<a href="redis.mdx#griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.index" title="index = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute (griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.index)"><code class="language-python">index</code></a>

str

The name of the index to use.<br/>

<a href="redis.mdx#griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.conversation_id" title="conversation_id = field(kw_only=True, default=uuid.uuid4().hex) class-attribute instance-attribute (griptape.drivers.memory.conversation.redis.RedisConversationMemoryDriver.conversation_id)"><code class="language-python">conversation_id</code></a>

str

The id of the conversation.<br/>

Source Code in griptape/drivers/memory/conversation/redis_conversation_memory_driver.py

@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

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

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)))
Contents