of this series, I built a stateful LangGraph agent that handles a 15-minute booking process and wrapped it up with a Streamlit UI to improve user experience.
The agent handles the entire booking process like a real customer service representative. It’s a LangGraph-based agent that orchestrates the following operations:
- Responds to customer queries and understands their needs.
- Calculates the price for the service and informs the customer.
- Handles the customer’s acceptance or rejection.
- Proposes optimized time slots.
- Confirms and records the appointment.
The next phase is to build a proper backend and we start by implementing a Postgres database instead of keeping everything in memory.
We keep Streamlit as the user interface and replace the in-memory adapters with PostgreSQL.
This will also allow us to have multiple fronts (e.g. WhatsApp, Streamlit) that share the same backend. So we are turning this into a proper product that will handle a real business.
The full source code of this project is available on GitHub at customer-service-agent. Feel free to clone the repo and test it yourself.
What the database looks like now
It’s hard to even call it a database as it’s just two Python objects that lived inside the process:
The first one is a LangGraph checkpointer, which is a state persistence layer that saves a snapshot of an agent’s graph state at every step of execution.
When the graph is compiled, conversation state is stored in memory.
graph.compile(checkpointer=checkpointer or MemorySaver())The checkpointer allows the agent resume across turns. If we don’t have it, every customer message would be a new conversation.
The second object is a Python list behind a lock. Confirmed appointments are stored in an in-memory repository that looks like this:
class InMemoryBookingRepository:
def __init__(self) -> None:
self._lock = threading.RLock()
self.technicians = {...} # hardcoded cleaners
self._bookings: list[Booking] = []
def list_bookings(self) -> list[Booking]:
with self._lock:
return list(self._bookings)
def create_booking(self, option, details, price) -> Booking:
# check overlap in Python, then append to self._bookings
...The scheduling engine called list_bookings() to avoid double-booking. Confirmation called create_booking(), which re-checked overlap and appended to the list.
This is a very simple structure designed for initial testing and demo purposes. It allows us to test LangGraph routing and logic.
Why we need a proper database
The current “database” relies on in-memory persistence so it fails as soon as we leave a single demo process.
When the process restarts, conversation checkpoints and bookings vanish.
Since it’s in memory, there is no shared availability. Session A cannot see bookings created by Session B, which means every process has its own calendar.
Even worse for a booking product is that the agent can offer a slot based on a stale in-memory view, then “confirm” a booking that another session already took.
When we use the Streamlit UI, it looked like a product but the storage still behaved like a notebook kernel.
Long story short, we need a proper database for our agent to be considered as a product.
We will use Postgres, which is a free and open-source relational database system. We need a relational database with bookings and technician information stored in separate (and related) tables.
Before Postgres implementation, the agent structure looks like this:
And after we complete Postgres implementation, it will look like this:
After the Postgres backend, AgentState will still be the working memory of the graph but it will be persisted through a checkpointer and a booking engine.
We will learn how these are implemented and they function in the remaining part of the article.
Postgres implementation
We first create a protocol so that the graph and engines can depend on a stable interface, not on Postgres (or memory) specifically.
from typing import Protocol
class BookingRepository(Protocol):
"""Persistence interface used by scheduling and confirmation."""
@property
def technicians(self) -> dict[str, Technician]:
"""Return technicians keyed by id."""
def list_bookings(self) -> list[Booking]:
"""Return all confirmed bookings."""
def create_booking(
self, option: TimeOption, details: BookingDetails, price: float
) -> Booking:
"""Persist a booking after re-checking overlap; raise ValueError if taken."""With this protocol, we just plug PostgresBookingRepository or InMemoryBookingRepository at startup (depending on using Postgres or in-memory). Then, the nodes can call list_bookings and create_booking functions.
When we use InMemoryBookingRepository, no database tables are created. Confirmed bookings are kept in a Python list inside the running process, and the same repository methods (list_bookings, create_booking) still work. They just never touch Postgres.
The in-memory mode ideal for unit tests and quick local demos. It’s important to also mention that, with the in-memory mode, everything disappears when the app restarts.
When we use PostgresBookingRepository , there is an actual database. Inside the postgres.py script, you can see the database schema that consists of two tables, which are technicians and bookings .
You can also see the definition of the PostgresBookingRepository class. I won’t copy it here because it’s close to 100 lines of code. We also define the functions list_bookings and create_booking inside this class.
At app startup, create_persistence() chooses Postgres vs in-memory. When DATABASE_URL is set, both the booking repository and LangGraph checkpointer use Postgres. Otherwise both stay in memory.
So the repository is either a PostgresBookingRepository or InMemoryBookingRepository (both satisfy the BookingRepository protocol), and that instance is passed into the build_graph function:
def build_graph(
llm: BaseChatModel,
*,
repository: BookingRepository | None = None,
checkpointer: Any | None = None,
) -> Any:
"""Build a compiled, multi-turn booking graph."""
repository = repository or InMemoryBookingRepository()
graph = StateGraph(AgentState)
# truncatedThe repository is then used by the graph nodes to interact with the database.
For example, we define the confirm_booking_node function as follows:
def confirm_booking_node(state: AgentState) -> dict[str, Any]:
option = state.get("selected_slot")
if option is None:
raise ValueError("A slot must be selected before confirmation.")
booking = repository.create_booking(
option, state["booking_details"], float(state["calculated_price"])
)
return {
"booking_id": booking.id,
"status": "confirmed",
"messages": [
AIMessage(
content=(
f"Confirmed! Booking {booking.id} is scheduled for "
f"{option.start_at}. Your total is ${booking.price:.2f}."
)
)
],
}We can see that it’s using the repository to create a booking in the database.
Database interactions
The current agentic workflow is as follows:
During a booking session, conversation lives in AgentState, which can be considered as the working memory of the graph. Each node returns a partial update, and LangGraph merges it into that state. The checkpointer persists it across turns with Postgres.
Only two nodes talk to the booking repository (either PostgresBookingRepository or InMemoryBookingRepository):
- generate options (read): Load existing bookings (+ technicians), then compute free slots
- confirm booking (write): Insert the confirmed appointment
The other nodes only read or update the AgentState. They do not query the bookings table.
To propose appointments, we add generate_schedule_options_node to the graph:
def generate_schedule_options_node(state: AgentState) -> dict[str, Any]:
options = generate_schedule_options(state["booking_details"], repository)
lines = ["Great—please choose one of these optimized appointments:"]
for index, option in enumerate(options, 1):
lines.append(f"{index}. {option.start_at} ({option.technician_id})")
return {
"time_options": options,
"status": "awaiting_slot_selection",
"messages": [AIMessage(content="n".join(lines))],
}This node calls the generate_schedule_options() function from engines.py , which:
- Calls
repository.list_bookings()(aSELECTfrombookingswhen using Postgres) - Uses
repository.technicians - Applies deterministic rules (next 7 days, skip Sundays, fixed start times, duration, travel scoring)
- Returns the best 3 available time options
LangGraph handles merging this information into AgentState, updating time_options, status, and messages :
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
booking_details: BookingDetails
calculated_price: NotRequired[float | None]
time_options: NotRequired[list[TimeOption]]
selected_slot: NotRequired[TimeOption | None]
status: BookingStatus
booking_id: NotRequired[str | None]After the customer confirms a slot, select_slot_node only sets selected_slot in AgentState. The write happens in confirm_booking_node:
def confirm_booking_node(state: AgentState) -> dict[str, Any]:
option = state.get("selected_slot")
if option is None:
raise ValueError("A slot must be selected before confirmation.")
booking = repository.create_booking(
option, state["booking_details"], float(state["calculated_price"])
)
return {
"booking_id": booking.id,
"status": "confirmed",
"messages": [
AIMessage(
content=(
f"Confirmed! Booking {booking.id} is scheduled for "
f"{option.start_at}. Your total is ${booking.price:.2f}."
)
)
],
}On success, LangGraph merges booking_id, status="confirmed", and the confirmation message into AgentState, and the checkpointer saves that snapshot for the conversation thread_id.
We now have a proper Postgres backend for our customer service agent. In the next article, I’ll walk through how to run and verify this setup with Docker, and how to point the same app at a hosted Postgres instance.
Thank you for reading.