Constructing a Correct Backend for My LangGraph AI Agent

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 consumer expertise.

The agent handles all the 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 shopper.
• Handles the shopper’s acceptance or rejection.
• Proposes optimized time slots.
• Confirms and data the appointment.

The following part is to construct a correct backend and we begin by implementing a Postgres database as a substitute of maintaining every thing in reminiscence.

We maintain Streamlit because the consumer interface and exchange the in-memory adapters with PostgreSQL.

This may 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 undertaking is offered on GitHub at customer-service-agent. Be at liberty to clone the repo and take a look at it your self.

What the database appears to be like like now

It’s laborious 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 record 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: record[Booking] = []

    def list_bookings(self) -> record[Booking]:
        with self._lock:
            return record(self._bookings)

    def create_booking(self, possibility, particulars, worth) -> Reserving:
        # test overlap in Python, then append to self._bookings
        ...

The scheduling engine referred to as list_bookings() to keep away from double-booking. Affirmation referred to as create_booking(), which re-checked overlap and appended to the record.

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 want 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 is no such thing as a shared availability. Session A can’t see bookings created by Session B, which implies each course of has its personal calendar.

Even worse for a reserving product is that the agent can provide a slot based mostly on a stale in-memory view, then “affirm” a reserving that one other session already took.

Once we use the Streamlit UI, it appeared like a product however the storage nonetheless behaved like a pocket book kernel.

Lengthy story quick, we want 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 want 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 should appear like this:

After the Postgres backend, AgentState will nonetheless be the working reminiscence of the graph however it is going to be endured by means of a checkpointer and a reserving engine.

We’ll learn the way these are applied and so they operate within the remaining a part of the article.

Postgres implementation

We first create a protocol in order that the graph and engines can rely upon a steady 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) -> record[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.

Once we use InMemoryBookingRepository, no database tables are created. Confirmed bookings are saved in a Python record 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 best for unit checks and fast native demos. It’s essential to additionally point out that, with the in-memory mode, every thing disappears when the app restarts.

Once we use PostgresBookingRepository , there may be an precise database. Contained in the postgres.py script, you’ll be able to see the database schema that consists of two tables, that are technicians and bookings .

You may also see the definition of the PostgresBookingRepository class. I gained’t copy it right here as a result of it’s near 100 strains 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 ready, 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 operate:

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)

    # truncated

The repository is then utilized by the graph nodes to work together with the database.

For instance, we outline the confirm_booking_node operate as follows:

    def confirm_booking_node(state: AgentState) -> dict[str, Any]:
        possibility = state.get("selected_slot")
        if possibility is None:
            elevate ValueError("A slot have to 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 will 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 could 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 present bookings (+ technicians), then compute free slots
• affirm 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)
    strains = ["Great—please choose one of these optimized appointments:"]
    for index, possibility in enumerate(choices, 1):
        strains.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() operate from engines.py , which:

1. Calls repository.list_bookings() (a SELECT from bookings when utilizing Postgres)
2. Makes use of repository.technicians
3. Applies deterministic guidelines (subsequent 7 days, skip Sundays, fastened begin instances, period, journey scoring)
4. Returns the greatest 3 accessible 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 shopper 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 have to 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 means of how one can run and confirm this setup with Docker, and how one can level the identical app at a hosted Postgres occasion.

Thanks for studying.