Persistence with PostgresSaver and Redis

Capsule overview

MemorySaver works perfectly on your machine. You open the notebook, talk to the agent, the checkpoints get saved, time-travel works. You close the notebook. You open it the next day. Everything is gone. Every checkpoint, every conversation, every state history — evaporated.

That's fine in development. In production it's unacceptable. Your user talked to the agent yesterday, closed the tab, and today expects to continue. Your server restarted at 3am to apply a security patch, and the 200 active threads were lost. Your app runs on 4 Kubernetes instances, and the user who was on instance 2 now lands on instance 3 — with no context.

PostgresSaver solves all of it. Checkpoints get stored in a PostgreSQL database. They survive crashes, restarts, deploys, and instance migrations. And migrating from MemorySaver is trivial: you change one line of code. One. Everything else — thread_id, get_state(), get_state_history(), time-travel — works exactly the same.

The migration: one line of code

This is the complete transition from development to production:

Before (development with MemorySaver)

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

After (production with PostgresSaver)

from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@localhost:5432/mydb")
checkpointer.setup()
graph = graph_builder.compile(checkpointer=checkpointer)

That's it. The rest of your code — the nodes, the edges, the invokes with thread_id, the calls to get_state(), the time-travel — doesn't change a single line. That's the power of LangGraph's checkpointer abstraction: the graph neither knows nor cares where the checkpoints are stored. It only knows that they are.

PostgresSaver: full setup

Installation

pip install langgraph-checkpoint-postgres psycopg[binary]

langgraph-checkpoint-postgres is the package that ships PostgresSaver. psycopg[binary] is the PostgreSQL driver for Python — the binary variant bundles precompiled C libraries so you don't need a compiler.

Connection string

The format is standard PostgreSQL:

postgresql://user:password@host:port/database

In production, never put the password in the code. Use environment variables:

import os

conn_string = os.environ["DATABASE_URL"]
# Example: DATABASE_URL=postgresql://user:pass@db.example.com:5432/langgraph_prod

Creating the required tables

PostgresSaver needs tables to store checkpoints. The setup() method creates them automatically:

from langgraph.checkpoint.postgres import PostgresSaver

conn_string = "postgresql://postgres:postgres@localhost:5432/langgraph_dev"
checkpointer = PostgresSaver.from_conn_string(conn_string)
checkpointer.setup()

setup() is idempotent — you can run it as many times as you want. If the tables already exist, it does nothing. It's safe to call at application startup.

The async version

For async applications (FastAPI, for instance), use the AsyncPostgresSaver variant:

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

conn_string = "postgresql://postgres:postgres@localhost:5432/langgraph_dev"

async def create_graph():
    checkpointer = AsyncPostgresSaver.from_conn_string(conn_string)
    await checkpointer.setup()
    graph = graph_builder.compile(checkpointer=checkpointer)
    return graph

Complete example: chatbot with PostgresSaver

import os
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

conn_string = os.environ.get(
    "DATABASE_URL",
    "postgresql://postgres:postgres@localhost:5432/langgraph_dev"
)
checkpointer = PostgresSaver.from_conn_string(conn_string)
checkpointer.setup()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user_001"}}

result = graph.invoke({"messages": [("user", "What is RAG?")]}, config)
print(result["messages"][-1].content)
# Output: "RAG (Retrieval-Augmented Generation) is a technique that..."

result = graph.invoke({"messages": [("user", "Give me examples")]}, config)
print(result["messages"][-1].content)
# Output: "Here are some RAG examples: 1) A chatbot that queries
#          internal documentation..."

The code is nearly identical to the MemorySaver version. The only differences:

1. from langgraph.checkpoint.postgres import PostgresSaver instead of from langgraph.checkpoint.memory import MemorySaver
2. PostgresSaver.from_conn_string(conn_string) instead of MemorySaver()
3. checkpointer.setup() to create the tables

Everything else — thread_id, invoke, get_state(), get_state_history() — is identical. If tomorrow you need to move from Postgres to another backend, you change those 2-3 lines.

Durability: the real difference

The difference between MemorySaver and PostgresSaver isn't the API — it's what happens when something goes wrong.

When to use which checkpointer

Decision tree

Are you in development/testing?
  └── Yes → MemorySaver (no setup, no dependencies)
  └── No → Do you need full durability?
              └── Yes → PostgresSaver (the production standard)
              └── No → Do you need maximum speed?
                        └── Yes → RedisSaver (ephemeral sessions, high frequency)
                        └── No → PostgresSaver (the production default)

The recommendation for 90% of cases: MemorySaver for development, PostgresSaver for production. RedisSaver is for specific scenarios where read/write latency matters more than absolute durability.

RedisSaver: when you need speed

RedisSaver stores checkpoints in Redis instead of PostgreSQL. Redis runs in memory (like MemorySaver) but it's a separate service that persists data to disk and survives restarts of the Python process.

Installation

pip install langgraph-checkpoint-redis

Basic usage

from langgraph.checkpoint.redis import RedisSaver

checkpointer = RedisSaver.from_conn_string("redis://localhost:6379")
graph = graph_builder.compile(checkpointer=checkpointer)

PostgresSaver vs RedisSaver

RedisSaver shines when:

• You have thousands of requests per second and every millisecond counts
• Checkpoints are ephemeral (chat sessions that expire in 24h)
• You already run Redis in your infrastructure

PostgresSaver wins when:

• You need guaranteed durability
• You want to query the checkpoints (analytics, debugging)
• Checkpoints must persist indefinitely
• The data volume grows past what fits in RAM

Connection pooling: handling connections efficiently

In production, every request that needs the checkpointer opens and closes a database connection. With 100 concurrent requests, that's 100 open connections at once. PostgreSQL has a limit (default: 100), and opening/closing them constantly is expensive.

Connection pooling solves this: it keeps a pool of reusable connections.

With the psycopg pool

import os
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver

conn_string = os.environ.get(
    "DATABASE_URL",
    "postgresql://postgres:postgres@localhost:5432/langgraph_dev"
)

pool = ConnectionPool(
    conninfo=conn_string,
    min_size=5,
    max_size=20,
)

checkpointer = PostgresSaver(conn=pool)
checkpointer.setup()
graph = graph_builder.compile(checkpointer=checkpointer)

Rule of thumb: max_size ≤ PostgreSQL's max_connections / number of app instances. If PostgreSQL has max_connections=100 and you run 4 instances, max_size=20 per instance (4 × 20 = 80, leaving headroom).

For async (FastAPI), use AsyncConnectionPool from psycopg_pool with AsyncPostgresSaver — same parameters, same logic, but with await pool.open().

Checkpoint retention and cleanup

Checkpoints pile up. Without a retention policy, the database grows forever. Three strategies: by time (delete anything older than N days), by count (keep the last N per thread), or by activity (inactive threads).

In PostgreSQL, direct cleanup with SQL:

DELETE FROM checkpoints WHERE created_at < NOW() - INTERVAL '30 days';

In Redis, native TTL — the checkpoints expire on their own:

checkpointer = RedisSaver.from_conn_string("redis://localhost:6379", ttl={"default": 86400})

Storage rules: don't put large payloads in the state (use references, not content), apply message trimming (capsule 02), and monitor size with periodic SQL queries.

Production pattern: picking the checkpointer by environment

import os

def get_checkpointer():
    env = os.environ.get("ENVIRONMENT", "development")
    if env == "development":
        from langgraph.checkpoint.memory import MemorySaver
        return MemorySaver()
    elif env in ("production", "staging"):
        from langgraph.checkpoint.postgres import PostgresSaver
        checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
        checkpointer.setup()
        return checkpointer
    raise ValueError(f"Unknown environment: {env}")

graph = graph_builder.compile(checkpointer=get_checkpointer())

The graph doesn't change. The nodes don't change. The tests don't change. Only the checkpointer varies by environment.

Complete example: FastAPI with async PostgresSaver

A production pattern with FastAPI: the lifespan creates the pool and the checkpointer at startup, and closes them at shutdown. The pool is shared across all requests. Each request passes its thread_id and uses ainvoke (async). The server can restart and the conversations persist.

import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from pydantic import BaseModel
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.graph import StateGraph, MessagesState, START, END
from psycopg_pool import AsyncConnectionPool

graph = None


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)


@asynccontextmanager
async def lifespan(app: FastAPI):
    global graph
    conn_string = os.environ["DATABASE_URL"]
    pool = AsyncConnectionPool(conninfo=conn_string, min_size=5, max_size=20)
    await pool.open()
    checkpointer = AsyncPostgresSaver(conn=pool)
    await checkpointer.setup()
    graph = graph_builder.compile(checkpointer=checkpointer)
    yield
    await pool.close()


app = FastAPI(lifespan=lifespan)


class ChatRequest(BaseModel):
    message: str
    thread_id: str


@app.post("/chat")
async def chat(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}
    result = await graph.ainvoke(
        {"messages": [("user", request.message)]}, config
    )
    return {"response": result["messages"][-1].content, "thread_id": request.thread_id}

# POST /chat {"message": "What is RAG?", "thread_id": "user_001"}
# → {"response": "RAG is...", "thread_id": "user_001"}

Troubleshooting

Problem 1: "relation 'checkpoints' does not exist"

Symptom: An error on the first invoke after configuring PostgresSaver. Cause: You didn't call checkpointer.setup() to create the tables. Fix: Add checkpointer.setup() (or await checkpointer.setup() in async) after creating the checkpointer. It's idempotent — safe to call on every startup.

Problem 2: "connection refused" when connecting to PostgreSQL

Symptom: psycopg.OperationalError: connection to server at "localhost"... refused. Cause: PostgreSQL isn't running, or the port/host is wrong. Fix: Check that PostgreSQL is up (pg_isready -h localhost -p 5432). If you use Docker: docker run -d -p 5432:5432 -e POSTGRES_PASSWORD=postgres postgres:16.

Problem 3: "too many clients already" in production

Symptom: psycopg.OperationalError: too many clients already under load. Cause: Every request opens a new connection and no pool is configured, or the pool's max_size exceeds PostgreSQL's max_connections. Fix: Use ConnectionPool with an appropriate max_size (see the connection pooling section). Rule: max_size × num_instances < max_connections.

Problem 4: Degraded performance with many checkpoints

Symptom: get_state_history() is slow after thousands of invocations on a thread. Cause: The checkpoints table grew without extra indexing or a retention policy. Fix: Implement a retention policy (delete old checkpoints) and confirm the table's indexes are in place. See the checkpoint management section.

Problem 5: MemorySaver in production "works" but loses data

Symptom: Everything is fine until the server restarts and the conversations vanish. Cause: MemorySaver is running in production. Fix: Migrate to PostgresSaver. It's literally a one-line change (see "The migration").

Exercises

Exercise 1: MemorySaver → PostgresSaver migration (Easy)

Here's some code using MemorySaver. Change it to use PostgresSaver with the connection string "postgresql://postgres:postgres@localhost:5432/langgraph_dev". Don't change anything except the checkpointer lines.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END

def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}

graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

conn_string = "postgresql://postgres:postgres@localhost:5432/langgraph_dev"
checkpointer = PostgresSaver.from_conn_string(conn_string)
checkpointer.setup()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "migration_test"}}
result = graph.invoke({"messages": [("user", "Did the migration work?")]}, config)
print(result["messages"][-1].content)
# Expected output: Yes, the migration works. The agent answers normally.

state = graph.get_state(config)
print(f"Checkpoint stored in Postgres: {len(state.values['messages'])} messages")
print("✅ Migration successful — checkpoints now persist in PostgreSQL")
# Expected output:
# Checkpoint stored in Postgres: 2 messages
# ✅ Migration successful — checkpoints now persist in PostgreSQL

Exercise 2: Picking a checkpointer by environment (Easy)

Write a function get_checkpointer(env: str) that returns MemorySaver for "development", PostgresSaver for "production" (with the connection string from an environment variable), and raises ValueError for anything else. Test it with "development" and check that it returns a MemorySaver.

import os
from langgraph.checkpoint.memory import MemorySaver


def get_checkpointer(env: str):
    if env == "development":
        return MemorySaver()

    elif env == "production":
        from langgraph.checkpoint.postgres import PostgresSaver
        conn_string = os.environ.get("DATABASE_URL")
        if not conn_string:
            raise ValueError("DATABASE_URL environment variable required for production")
        checkpointer = PostgresSaver.from_conn_string(conn_string)
        checkpointer.setup()
        return checkpointer

    raise ValueError(f"Unknown environment: {env}")


dev_checkpointer = get_checkpointer("development")
print(f"Dev checkpointer: {type(dev_checkpointer).__name__}")
assert type(dev_checkpointer).__name__ == "MemorySaver"

try:
    get_checkpointer("unknown")
except ValueError as e:
    print(f"Expected error: {e}")

print("✅ The checkpointer selector works correctly")
# Expected output:
# Dev checkpointer: MemorySaver
# Expected error: Unknown environment: unknown
# ✅ The checkpointer selector works correctly

Exercise 3: Verify persistence across a simulated restart (Medium)

Build a graph with MemorySaver. Make 3 invocations on one thread. Then simulate a "restart" by creating a fresh MemorySaver() and compiling a new graph. Verify that the new graph does NOT have the previous checkpoints. Then repeat the experiment reusing the same checkpointer object — verify that it DOES keep the checkpoints. This is exactly why MemorySaver isn't fit for production.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def build_graph(checkpointer):
    def chatbot(state: MessagesState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        response = model.invoke(state["messages"])
        return {"messages": [response]}

    builder = StateGraph(MessagesState)
    builder.add_node("chatbot", chatbot)
    builder.add_edge(START, "chatbot")
    builder.add_edge("chatbot", END)
    return builder.compile(checkpointer=checkpointer)


config = {"configurable": {"thread_id": "persist_test"}}

print("=== Scenario 1: New MemorySaver (simulates a restart) ===")
checkpointer_v1 = MemorySaver()
graph_v1 = build_graph(checkpointer_v1)

graph_v1.invoke({"messages": [("user", "Hi, this is turn 1")]}, config)
graph_v1.invoke({"messages": [("user", "This is turn 2")]}, config)
graph_v1.invoke({"messages": [("user", "And this is turn 3")]}, config)

state_v1 = graph_v1.get_state(config)
print(f"Before the restart: {len(state_v1.values['messages'])} messages")

checkpointer_v2 = MemorySaver()
graph_v2 = build_graph(checkpointer_v2)

state_v2 = graph_v2.get_state(config)
has_state = state_v2.values is not None and len(state_v2.values.get("messages", [])) > 0
print(f"After the restart: {'has data' if has_state else 'empty'}")
assert not has_state, "It shouldn't have data after a simulated restart"
print("❌ Checkpoints lost — MemorySaver doesn't persist across restarts\n")

print("=== Scenario 2: Same MemorySaver (no real restart) ===")
shared_checkpointer = MemorySaver()
graph_a = build_graph(shared_checkpointer)
graph_a.invoke({"messages": [("user", "Message in graph A")]}, config)

graph_b = build_graph(shared_checkpointer)
state_b = graph_b.get_state(config)
has_state_b = state_b.values is not None and len(state_b.values.get("messages", [])) > 0
print(f"Graph B with the same checkpointer: {'has data' if has_state_b else 'empty'}")
assert has_state_b
print("✅ Checkpoints preserved — same object in memory")
print("Takeaway: MemorySaver loses data when you create a new instance. PostgresSaver doesn't.")
# Expected output:
# Before the restart: 6 messages → After: empty
# ❌ MemorySaver doesn't persist → ✅ The same object does persist

Exercise 4: Connection pool with validation (Medium)

Write create_pooled_checkpointer(conn_string, min_size, max_size) with validation: min_size >= 1, max_size >= min_size, max_size <= 50. Test that the validations reject dangerous configurations.

from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver


def create_pooled_checkpointer(conn_string: str, min_size: int = 5, max_size: int = 20):
    if min_size < 1:
        raise ValueError(f"min_size must be >= 1, got: {min_size}")
    if max_size < min_size:
        raise ValueError(f"max_size ({max_size}) must be >= min_size ({min_size})")
    if max_size > 50:
        raise ValueError(f"max_size must be <= 50, got: {max_size}")

    pool = ConnectionPool(conninfo=conn_string, min_size=min_size, max_size=max_size)
    checkpointer = PostgresSaver(conn=pool)
    checkpointer.setup()
    return checkpointer


invalid_cases = [(0, 10), (10, 5), (5, 100)]
for min_s, max_s in invalid_cases:
    try:
        create_pooled_checkpointer("postgresql://localhost/test", min_s, max_s)
        print(f"  ❌ min={min_s}, max={max_s}: should have failed")
    except ValueError as e:
        print(f"  ✅ min={min_s}, max={max_s}: rejected ({e})")

print("✅ The pool validations work correctly")
# Expected output:
#   ✅ min=0, max=10: rejected (min_size must be >= 1...)
#   ✅ min=10, max=5: rejected (max_size (5) must be >= min_size...)
#   ✅ min=5, max=100: rejected (max_size must be <= 50...)
# ✅ The pool validations work correctly

Exercise 5: Checkpoint cleanup function (Medium-Advanced)

Write a function cleanup_old_threads(graph, known_thread_ids, max_age_turns, active_thread_ids) that returns the thread_ids that should be cleaned up (inactive ones with more than max_age_turns turns). Simulate it with MemorySaver.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    return {"messages": [model.invoke(state["messages"])]}

graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)


def populate_thread(graph, thread_id: str, num_turns: int):
    config = {"configurable": {"thread_id": thread_id}}
    for i in range(num_turns):
        graph.invoke({"messages": [("user", f"Turn {i+1}")]}, config)


def cleanup_old_threads(graph, known_ids, max_age_turns, active_ids):
    to_clean = []
    for tid in known_ids:
        if tid in active_ids:
            continue
        state = graph.get_state({"configurable": {"thread_id": tid}})
        if state.values and len(state.values.get("messages", [])) // 2 > max_age_turns:
            to_clean.append(tid)
    return to_clean


populate_thread(graph, "alice", 3)
populate_thread(graph, "bob", 15)
populate_thread(graph, "carol", 8)
populate_thread(graph, "dave", 2)
populate_thread(graph, "eve", 20)

to_clean = cleanup_old_threads(
    graph, ["alice", "bob", "carol", "dave", "eve"],
    max_age_turns=10, active_ids={"alice", "carol"}
)
print(f"Threads to clean up: {to_clean}")
assert "bob" in to_clean and "eve" in to_clean
assert "alice" not in to_clean and "dave" not in to_clean
print("✅ Cleanup logic verified")
# Expected output:
# Threads to clean up: ['bob', 'eve']
# ✅ Cleanup logic verified

Exercise 6: FastAPI endpoint with persistence (Advanced)

Write a POST /chat endpoint in FastAPI that takes message and thread_id, uses a graph with MemorySaver, and returns the response. The checkpointer must be initialized in the lifespan. Include comments showing what to change to migrate to AsyncPostgresSaver.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from pydantic import BaseModel
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END

graph = None


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)


@asynccontextmanager
async def lifespan(app: FastAPI):
    global graph
    # To migrate to PostgresSaver, change these lines:
    # from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
    # checkpointer = AsyncPostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
    # await checkpointer.setup()
    checkpointer = MemorySaver()
    graph = graph_builder.compile(checkpointer=checkpointer)
    yield


app = FastAPI(lifespan=lifespan)


class ChatRequest(BaseModel):
    message: str
    thread_id: str


@app.post("/chat")
async def chat(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}
    result = await graph.ainvoke(
        {"messages": [("user", request.message)]}, config
    )
    return {
        "response": result["messages"][-1].content,
        "thread_id": request.thread_id,
        "total_messages": len(result["messages"])
    }


# uvicorn module_name:app --reload
# curl -X POST http://localhost:8000/chat -H "Content-Type: application/json" \
#      -d '{"message": "What is RAG?", "thread_id": "test_001"}'
# → {"response": "RAG is...", "thread_id": "test_001", "total_messages": 2}

Summary

In this capsule you learned:

• Migrating from MemorySaver to PostgresSaver is one line of code. Swap MemorySaver() for PostgresSaver.from_conn_string(conn_string) — everything else (thread_id, get_state, get_state_history, time-travel) behaves identically. That's the power of the checkpointer abstraction
• PostgresSaver = real durability. Checkpoints survive crashes, restarts, deploys, and instance migrations. In a multi-instance setup (Kubernetes), every pod shares the same database
• RedisSaver is for high frequency. Sub-millisecond latency, native TTL to expire checkpoints automatically. Ideal for ephemeral sessions. But durability isn't absolute like it is with Postgres
• Connection pooling is mandatory in production. Without a pool, every request opens a fresh connection. With 100 concurrent requests, you exhaust PostgreSQL's connections. ConnectionPool(min_size=5, max_size=20) solves it
• Checkpoints accumulate and need retention. A cleanup policy by time, by count, or by activity. In Redis, native TTL. In PostgreSQL, periodic SQL queries
• The production pattern: MemorySaver for development (no dependencies), PostgresSaver + connection pooling for production (durability + scale), selected via an environment variable

Further reading

1. LangGraph — PostgresSaver — API reference for PostgresSaver with every method and configuration option
2. LangGraph — Persistence How-to — Official guide to implementing persistence with different backends
3. psycopg3 — Connection Pools — Docs for psycopg3's connection pool, the driver PostgresSaver uses
4. Redis Persistence — redis.io — How Redis persists data to disk: RDB snapshots vs the AOF log. Important for understanding RedisSaver's durability
5. PostgreSQL Connection Management — Official PostgreSQL docs on max_connections and connection management

Module 8 — LangChain & LangGraph: From Chains to Agents