How to add thread-level persistence with Redis (functional API)#
!!! info “Prerequisites”
This guide assumes familiarity with the following:
- [Functional API](../../concepts/functional_api/)
- [Persistence](../../concepts/persistence/)
- [Memory](../../concepts/memory/)
- [Chat Models](https://python.langchain.com/docs/concepts/chat_models/)
!!! info “Not needed for LangGraph API users”
If you're using the LangGraph API, you needn't manually implement a checkpointer. The API automatically handles checkpointing for you. This guide is relevant when implementing LangGraph in your own custom server.
Many AI applications need memory to share context across multiple interactions on the same thread (e.g., multiple turns of a conversation). In LangGraph functional API, this kind of memory can be added to any [entrypoint()][langgraph.func.entrypoint] workflow using thread-level persistence.
When creating a LangGraph workflow, you can set it up to persist its results by using a checkpointer:
Create an instance of a Redis checkpointer:
from langgraph.checkpoint.redis import RedisSaver # Set up Redis connection for checkpointer REDIS_URI = "redis://redis:6379" checkpointer = None with RedisSaver.from_conn_string(REDIS_URI) as cp: cp.setup() checkpointer = cp
Pass
checkpointerinstance to theentrypoint()decorator:from langgraph.func import entrypoint @entrypoint(checkpointer=checkpointer) def workflow(inputs) ...
Optionally expose
previousparameter in the workflow function signature:@entrypoint(checkpointer=checkpointer) def workflow( inputs, *, # you can optionally specify `previous` in the workflow function signature # to access the return value from the workflow as of the last execution previous ): previous = previous or [] combined_inputs = previous + inputs result = do_something(combined_inputs) ...
Optionally choose which values will be returned from the workflow and which will be saved by the checkpointer as
previous:@entrypoint(checkpointer=checkpointer) def workflow(inputs, *, previous): ... result = do_something(...) return entrypoint.final(value=result, save=combine(inputs, result))
This guide shows how you can add thread-level persistence to your workflow using Redis as the backing store.
!!! tip “Note”
If you need memory that is __shared__ across multiple conversations or users (cross-thread persistence), check out this [how-to guide](../cross-thread-persistence-functional).
!!! tip “Note”
If you need to add thread-level persistence to a `StateGraph`, check out this [how-to guide](../persistence).
Setup#
First we need to install the packages required
%%capture --no-stderr
%pip install --quiet -U langgraph langchain_anthropic
Next, we need to set API key for Anthropic (the LLM we will use).
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("ANTHROPIC_API_KEY")
Example: simple chatbot with short-term memory#
We will be using a workflow with a single task that calls a chat model.
Let’s first define the model we’ll be using:
from langchain_anthropic import ChatAnthropic
model = ChatAnthropic(model="claude-sonnet-4-20250514")
Now we can define our task and workflow. To add in persistence, we need to pass in a Checkpointer to the [entrypoint()][langgraph.func.entrypoint] decorator.
from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task
from langgraph.checkpoint.redis import RedisSaver
@task
def call_model(messages: list[BaseMessage]):
response = model.invoke(messages)
return response
# Set up Redis connection for checkpointer
REDIS_URI = "redis://redis:6379"
checkpointer = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
cp.setup()
checkpointer = cp
@entrypoint(checkpointer=checkpointer)
def workflow(inputs: list[BaseMessage], *, previous: list[BaseMessage]):
if previous:
inputs = add_messages(previous, inputs)
response = call_model(inputs).result()
return entrypoint.final(value=response, save=add_messages(inputs, response))
00:36:43 langgraph.checkpoint.redis INFO Redis client is a standalone client
00:36:43 redisvl.index.index INFO Index already exists, not overwriting.
00:36:43 redisvl.index.index INFO Index already exists, not overwriting.
00:36:43 redisvl.index.index INFO Index already exists, not overwriting.
If we try to use this workflow, the context of the conversation will be persisted across interactions:
!!! note Note
If you're using LangGraph Platform or LangGraph Studio, you __don't need__ to pass checkpointer to the entrypoint decorator, since it's done automatically.
We can now interact with the agent and see that it remembers previous messages!
config = {"configurable": {"thread_id": "1"}}
input_message = {"role": "user", "content": "hi! I'm bob"}
for chunk in workflow.stream([input_message], config, stream_mode="values"):
chunk.pretty_print()
00:36:45 httpx INFO HTTP Request: POST https://api.anthropic.com/v1/messages "HTTP/1.1 200 OK"
================================== Ai Message ==================================
Hi Bob! Good to hear from you again. How are you doing today?
You can always resume previous threads:
input_message = {"role": "user", "content": "what's my name?"}
for chunk in workflow.stream([input_message], config, stream_mode="values"):
chunk.pretty_print()
00:36:48 httpx INFO HTTP Request: POST https://api.anthropic.com/v1/messages "HTTP/1.1 200 OK"
================================== Ai Message ==================================
Your name is Bob! You introduced yourself to me at the beginning of our conversation.
If we want to start a new conversation, we can pass in a different thread_id. Poof! All the memories are gone!
input_message = {"role": "user", "content": "what's my name?"}
for chunk in workflow.stream(
[input_message],
{"configurable": {"thread_id": "2"}},
stream_mode="values",
):
chunk.pretty_print()
00:36:50 httpx INFO HTTP Request: POST https://api.anthropic.com/v1/messages "HTTP/1.1 200 OK"
================================== Ai Message ==================================
I don't actually know your name. I don't have access to information about who you are unless you tell me directly in our conversation. If you'd like me to know your name, please feel free to share it with me!
!!! tip “Streaming tokens”
If you would like to stream LLM tokens from your chatbot, you can use `stream_mode="messages"`. Check out this [how-to guide](../streaming-tokens) to learn more.