RedisEntityStore Class — langchain Architecture

Architecture documentation for the RedisEntityStore class in entity.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  c539be70_4b09_e675_2ee4_a53dadca5460["RedisEntityStore"]
  d6a371b0_afe3_1ceb_cd58_c254c31b562e["BaseEntityStore"]
  c539be70_4b09_e675_2ee4_a53dadca5460 -->|extends| d6a371b0_afe3_1ceb_cd58_c254c31b562e
  1ed59830_902e_f3ce_f9f6_76963a6fd122["entity.py"]
  c539be70_4b09_e675_2ee4_a53dadca5460 -->|defined in| 1ed59830_902e_f3ce_f9f6_76963a6fd122
  15f72ac1_48c7_e4d4_46d0_ef4e3a39c901["__init__()"]
  c539be70_4b09_e675_2ee4_a53dadca5460 -->|method| 15f72ac1_48c7_e4d4_46d0_ef4e3a39c901
  c791716d_3809_7b67_d208_a5c3252d0cc0["full_key_prefix()"]
  c539be70_4b09_e675_2ee4_a53dadca5460 -->|method| c791716d_3809_7b67_d208_a5c3252d0cc0
  09510174_d954_7a6b_5dbb_826011d7072f["get()"]
  c539be70_4b09_e675_2ee4_a53dadca5460 -->|method| 09510174_d954_7a6b_5dbb_826011d7072f
  2e114e1c_3610_cc07_2105_0273d9788c9e["set()"]
  c539be70_4b09_e675_2ee4_a53dadca5460 -->|method| 2e114e1c_3610_cc07_2105_0273d9788c9e
  243a8f75_3643_df4b_c406_a947cdc7d0af["delete()"]
  c539be70_4b09_e675_2ee4_a53dadca5460 -->|method| 243a8f75_3643_df4b_c406_a947cdc7d0af
  50bc4ae1_8e25_46d8_33bc_109fb37f2ff8["exists()"]
  c539be70_4b09_e675_2ee4_a53dadca5460 -->|method| 50bc4ae1_8e25_46d8_33bc_109fb37f2ff8
  c0299cc2_b89a_c73d_6f9e_8a2e27823eb4["clear()"]
  c539be70_4b09_e675_2ee4_a53dadca5460 -->|method| c0299cc2_b89a_c73d_6f9e_8a2e27823eb4

Relationship Graph

Source Code

libs/langchain/langchain_classic/memory/entity.py lines 219–332

class RedisEntityStore(BaseEntityStore):
    """Redis-backed Entity store.

    Entities get a TTL of 1 day by default, and
    that TTL is extended by 3 days every time the entity is read back.
    """

    redis_client: Any
    session_id: str = "default"
    key_prefix: str = "memory_store"
    ttl: int | None = 60 * 60 * 24
    recall_ttl: int | None = 60 * 60 * 24 * 3

    def __init__(
        self,
        session_id: str = "default",
        url: str = "redis://localhost:6379/0",
        key_prefix: str = "memory_store",
        ttl: int | None = 60 * 60 * 24,
        recall_ttl: int | None = 60 * 60 * 24 * 3,
        *args: Any,
        **kwargs: Any,
    ):
        """Initializes the RedisEntityStore.

        Args:
            session_id: Unique identifier for the session.
            url: URL of the Redis server.
            key_prefix: Prefix for keys in the Redis store.
            ttl: Time-to-live for keys in seconds (default 1 day).
            recall_ttl: Time-to-live extension for keys when recalled (default 3 days).
            *args: Additional positional arguments.
            **kwargs: Additional keyword arguments.
        """
        try:
            import redis
        except ImportError as e:
            msg = (
                "Could not import redis python package. "
                "Please install it with `pip install redis`."
            )
            raise ImportError(msg) from e

        super().__init__(*args, **kwargs)

        try:
            from langchain_community.utilities.redis import get_client
        except ImportError as e:
            msg = (
                "Could not import langchain_community.utilities.redis.get_client. "
                "Please install it with `pip install langchain-community`."
            )
            raise ImportError(msg) from e

        try:
            self.redis_client = get_client(redis_url=url, decode_responses=True)
        except redis.exceptions.ConnectionError:
            logger.exception("Redis client could not connect")

        self.session_id = session_id
        self.key_prefix = key_prefix
        self.ttl = ttl
        self.recall_ttl = recall_ttl or ttl

    @property
    def full_key_prefix(self) -> str:
        """Returns the full key prefix with session ID."""
        return f"{self.key_prefix}:{self.session_id}"

    @override
    def get(self, key: str, default: str | None = None) -> str | None:
        res = (
            self.redis_client.getex(f"{self.full_key_prefix}:{key}", ex=self.recall_ttl)
            or default
            or ""
        )
        logger.debug("REDIS MEM get '%s:%s': '%s'", self.full_key_prefix, key, res)
        return res

    @override
    def set(self, key: str, value: str | None) -> None: