LangGraph Tutorial (Python): connecting to PostgreSQL for beginners
This tutorial shows you how to build a small LangGraph app in Python that reads and writes state in PostgreSQL. You need this when your agent must survive process restarts, share state across workers, or keep conversation history outside memory.
What You'll Need
- •Python 3.10+
- •A running PostgreSQL instance
- •A PostgreSQL connection string, for example:
- •
postgresql://postgres:postgres@localhost:5432/langgraph_demo
- •
- •
pipfor installing packages - •These Python packages:
- •
langgraph - •
psycopg[binary] - •
python-dotenvif you want to load env vars from a.envfile
- •
- •Basic familiarity with:
- •LangGraph state graphs
- •Python functions and type hints
Step-by-Step
- •Install the dependencies.
We need LangGraph plus a PostgreSQL driver that supports the checkpoint store. Keep this minimal and pin versions in your own project later.
pip install langgraph psycopg[binary] python-dotenv
- •Create a PostgreSQL database and set your connection string.
The example below assumes you already have a database calledlanggraph_demo. If you prefer environment variables, put the URL in.envso your code stays clean.
export POSTGRES_URL="postgresql://postgres:postgres@localhost:5432/langgraph_demo"
- •Build a simple graph with one node that updates state.
This graph takes a counter in state, increments it, and returns the new value. The important part is that we compile it with a PostgreSQL checkpointer so state can be persisted between runs.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
class CounterState(TypedDict):
count: int
def increment(state: CounterState) -> CounterState:
return {"count": state["count"] + 1}
builder = StateGraph(CounterState)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)
POSTGRES_URL = "postgresql://postgres:postgres@localhost:5432/langgraph_demo"
with PostgresSaver.from_conn_string(POSTGRES_URL) as checkpointer:
checkpointer.setup()
app = builder.compile(checkpointer=checkpointer)
- •Run the graph with a thread ID so LangGraph can store checkpoints per conversation.
Thethread_idis the key that tells LangGraph which saved state belongs to which run. Run it twice with the same thread ID and you will see persisted state behavior.
from langchain_core.runnables import RunnableConfig
config = RunnableConfig(configurable={"thread_id": "demo-thread-1"})
result1 = app.invoke({"count": 0}, config=config)
print("First run:", result1)
result2 = app.invoke({"count": result1["count"]}, config=config)
print("Second run:", result2)
- •Load the latest checkpoint back from PostgreSQL.
This is the part beginners usually want to confirm first: the data is not only in memory. You can inspect the stored state by asking LangGraph for the latest snapshot using the same thread ID.
snapshot = app.get_state(config)
print("Latest saved state:", snapshot.values)
print("Next nodes:", snapshot.next)
- •Wrap it into a reusable script.
In production, keep your graph definition separate from runtime configuration so you can reuse the same graph across API workers, batch jobs, and tests.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_core.runnables import RunnableConfig
class CounterState(TypedDict):
count: int
def increment(state: CounterState) -> CounterState:
return {"count": state["count"] + 1}
def build_app(postgres_url: str):
builder = StateGraph(CounterState)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)
with PostgresSaver.from_conn_string(postgres_url) as checkpointer:
checkpointer.setup()
return builder.compile(checkpointer=checkpointer)
app = build_app(POSTGRES_URL)
config = RunnableConfig(configurable={"thread_id": "demo-thread-2"})
print(app.invoke({"count": 10}, config=config))
Testing It
Run the script once and confirm you get an output like {'count': 1} or {'count': 11} depending on your input. Then run it again with the same thread_id and verify that app.get_state(config) returns the saved checkpoint from PostgreSQL.
If nothing persists, check these first:
- •Your PostgreSQL server is running
- •The database name in
POSTGRES_URLexists - •The user has permission to create tables
- •You called
checkpointer.setup()
A good sanity test is to stop your Python process after one invocation, restart it, and query the same thread again. If the state comes back, your LangGraph-to-PostgreSQL connection is working correctly.
Next Steps
- •Add a second node and route between nodes using conditional edges.
- •Replace the counter with real agent state, such as messages or form fields.
- •Add FastAPI around this graph so each HTTP request gets its own
thread_id.
Keep learning
- •The complete AI Agents Roadmap — my full 8-step breakdown
- •Free: The AI Agent Starter Kit — PDF checklist + starter code
- •Work with me — I build AI for banks and insurance companies
By Cyprian Aarons, AI Consultant at Topiax.
Want the complete 8-step roadmap?
Grab the free AI Agent Starter Kit — architecture templates, compliance checklists, and a 7-email deep-dive course.
Get the Starter Kit