langgraph/how-tos/persistence_redis/

[giscus[bot]]

langgraph/how-tos/persistence_redis/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/how-tos/persistence_redis/

[slgabbar]

I ran into an issue where my graph was not always grabbing the most recent checkpointed state.

Upon investigation I found that the thread_ts looks something like "2024-07-24T09:00:00.000000+00:00".

I think there is an issue with the logic to grab the latest key in aget_tuple. We are splitting the key on a semicolon and comparing just the last part of the key, which, based on the timestamp format will always just compare "00" to "00" (assuming timestamps are UTC):
latest_key = max(all_keys, key=lambda k: k.decode().split(":")[-1])

Wouldn't we want to compare the entire timestamp string? something like:
latest_key = max(all_keys, key=lambda k: ":".join(k.decode().split(":")[2:]))

[OmarHory]

I have the same problem. It is a bit annoying, i tried your solution, works well!

Also newer versions of langgraph ( try to update to the latest version of 0.1.14) needs put_writes and aput_writes functions for the Redis, in which I have written them.

[dzubke]

Not really answering your question, but I wanted to clarify that the thread_ts looks something like: "1ef4ca79-8147-61f2-8300-38d4cd2a59d0". The checkpoint["ts"] value is the timestamp that you mentioned above: "2024-07-24T09:00:00.000000+00:00".

Just wanted to add for other's reference as this isn't yet that well documented.

[dzubke]

From my limited testing, sorting by thread_ts and checkpoint["ts"] results in the same sort order.

[vbarda]

Please see the latest version of the how-to -- it doesn't rely on timestamps at all and uses checkpoint_id UUID instead!

[ishanExtreme]

The one given in the above guide have a bug in "put/aput" function I have changed the code a little bit with split changed to "$"

"""Implementation of a langgraph checkpoint saver using Redis."""

from contextlib import asynccontextmanager, contextmanager
from typing import Any, AsyncGenerator, Generator, List, Union, Tuple, Optional

import redis
from redis.asyncio import Redis as AsyncRedis, ConnectionPool as AsyncConnectionPool
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint import BaseCheckpointSaver
from langgraph.serde.jsonplus import JsonPlusSerializer
from langgraph.checkpoint.base import Checkpoint, CheckpointMetadata, CheckpointTuple
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class JsonAndBinarySerializer(JsonPlusSerializer):
    def _default(self, obj: Any) -> Any:
        if isinstance(obj, (bytes, bytearray)):
            return self._encode_constructor_args(
                obj.__class__, method="fromhex", args=[obj.hex()]
            )
        return super()._default(obj)

    def dumps(self, obj: Any) -> str:
        try:
            if isinstance(obj, (bytes, bytearray)):
                return obj.hex()
            return super().dumps(obj)
        except Exception as e:
            logger.error(f"Serialization error: {e}")
            raise

    def loads(self, s: str, is_binary: bool = False) -> Any:
        try:
            if is_binary:
                return bytes.fromhex(s)
            return super().loads(s)
        except Exception as e:
            logger.error(f"Deserialization error: {e}")
            raise


def initialize_sync_pool(
    host: str = "localhost", port: int = 6379, db: int = 0, **kwargs
) -> redis.ConnectionPool:
    """Initialize a synchronous Redis connection pool."""
    try:
        pool = redis.ConnectionPool(host=host, port=port, db=db, **kwargs)
        logger.info(
            f"Synchronous Redis pool initialized with host={host}, port={port}, db={db}"
        )
        return pool
    except Exception as e:
        logger.error(f"Error initializing sync pool: {e}")
        raise


def initialize_async_pool(
    url: str = "redis://localhost", **kwargs
) -> AsyncConnectionPool:
    """Initialize an asynchronous Redis connection pool."""
    try:
        pool = AsyncConnectionPool.from_url(url, **kwargs)
        logger.info(f"Asynchronous Redis pool initialized with url={url}")
        return pool
    except Exception as e:
        logger.error(f"Error initializing async pool: {e}")
        raise


@contextmanager
def _get_sync_connection(
    connection: Union[redis.Redis, redis.ConnectionPool, None]
) -> Generator[redis.Redis, None, None]:
    conn = None
    try:
        if isinstance(connection, redis.Redis):
            yield connection
        elif isinstance(connection, redis.ConnectionPool):
            conn = redis.Redis(connection_pool=connection)
            yield conn
        else:
            raise ValueError("Invalid sync connection object.")
    except redis.ConnectionError as e:
        logger.error(f"Sync connection error: {e}")
        raise
    finally:
        if conn:
            conn.close()


@asynccontextmanager
async def _get_async_connection(
    connection: Union[AsyncRedis, AsyncConnectionPool, None]
) -> AsyncGenerator[AsyncRedis, None]:
    conn = None
    try:
        if isinstance(connection, AsyncRedis):
            yield connection
        elif isinstance(connection, AsyncConnectionPool):
            conn = AsyncRedis(connection_pool=connection)
            yield conn
        else:
            raise ValueError("Invalid async connection object.")
    except redis.ConnectionError as e:
        logger.error(f"Async connection error: {e}")
        raise
    finally:
        if conn:
            await conn.aclose()


class RedisSaver(BaseCheckpointSaver):
    sync_connection: Optional[Union[redis.Redis, redis.ConnectionPool]] = None
    async_connection: Optional[Union[AsyncRedis, AsyncConnectionPool]] = None

    def __init__(
        self,
        sync_connection: Optional[Union[redis.Redis, redis.ConnectionPool]] = None,
        async_connection: Optional[Union[AsyncRedis, AsyncConnectionPool]] = None,
    ):
        super().__init__(serde=JsonAndBinarySerializer())
        self.sync_connection = sync_connection
        self.async_connection = async_connection

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
    ) -> RunnableConfig:
        thread_id = config["configurable"]["thread_id"]
        parent_ts = config["configurable"].get("thread_ts")

        key = f"checkpoint${thread_id}${checkpoint['ts']}"
        try:
            with _get_sync_connection(self.sync_connection) as conn:
                conn.hset(
                    key,
                    mapping={
                        "checkpoint": self.serde.dumps(checkpoint),
                        "metadata": self.serde.dumps(metadata),
                        "parent_ts": parent_ts if parent_ts else "",
                    },
                )
                logger.info(
                    f"Checkpoint stored successfully for thread_id: {thread_id}, ts: {checkpoint['ts']}"
                )
        except Exception as e:
            logger.error(f"Failed to put checkpoint: {e}")
            raise
        return {
            "configurable": {
                "thread_id": thread_id,
                "thread_ts": checkpoint["ts"],
            },
        }

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
    ) -> RunnableConfig:
        thread_id = config["configurable"]["thread_id"]
        parent_ts = config["configurable"].get("thread_ts")
        key = f"checkpoint${thread_id}${checkpoint['ts']}"
        try:
            async with _get_async_connection(self.async_connection) as conn:
                await conn.hset(
                    key,
                    mapping={
                        "checkpoint": self.serde.dumps(checkpoint),
                        "metadata": self.serde.dumps(metadata),
                        "parent_ts": parent_ts if parent_ts else "",
                    },
                )
                logger.info(
                    f"Checkpoint stored successfully for thread_id: {thread_id}, ts: {checkpoint['ts']}"
                )
        except Exception as e:
            logger.error(f"Failed to aput checkpoint: {e}")
            raise
        return {
            "configurable": {
                "thread_id": thread_id,
                "thread_ts": checkpoint["ts"],
            },
        }

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        thread_id = config["configurable"]["thread_id"]
        thread_ts = config["configurable"].get("thread_ts", None)

        try:
            with _get_sync_connection(self.sync_connection) as conn:
                if thread_ts:
                    key = f"checkpoint${thread_id}${thread_ts}"
                else:
                    all_keys = conn.keys(f"checkpoint${thread_id}$*")
                    if not all_keys:
                        logger.info(f"No checkpoints found for thread_id: {thread_id}")
                        return None
                    # checkpoint:125:2024-07-26T11:49:46.662715+00:00
                    # get according to the latest timestamp
                    latest_key = max(all_keys, key=lambda k: k.decode().split("$")[-1])
                    key = latest_key.decode()
                checkpoint_data = conn.hgetall(key)
                if not checkpoint_data:
                    logger.info(f"No valid checkpoint data found for key: {key}")
                    return None
                checkpoint = self.serde.loads(checkpoint_data[b"checkpoint"].decode())
                metadata = self.serde.loads(checkpoint_data[b"metadata"].decode())
                parent_ts = checkpoint_data.get(b"parent_ts", b"").decode()
                parent_config = (
                    {"configurable": {"thread_id": thread_id, "thread_ts": parent_ts}}
                    if parent_ts
                    else None
                )
                logger.info(
                    f"Checkpoint retrieved successfully for thread_id: {thread_id}, ts: {thread_ts}"
                )
                return CheckpointTuple(
                    config=config,
                    checkpoint=checkpoint,
                    metadata=metadata,
                    parent_config=parent_config,
                )
        except Exception as e:
            logger.error(f"Failed to get checkpoint tuple: {e}")
            raise

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        thread_id = config["configurable"]["thread_id"]
        thread_ts = config["configurable"].get("thread_ts", None)
        try:
            async with _get_async_connection(self.async_connection) as conn:
                if thread_ts:
                    key = f"checkpoint${thread_id}${thread_ts}"
                else:
                    all_keys = await conn.keys(f"checkpoint${thread_id}$*")
                    if not all_keys:
                        logger.info(f"No checkpoints found for thread_id: {thread_id}")
                        return None
                    latest_key = max(all_keys, key=lambda k: k.decode().split("$")[-1])
                    key = latest_key.decode()
                checkpoint_data = await conn.hgetall(key)
                if not checkpoint_data:
                    logger.info(f"No valid checkpoint data found for key: {key}")
                    return None
                checkpoint = self.serde.loads(checkpoint_data[b"checkpoint"].decode())
                metadata = self.serde.loads(checkpoint_data[b"metadata"].decode())
                parent_ts = checkpoint_data.get(b"parent_ts", b"").decode()
                parent_config = (
                    {"configurable": {"thread_id": thread_id, "thread_ts": parent_ts}}
                    if parent_ts
                    else None
                )
                logger.info(
                    f"Checkpoint retrieved successfully for thread_id: {thread_id}, ts: {thread_ts}"
                )
                return CheckpointTuple(
                    config=config,
                    checkpoint=checkpoint,
                    metadata=metadata,
                    parent_config=parent_config,
                )
        except Exception as e:
            logger.error(f"Failed to get checkpoint tuple: {e}")
            raise

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Generator[CheckpointTuple, None, None]:
        thread_id = config["configurable"]["thread_id"] if config else "*"
        pattern = f"checkpoint:{thread_id}:*"
        try:
            with _get_sync_connection(self.sync_connection) as conn:
                keys = conn.keys(pattern)
                if before:
                    keys = [
                        k
                        for k in keys
                        if k.decode().split("$")[-1]
                        < before["configurable"]["thread_ts"]
                    ]
                keys = sorted(
                    keys, key=lambda k: k.decode().split("$")[-1], reverse=True
                )
                if limit:
                    keys = keys[:limit]
                for key in keys:
                    data = conn.hgetall(key)
                    if data and "checkpoint" in data and "metadata" in data:
                        thread_ts = key.decode().split("$")[-1]
                        yield CheckpointTuple(
                            config={
                                "configurable": {
                                    "thread_id": thread_id,
                                    "thread_ts": thread_ts,
                                }
                            },
                            checkpoint=self.serde.loads(data["checkpoint"].decode()),
                            metadata=self.serde.loads(data["metadata"].decode()),
                            parent_config=(
                                {
                                    "configurable": {
                                        "thread_id": thread_id,
                                        "thread_ts": data.get(
                                            "parent_ts", b""
                                        ).decode(),
                                    }
                                }
                                if data.get("parent_ts")
                                else None
                            ),
                        )
                        logger.info(
                            f"Checkpoint listed for thread_id: {thread_id}, ts: {thread_ts}"
                        )
        except Exception as e:
            logger.error(f"Failed to list checkpoints: {e}")
            raise

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncGenerator[CheckpointTuple, None]:
        thread_id = config["configurable"]["thread_id"] if config else "*"
        pattern = f"checkpoint:{thread_id}:*"
        try:
            async with _get_async_connection(self.async_connection) as conn:
                keys = await conn.keys(pattern)
                if before:
                    keys = [
                        k
                        for k in keys
                        if k.decode().split("$")[-1]
                        < before["configurable"]["thread_ts"]
                    ]
                keys = sorted(
                    keys, key=lambda k: k.decode().split("$")[-1], reverse=True
                )
                if limit:
                    keys = keys[:limit]
                for key in keys:
                    data = await conn.hgetall(key)
                    if data and "checkpoint" in data and "metadata" in data:
                        thread_ts = key.decode().split("$")[-1]
                        yield CheckpointTuple(
                            config={
                                "configurable": {
                                    "thread_id": thread_id,
                                    "thread_ts": thread_ts,
                                }
                            },
                            checkpoint=self.serde.loads(data["checkpoint"].decode()),
                            metadata=self.serde.loads(data["metadata"].decode()),
                            parent_config=(
                                {
                                    "configurable": {
                                        "thread_id": thread_id,
                                        "thread_ts": data.get(
                                            "parent_ts", b""
                                        ).decode(),
                                    }
                                }
                                if data.get("parent_ts")
                                else None
                            ),
                        )
                        logger.info(
                            f"Checkpoint listed for thread_id: {thread_id}, ts: {thread_ts}"
                        )
        except Exception as e:
            logger.error(f"Failed to list checkpoints: {e}")
            raise

[tmishinev]

I think it should be applied also to :

pattern = f"checkpoint:{thread_id}:*"

in list/alist

[ishanExtreme]

If you are copying above code please edit list/alist also...
Thanks for the correction @tmishinev. There's no edit option.

[lawtech0902]

missing put_writes and aput_writes functions

[vbarda]

Updated -- please see the latest version of the guide!

[vbarda]

Hi all! The guide has been updated with the correct implementtions of .put_writes and .aput_writes. The new guide also has two separate checkpointer implementations for convenience -- RedisSaver (for using graphs synchronously) and AsyncRedisSaver (for using graphs asynchronously)

[jamesheavey]

Is it possible to use custom serializable objects in the state or do they have to be specific Langchain objects?

i.e. I made an Action class like this

class Action(Serializable):
    thought: str
    action: str
    action_input: Union[str, dict]
    log: str = None

    type: Literal["Action"] = "Action"

    def __init__(self, thought: str, action: str, action_input: Union[str, dict], log: str = None, **kwargs: Any):
        super().__init__(thought=thought, action=action, action_input=action_input, log=log, **kwargs)

    @classmethod
    def is_lc_serializable(cls) -> bool:
        return True

    @classmethod
    def get_lc_namespace(cls) -> List[str]:
        return ["langchain", "schema"]

Obviously the namespace doesnt exist, but is there a way to make it work in my graph state memory?

class State(TypedDict):
    input: str
    messages: Annotated[list[AnyMessage], add_messages]
    steps: Annotated[list[Union[Action, Observation, Finish]], operator.add]

[vbarda]

you would likely need to subclass jsonplus serializer implementation here and add support for your specific object. i definitely would not recommend trying to coerce your custom classes to langchain serializable objects

[jamesheavey]

Ok thank you, so only langchain objects can be used in the state with memory

[vbarda]

that's not exactly correct, you can use any objects as long as you add custom serialization logic to .loads_typed and .dumps_typed methods on the serializer to handle your custom objects. also, the serializer I linked above handles a lot more than just langchain objects

[jamesheavey]

So those classes I made are subclasses of the serialisable class and are basically copies of the existing LC classes AgentAction, AgentStep etc. i basically wanted a similar class but with my own variable names.

https://api.python.langchain.com/en/latest/agents/langchain_core.agents.AgentAction.html

The issue is that the classes I made dont have a valid LC namespace (I.e. the is_lc_serializable and get_lc_namespace functions?)

Are you saying that It isnt necessary to have those lc functions and I should make the classes a child of JsonPlusSerializer instead of Serializable?

Thank you again for your help!

[jamesheavey]

for anyone curious, I have found the solution to be to make the classes a child of JsonPlusSerialiser and making them dataclasses

from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
from typing import Union
from dataclasses import dataclass


@dataclass
class Action(JsonPlusSerializer):
    thought: str
    action: str
    action_input: Union[str, dict]
    log: str = None


@dataclass
class Observation(JsonPlusSerializer):
    observation: str
    log: str = None


@dataclass
class Finish(JsonPlusSerializer):
    output: str
    log: str = None

[whyxup]

When using langserve and langgraph, I have the following code:
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
---------------------------error--------------------------------
TypeError: MemorySaver.aput() missing 1 required positional argument: 'new_versions'

[vbarda]

you need to update langgraph

[pelguetat]

I'm having issues with the size of the checkpoint exceeding the allowed size in Firestore. Is there any way of limiting or splitting the updates?

[chris-speebly]

Are there any issues creating a custom persistent solution outside the checkpoint protocol, where I manually populate the state from some data source at the start node and save it at the end node or some other node?

[Bennoo]

This is not clear for me that the memory is keeping the whole graph or if I need to rebuild it from scratch for a stateless application.

If I keep only the thread do I have to recompile the whole graph or can I retrieve the whole workflow from memory?

[SujayC66]

Can anyone please add memory checkpointer to agentic rag tutorial in langgraph?

[Bear1110]

The code works well, but I have a question about the following part:

if not get_checkpoint_id(config):
            config = {
                "configurable": {
                    "thread_id": thread_id,
                    "checkpoint_ns": checkpoint_ns,
                    "checkpoint_id": checkpoint_id,
                }
            }

From the context of the function, it seems like this newly assigned config isn't used afterward. Is this code redundant, and can it be safely removed? Or am I missing something?

So it is a meaningless code, Could we remove it?

[vbarda]

yup, redundant, will remove

[Gunoo1]

I literally copied the tutorial code (https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/time-travel/#build-the-agent) recently on jupyter notebook just to see if it was streamlit working in tangent with MemorySaver() which was not allowing the agent to store memory, but somehow when I run it it loses all Memory even with the exact same code, and the code has no previous context. Does memorysaver not work or am I doing something wrong

[tmishinev]

is there anyway we can define that one as a global checkpointer ?

async with AsyncRedisSaver.from_conn_info(
host="localhost", port=6379, db=0
) as checkpointer:
graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)
config = {"configurable": {"thread_id": "2"}}
res = await graph.ainvoke(
{"messages": [("human", "what's the weather in nyc")]}, config
)

latest_checkpoint = await checkpointer.aget(config)
latest_checkpoint_tuple = await checkpointer.aget_tuple(config)
checkpoint_tuples = [c async for c in checkpointer.alist(config)]

[pedrorjbr]

I am getting:
UnicodeDecodeError: 'utf-8' codec can't decode byte 0x87 in position 0: invalid start byte
on
serde.loads_typed(
(data[b"type"].decode(), data[b"checkpoint"].decode())
)
Change for this makes it work.
serde.loads_typed(
(data[b"type"].decode(), data[b"checkpoint"])
)

[pedrorjbr]

def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
    if isinstance(obj, bytes):
        return "bytes", obj
    elif isinstance(obj, bytearray):
        return "bytearray", obj
    else:
        try:
            return "msgpack", _msgpack_enc(obj)
        except UnicodeEncodeError:
            return "json", self.dumps(obj)

def loads_typed(self, data: tuple[str, bytes]) -> Any:
    type_, data_ = data
    if type_ == "bytes":
        return data_
    elif type_ == "bytearray":
        return bytearray(data_)
    elif type_ == "json":
        return self.loads(data_)
    elif type_ == "msgpack":
        return msgpack.unpackb(data_, ext_hook=_msgpack_ext_hook)
    else:
        raise NotImplementedError(f"Unknown serialization type: {type_}")

On serializing a dict, istead of using msgpack, why not use json? The loads_typed is using it.

[zzaym]

Got the same error, but it worked fine for me before creating a new virtualenv and reinstalling dependencies. Not sure because of langgraph or redis-py upgrade.

[zzaym]

It seems due to 3b05279, which came out with langgraph 0.2.23

[zzaym]

@vbarda Could you please update the above implementation to accommodate this serialization change?

[nfcampos]

Hi, this is a bug in the redis guide, will be updated, not related to msgpack.

[andyops]

Can someone pls help on how to add delete checkpoint functionality. In the above code

[zzaym]

This is how I added such a method:

    async def adelete_checkpoint(self, thread_id: str) -> None:
        key_patterns = [
            REDIS_KEY_SEPARATOR.join(["checkpoint", thread_id, "*"]),
            REDIS_KEY_SEPARATOR.join(["writes", thread_id, "*"]),
        ]
        for pattern in key_patterns:
            all_keys = await self.conn.keys(pattern)
            for key in all_keys:
                await self.conn.delete(key)

[dyh2024]

How to update the checkpoint with redis? eg. the histrory messsages [mes1, mes2, mes3, mes4,mes5, mes6, mes7, mes8, mes9, mes10, mes11, mes12] , my state_modifier is keep the last 10 messages [mes3, mes4,mes5, mes6, mes7, mes8, mes9, mes10, mes11, mes12], after agent.invoke , agent.get_state show the whole history messages[mes1, mes2, mes3, mes4,mes5, mes6, mes7, mes8, mes9, mes10, mes11, mes12], but I want to [mes3, mes4,mes5, mes6, mes7, mes8, mes9, mes10, mes11, mes12]. I think it is because the checkpoint has not updated .

[wangxinzhang]

When the graph is executed, the MemorySaver keeps growing. What are some solutions to implement a memory eviction mechanism?

[adambaker]

These checkpointers don't work with subgraph state. I took the RedisSaver here and put it at the top of the subgraph state management tutorial and replaced the MemorySaver there with this RedisSaver, and I get the following error:

     59 def _parse_redis_checkpoint_key(redis_key: str) -> dict:
---> 60     namespace, thread_id, checkpoint_ns, checkpoint_id = redis_key.split(
     61         REDIS_KEY_SEPARATOR
     62     )
     63     if namespace != "checkpoint":
     64         raise ValueError("Expected checkpoint key to start with 'checkpoint'")

ValueError: too many values to unpack (expected 4)

from the Resuming from breakpoints section, the third cell:

state = graph.get_state(config, subgraphs=True)
state.tasks[0]

It got 5 values. Looks like handing subgraphs correctly requires handling a sequence of checkpoint ids instead of a single one?