of this sequence, I constructed a stateful LangGraph agent that handles a 15-minute reserving course of and wrapped it up with a Streamlit UI to enhance person expertise.
The agent handles your entire reserving course of like an actual customer support consultant. It’s a LangGraph-based agent that orchestrates the next operations:
- Responds to buyer queries and understands their wants.
- Calculates the value for the service and informs the client.
- Handles the client’s acceptance or rejection.
- Proposes optimized time slots.
- Confirms and data the appointment.
The subsequent section is to construct a correct backend and we begin by implementing a Postgres database as an alternative of holding every part in reminiscence.
We hold Streamlit because the person interface and change the in-memory adapters with PostgreSQL.
This will even permit us to have a number of fronts (e.g. WhatsApp, Streamlit) that share the identical backend. So we’re turning this into a correct product that can deal with an actual enterprise.
The complete supply code of this mission is offered on GitHub at customer-service-agent. Be at liberty to clone the repo and check it your self.
What the database appears to be like like now
It’s onerous to even name it a database because it’s simply two Python objects that lived inside the method:
The primary one is a LangGraph checkpointer, which is a state persistence layer that saves a snapshot of an agent’s graph state at each step of execution.
When the graph is compiled, dialog state is saved in reminiscence.
graph.compile(checkpointer=checkpointer or MemorySaver())The checkpointer permits the agent resume throughout turns. If we don’t have it, each buyer message could be a brand new dialog.
The second object is a Python listing behind a lock. Confirmed appointments are saved in an in-memory repository that appears like this:
class InMemoryBookingRepository:
def __init__(self) -> None:
self._lock = threading.RLock()
self.technicians = {...} # hardcoded cleaners
self._bookings: listing[Booking] = []
def list_bookings(self) -> listing[Booking]:
with self._lock:
return listing(self._bookings)
def create_booking(self, possibility, particulars, worth) -> Reserving:
# examine overlap in Python, then append to self._bookings
...The scheduling engine known as list_bookings() to keep away from double-booking. Affirmation known as create_booking(), which re-checked overlap and appended to the listing.
This can be a quite simple construction designed for preliminary testing and demo functions. It permits us to check LangGraph routing and logic.
Why we’d like a correct database
The present “database” depends on in-memory persistence so it fails as quickly as we go away a single demo course of.
When the method restarts, dialog checkpoints and bookings vanish.
Because it’s in reminiscence, there isn’t any shared availability. Session A can’t see bookings created by Session B, which suggests each course of has its personal calendar.
Even worse for a reserving product is that the agent can supply a slot primarily based on a stale in-memory view, then “verify” a reserving that one other session already took.
After we use the Streamlit UI, it seemed like a product however the storage nonetheless behaved like a pocket book kernel.
Lengthy story brief, we’d like a correct database for our agent to be thought-about as a product.
We’ll use Postgres, which is a free and open-source relational database system. We’d like a relational database with bookings and technician data saved in separate (and associated) tables.
Earlier than Postgres implementation, the agent construction appears to be like like this:
And after we full Postgres implementation, it would appear like this:
After the Postgres backend, AgentState will nonetheless be the working reminiscence of the graph however it will likely be persevered by way of a checkpointer and a reserving engine.
We’ll learn the way these are applied they usually perform within the remaining a part of the article.
Postgres implementation
We first create a protocol in order that the graph and engines can depend upon a secure interface, not on Postgres (or reminiscence) particularly.
from typing import Protocol
class BookingRepository(Protocol):
"""Persistence interface utilized by scheduling and affirmation."""
@property
def technicians(self) -> dict[str, Technician]:
"""Return technicians keyed by id."""
def list_bookings(self) -> listing[Booking]:
"""Return all confirmed bookings."""
def create_booking(
self, possibility: TimeOption, particulars: BookingDetails, worth: float
) -> Reserving:
"""Persist a reserving after re-checking overlap; elevate ValueError if taken."""With this protocol, we simply plug PostgresBookingRepository or InMemoryBookingRepository at startup (relying on utilizing Postgres or in-memory). Then, the nodes can name list_bookings and create_booking capabilities.
After we use InMemoryBookingRepository, no database tables are created. Confirmed bookings are saved in a Python listing contained in the working course of, and the identical repository strategies (list_bookings, create_booking) nonetheless work. They only by no means contact Postgres.
The in-memory mode ultimate for unit exams and fast native demos. It’s essential to additionally point out that, with the in-memory mode, every part disappears when the app restarts.
After we use PostgresBookingRepository , there’s an precise database. Contained in the postgres.py script, you possibly can see the database schema that consists of two tables, that are technicians and bookings .
It’s also possible to see the definition of the PostgresBookingRepository class. I gained’t copy it right here as a result of it’s near 100 traces of code. We additionally outline the capabilities list_bookings and create_booking inside this class.
At app startup, create_persistence() chooses Postgres vs in-memory. When DATABASE_URL is about, each the reserving repository and LangGraph checkpointer use Postgres. In any other case each keep in reminiscence.
So the repository is both a PostgresBookingRepository or InMemoryBookingRepository (each fulfill the BookingRepository protocol), and that occasion is handed into the build_graph perform:
def build_graph(
llm: BaseChatModel,
*,
repository: BookingRepository | None = None,
checkpointer: Any | None = None,
) -> Any:
"""Construct a compiled, multi-turn reserving graph."""
repository = repository or InMemoryBookingRepository()
graph = StateGraph(AgentState)
# truncatedThe repository is then utilized by the graph nodes to work together with the database.
For instance, we outline the confirm_booking_node perform as follows:
def confirm_booking_node(state: AgentState) -> dict[str, Any]:
possibility = state.get("selected_slot")
if possibility is None:
elevate ValueError("A slot should be chosen earlier than affirmation.")
reserving = repository.create_booking(
possibility, state["booking_details"], float(state["calculated_price"])
)
return {
"booking_id": reserving.id,
"standing": "confirmed",
"messages": [
AIMessage(
content=(
f"Confirmed! Booking {booking.id} is scheduled for "
f"{option.start_at}. Your total is ${booking.price:.2f}."
)
)
],
}We are able to see that it’s utilizing the repository to create a reserving within the database.
Database interactions
The present agentic workflow is as follows:
Throughout a reserving session, dialog lives in AgentState, which might be thought-about because the working reminiscence of the graph. Every node returns a partial replace, and LangGraph merges it into that state. The checkpointer persists it throughout turns with Postgres.
Solely two nodes discuss to the reserving repository (both PostgresBookingRepository or InMemoryBookingRepository):
- generate choices (learn): Load current bookings (+ technicians), then compute free slots
- verify reserving (write): Insert the confirmed appointment
The opposite nodes solely learn or replace the AgentState. They don’t question the bookings desk.
To suggest appointments, we add generate_schedule_options_node to the graph:
def generate_schedule_options_node(state: AgentState) -> dict[str, Any]:
choices = generate_schedule_options(state["booking_details"], repository)
traces = ["Great—please choose one of these optimized appointments:"]
for index, possibility in enumerate(choices, 1):
traces.append(f"{index}. {possibility.start_at} ({possibility.technician_id})")
return {
"time_options": choices,
"standing": "awaiting_slot_selection",
"messages": [AIMessage(content="n".join(lines))],
}This node calls the generate_schedule_options() perform from engines.py , which:
- Calls
repository.list_bookings()(aSELECTfrombookingswhen utilizing Postgres) - Makes use of
repository.technicians - Applies deterministic guidelines (subsequent 7 days, skip Sundays, fastened begin instances, length, journey scoring)
- Returns the finest 3 obtainable time choices
LangGraph handles merging this data into AgentState, updating time_options, standing, 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]
standing: BookingStatus
booking_id: NotRequired[str | None]After the client confirms a slot, select_slot_node solely units selected_slot in AgentState. The write occurs in confirm_booking_node:
def confirm_booking_node(state: AgentState) -> dict[str, Any]:
possibility = state.get("selected_slot")
if possibility is None:
elevate ValueError("A slot should be chosen earlier than affirmation.")
reserving = repository.create_booking(
possibility, state["booking_details"], float(state["calculated_price"])
)
return {
"booking_id": reserving.id,
"standing": "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, standing="confirmed", and the affirmation message into AgentState, and the checkpointer saves that snapshot for the dialog thread_id.
We now have a correct Postgres backend for our customer support agent. Within the subsequent article, I’ll stroll by way of find out how to run and confirm this setup with Docker, and find out how to level the identical app at a hosted Postgres occasion.
Thanks for studying.
