Langgraph Agent with Postgres Persistence

This repo is a example of how to use Langgraph agent with Postgres DB for checkpointing. Agent will persist messages across sessions, and utilize past messages to inform responses.

Prerequisites

The example relies on these package versions for checkpointing functionality

    "langgraph>=0.6.7",
    "langgraph-checkpoint-postgres>=2.0.24",

Quick Setup (Local development)

Use docker compose to build a local version of this example. Write the necessary environment variables in a file called .env.local and .env.remote

OPENAI_API_KEY=...
WEATHER_API_KEY=...
TAVILY_API_KEY=...
AZURE_ENDPOINT=...
API_VERSION=...
AZURE_DEPLOYMENT=...
DB_PASSWORD="password"
DB_USERNAME="devuser"
DB_PORT="5432"
DB_NAME="langgraph"

mise run compose:local

For port-forwarding to the dev instance of postgres

Note: The Docker build downloads the AWS RDS certificate bundle. Make sure your VPN (or whichever private network gives access to AWS) is connected even when running the local profile.

Remote RDS Setup (via Port Forwarding)

To connect to a remote AWS RDS database through an EKS cluster:

1. Ensure your kubectl is configured to access your EKS cluster
2. Update the pod name and namespace in scripts/setup-port-forward-docker.sh if needed
3. Run with the remote profile:

mise run compose:remote

Production ready Setup (Kubernetes deployment)

You will need to package your agent as a Docker image and apply a pod definition within your Kubernetes namespace.

Endpoint in braid-delta namespace: gred-mlops-langgraph-tst-postgres.cluster-cmli0uctw7fb.us-west-2.rds.amazonaws.com

The credentials exist in GNESYS as a Kubernetes secret named langgraph-creds

The keys within the secret are described here:

Name:         langgraph-creds
Namespace:    braid-delta
Labels:       <none>
Annotations:  <none>

Type:  Opaque

Data
====
postgres-password:  64 bytes
postgres-port:      4 bytes
postgres-username:  9 bytes
postgres-db:        9 bytes
postgres-endpoint:  82 bytes

Create your env file

Set the following variables and save it as .env in this directory. See below for how to get root_certs.pem

OPENAI_API_KEY=""
WEATHER_API_KEY=""
TAVILY_API_KEY=""
AZURE_ENDPOINT=""
API_VERSION=""
AZURE_DEPLOYMENT=""
ROOT_CERT="root_certs.pem"

Connecting to Aurora Postgres

1. To connect to your Postgres instance, you'll have to use credentials stored in your Kubernetes namespace. The secret is langgraph-creds. They need to be loaded into your pod definition when you deploy your agent. See scripts/example-pod-def.yml.
2. The Dockerfile downloads the AWS RDS certificate bundle during the build (curl -o root_certs.pem https://truststore.pki.rds.amazonaws.com/${AWS_REGION}/${AWS_REGION}-bundle.pem). The download requires network access to AWS, so ensure VPN/private networking is available. If the build environment cannot reach AWS, download the file manually and place it at the repo root as root_certs.pem before building.

Build Docker image

Run the following command. The Dockerfile will copy source code, env variables, and your root_certs.pem file into the image. docker buildx build --platform=linux/amd64 -t <remote-repo>:<tag> -f scripts/Dockerfile --push .

Run in Kubernetes

Create a pod definition (use example in scripts/)

kubectl apply -f <pod-def>.yml

Checkpointer Usage

Example is implemented with async functions

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg_pool import AsyncConnectionPool
 pool = AsyncConnectionPool(
            conninfo=conn_string,
            max_size=20,
            max_lifetime=3600,
            kwargs={
                "row_factory": dict_row,
                "autocommit": True,
                "prepare_threshold": 0,
            },
        )

checkpointer = AsyncPostgresSaver(pool)
await checkpointer.setup()
compiled_graph = graph.compile(checkpointer=checkpointer)

Thread usage

Streaming LLM response

async with Thread(
        agent=agent
    ) as thread_manager:
        config = {
            "configurable": {"thread_id": thread_id, "user_id": user_id}
        }
        async for chunk in thread_manager.stream_agent_response(input_text, config):
            parsed_chunk = thread_manager.parse_stream_chunk(chunk)

Loading from checkpoint

async with Thread(
        agent=agent
    ) as thread_manager:
        config = {
            "configurable": {"thread_id": thread_id, "user_id": user_id}
        }
        messages = await thread_manager.populate_chat_history(config)

File Structure

src/
├── agent.py                # LangGraph agent definition
├── checkpoint_manager.py   # Checkpoint and graph compilation management
├── message_parser.py       # Message parsing and streaming
├── thread.py               # Orchestrated thread management
├── app.py                  # Streamlit application
└── settings.py             # Environment variable management

License

MIT