Building a Production-Ready Backend for LangGraph AI Agents

Transitioning an AI agent from a prototype running in a Jupyter Notebook to a production-ready application requires a fundamental shift in how you manage data. In a typical demo, developers rely on LangGraph's in-memory MemorySaver to track conversation state. However, as soon as the server restarts, all conversation history, user preferences, and intermediate agent states vanish.

For real-world applications—such as a booking assistant that manages flights, hotels, or appointments—you need a persistent backend. This backend must separate the agent's internal execution state (checkpoints) from the application's business data (bookings, user accounts).

In this guide, we will build a production-grade backend for a LangGraph agent using FastAPI, PostgreSQL, and SQLAlchemy. We will also configure the agent to use high-performance LLMs like Claude 3.5 Sonnet and DeepSeek-V3 routed through the unified API gateway at n1n.ai.

The Architecture: Agent State vs. Application State

Before writing code, it is crucial to understand the two distinct types of data your backend must manage:

Agent State (LangGraph Checkpoints): This is the metadata that tells LangGraph which node was executed last, what variables are currently in the graph's memory, and the history of messages within that specific thread. This is handled by a LangGraph checkpointer.
Application State (Business Database): This is your traditional relational database containing business entities (e.g., a bookings table with columns like booking_id, user_id, date, and status). The agent interacts with this data via tools (APIs or direct database queries), but it does not manage this table's schema directly.

Here is how the components interact:

[ User Client ] <---> [ FastAPI Backend ] <---> [ LangGraph Engine ]
                            |                          |
                            v                          v
                  [ Business DB Table ]      [ Postgres Checkpointer ]
                            |                          |
                            +---------> [ PostgreSQL ] <+

To ensure low latency and high availability when calling LLMs from our backend, we will use n1n.ai to route our LLM requests. This prevents us from having to manage multiple API providers and ensures fallback redundancy.

Step 1: Setting up the PostgreSQL Database

We will use PostgreSQL for both business data and LangGraph state persistence. First, let's define the database connection and the business schema for our booking system using SQLAlchemy.

# database.py
import os
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/agent_db")

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    name = Column(String)

class Booking(Base):
    __tablename__ = "bookings"
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    service_name = Column(String, nullable=False)
    booking_time = Column(DateTime, default=datetime.utcnow)
    status = Column(String, default="pending")  # pending, confirmed, cancelled

def init_db():
    Base.metadata.create_all(bind=engine)

Step 2: Implementing the LangGraph Postgres Checkpointer

LangGraph provides a native Postgres checkpointer class via the langgraph-checkpoint-postgres library. This replaces MemorySaver and automatically saves the state of your threads directly into PostgreSQL tables.

First, install the required dependency:

pip install langgraph-checkpoint-postgres psycopg

Now, let's initialize the checkpointer. In a production environment, you should use a connection pool to manage database connections efficiently:

# checkpointer.py
from contextlib import contextmanager
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver
import os

DB_URI = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/agent_db")

# Create a connection pool for the checkpointer
pool = ConnectionPool(conninfo=DB_URI, max_size=10)

@contextmanager
def get_checkpointer():
    with pool.connection() as conn:
        checkpointer = PostgresSaver(conn)
        # Ensure the checkpointer tables exist in the database
        checkpointer.setup()
        yield checkpointer

Step 3: Defining the LangGraph Agent and Tools

Our agent needs to interact with the database to create and retrieve bookings. We will write tool functions that the agent can call. These tools will receive the current user's context from the configuration.

To power our agent's reasoning, we will connect to n1n.ai. By using their unified endpoint, we can easily toggle between models like claude-3-5-sonnet for complex reasoning and deepseek-v3 for cost-effective processing.

# agent.py
from typing import Annotated, Dict, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from database import SessionLocal, Booking
import os

# Configure LangChain to use the n1n.ai API aggregator gateway
llm = ChatOpenAI(
    model="claude-3-5-sonnet",
    openai_api_key=os.getenv("N1N_API_KEY"),
    openai_api_base="https://api.n1n.ai/v1"
)

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    user_id: int

@tool
def create_booking(service_name: str, config: dict) -> str:
    """Book a service for the current user."""
    # Retrieve user_id from the graph configuration
    user_id = config["configurable"].get("user_id")
    if not user_id:
        return "Error: User not authenticated."

    db = SessionLocal()
    try:
        new_booking = Booking(user_id=user_id, service_name=service_name, status="confirmed")
        db.add(new_booking)
        db.commit()
        db.refresh(new_booking)
        return f"Successfully booked {service_name} (Booking ID: {new_booking.id})."
    except Exception as e:
        db.rollback()
        return f"Failed to create booking: {str(e)}"
    finally:
        db.close()

tools = [create_booking]
llm_with_tools = llm.bind_tools(tools)

def call_model(state: AgentState, config: dict):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages, config)
    return {"messages": [response]}

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_edge(START, "agent")

# We will compile the graph dynamically in the API layer to inject the checkpointer

Step 4: Creating the FastAPI Web Server

Now we will build the API layer. The FastAPI server will expose endpoints to chat with the agent. It will handle user authentication, retrieve the relevant thread ID, and pass these configurations to LangGraph.

# main.py
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from database import init_db, SessionLocal, User
from checkpointer import get_checkpointer
from agent import workflow
from langchain_core.messages import HumanMessage

app = FastAPI(title="LangGraph Production Backend")

@app.on_event("startup")
def startup_event():
    init_db()

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

@app.post("/chat")
def chat_with_agent(payload: ChatRequest):
    # Verify user exists in the business database
    db = SessionLocal()
    user = db.query(User).filter(User.id == payload.user_id).first()
    db.close()

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # Fetch the persistent checkpointer via context manager
    with get_checkpointer() as checkpointer:
        # Compile the graph with the persistent checkpointer
        compiled_graph = workflow.compile(checkpointer=checkpointer)

        # Configure the thread and metadata
        config = {
            "configurable": {
                "thread_id": payload.thread_id,
                "user_id": payload.user_id
            }
        }

        # Run the agent
        initial_state = {
            "messages": [HumanMessage(content=payload.message)],
            "user_id": payload.user_id
        }

        events = compiled_graph.stream(initial_state, config, stream_mode="values")

        # Extract the final response message
        final_message = ""
        for event in events:
            if "messages" in event:
                final_message = event["messages"][-1].content

        return {"response": final_message}

Comparison: State Persistence Solutions

When scaling your backend, choosing the right checkpointer is critical. Here is how PostgreSQL compares to other options:

Feature	MemorySaver (Demo)	PostgresSaver (Production)	Redis Checkpointer (Caching)
Persistence	Lost on restart	Permanent	Configurable TTL
Scalability	Single-instance only	Multi-instance / Horizontal	High throughput / Clustering
Setup Complexity	None	Medium (DB migration)	Medium (Redis instance)
Query Latency	< 1ms	5ms - 15ms	2ms - 5ms
Use Case	Prototyping	Business-critical data	High-concurrency chat

Pro Tips for Production Deployment

Thread Isolation & Security: Always validate that the authenticated user owns the thread_id they are requesting. Never trust the client-provided thread_id without verifying ownership in your application database.
Transaction Management: When writing tools that modify your business database, ensure you handle database sessions correctly. If an agent execution fails mid-run, you don't want partial database commits. Use SQLAlchemy context managers to guarantee clean rollbacks.
LLM Fallbacks with n1n.ai: If your primary model (e.g., Claude 3.5 Sonnet) hits a rate limit or encounters an outage, you can easily implement a fallback mechanism in your backend code to switch models. Since n1n.ai aggregates multiple providers, you can change the target model in your API call without having to change your SDK configurations or install new libraries.
Handling Long-Running Agents: If your agent has complex loops or human-in-the-loop steps, do not block the HTTP request. Instead, run the agent in a background task (using FastAPI's BackgroundTasks or Celery) and expose a WebSocket or polling endpoint to check the agent's status.

Testing the System

To test this setup, write a simple script to register a user and send a message:

# test.py
import requests
from database import SessionLocal, User, engine

# Create a test user directly in the database
db = SessionLocal()
if not db.query(User).filter(User.email == "[email protected]").first():
    dev_user = User(email="[email protected]", name="Developer")
    db.add(dev_user)
    db.commit()
db.close()

# Send a request to the FastAPI server
response = requests.post("http://localhost:8000/chat", json={
    "message": "Please book a flight ticket for me.",
    "thread_id": "session_abc_123",
    "user_id": 1
})
print(response.json())

By separating concerns between the LangGraph execution flow and your application's relational data, you create a robust foundation capable of scaling to thousands of concurrent users.

Get a free API key at n1n.ai

Source: https://towardsdatascience.com/building-a-proper-backend-for-my-langgraph-ai-agent/