Xây dựng Backend hoàn chỉnh cho AI Agent LangGraph: Từ Demo đến Sản phẩm thực tế

Trong loạt bài trước, chúng ta đã xây dựng một AI agent sử dụng LangGraph để xử lý toàn bộ quy trình đặt lịch hẹn trong 15 phút và tích hợp giao diện Streamlit. Tuy nhiên, toàn bộ dữ liệu vẫn được lưu trong bộ nhớ tạm, khiến hệ thống chỉ hoạt động tốt ở mức demo. Bài viết này sẽ hướng dẫn bạn thay thế lớp lưu trữ nội bộ bằng PostgreSQL — một bước quan trọng để biến ứng dụng thành một sản phẩm thực thụ.

Kiến trúc agent với backend PostgreSQL

Vì sao cần một cơ sở dữ liệu thực sự?

Giới hạn của bộ nhớ trong (in-memory)

Agent hiện tại sử dụng hai đối tượng Python lưu trong tiến trình:

• Checkpointer của LangGraph: lưu trạng thái đồ thị sau mỗi bước thực thi, nhưng nằm trong MemorySaver.
• Danh sách Python có khóa (lock): lưu các cuộc hẹn đã xác nhận trong lớp InMemoryBookingRepository.

Vấn đề nghiêm trọng xuất hiện khi:

• Khởi động lại tiến trình: toàn bộ checkpoint và lịch hẹn biến mất.
• Không chia sẻ dữ liệu giữa các phiên: Session A không thể thấy lịch của Session B, dẫn đến đặt trùng lịch.
• Thiếu tính nhất quán: agent có thể đề xuất khung giờ dựa trên dữ liệu cũ, rồi xác nhận một lịch mà người khác đã chiếm.

Đây là rào cản lớn nếu muốn đưa sản phẩm ra thị trường — nơi nhiều khách hàng tương tác cùng lúc qua nhiều kênh (WhatsApp, web, ứng dụng di động).

Kiến trúc mới với PostgreSQL

Sơ đồ so sánh kiến trúc cũ và mới

Sau khi tích hợp PostgreSQL, kiến trúc thay đổi như sau:

• AgentState vẫn là "bộ nhớ làm việc" của đồ thị, nhưng được lưu bền vững qua checkpointer.
• BookingRepository — giao diện trừu tượng — có thể được triển khai bằng Postgres hoặc bộ nhớ trong.

Mục tiêu là giữ cho đồ thị và các engine phụ thuộc vào một interface ổn định, không phụ thuộc trực tiếp vào Postgres hay bộ nhớ.

Triển khai chi tiết

1. Định nghĩa Protocol

Trước tiên, tạo một protocol để đồ thị làm việc với repository một cách linh hoạt:

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."""

Với protocol này, ta có thể cắm PostgresBookingRepository hoặc InMemoryBookingRepository lúc khởi động. Chế độ in-memory vẫn hữu ích cho unit test và demo nhanh — nhưng dữ liệu sẽ mất khi app khởi động lại.

2. Cấu trúc database

Trong file postgres.py, schema gồm hai bảng:

• technicians: danh sách kỹ thuật viên.
• bookings: các cuộc hẹn đã xác nhận.

Hàm create_persistence() quyết định chọn Postgres hay in-memory dựa trên biến môi trường DATABASE_URL:

• Nếu có DATABASE_URL: cả booking repository và LangGraph checkpointer dùng Postgres.
• Nếu không: cả hai nằm trong bộ nhớ.

Repository được truyền vào hàm build_graph():

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)
    # ... truncated

3. Tương tác với database trong quy trình agent

Quy trình xử lý đặt lịch của agent

AgentState đóng vai trò bộ nhớ làm việc, chứa:

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]

Chỉ hai node tương tác trực tiếp với repository:

• generate_options (đọc): gọi list_bookings() — thực chất là câu SELECT khi dùng Postgres — để tính khung giờ trống.
• confirm_booking (ghi): gọi create_booking() để chèn lịch hẹn mới.

Các node khác chỉ đọc/cập nhật AgentState, không query bảng bookings.

Ví dụ node tạo lịch đề xuất:

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))],
    }

Và node xác nhận:

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}."
                )
            )
        ],
    }

LangGraph tự động merge dữ liệu trả về vào AgentState và lưu snapshot qua checkpointer cho từng thread_id.

Tổng kết

Việc chuyển từ bộ nhớ trong sang PostgreSQL đã biến agent từ một "demo notebook kernel" thành một sản phẩm đáng tin cậy:

• Dữ liệu bền vững khi khởi động lại.
• Nhiều phiên/kênh dùng chung một nguồn dữ liệu — không còn đặt trùng lịch.
• Kiến trúc rõ ràng, dễ mở rộng.

Trong bài tiếp theo, tác giả sẽ hướng dẫn chạy và kiểm thử hệ thống với Docker, cũng như kết nối với PostgreSQL trên cloud. Toàn bộ mã nguồn có trên GitHub tại customer-service-agent — bạn có thể clone và tự trải nghiệm.