LangGraphのAIエージェントをデモから本番へ — インメモリをPostgreSQLに置き換えて再起動しても消えないバックエンドを作る

8月23日、Soner Yıldırımが「Building a Proper Backend for My LangGraph AI Agent」と題した記事を公開した。LangGraphで構築したAI予約エージェントのストレージをインメモリからPostgreSQLに移行する手順を詳しく紹介している。プロセスを再起動すると会話履歴も予約データもすべて消えるというインメモリの根本的な欠陥を解消し、本番運用に耐えるバックエンドを整備することが本稿のテーマだ。

デモから本番へ：インメモリの限界

Soner Yıldırımはこれまでのシリーズで、15分かかっていた予約プロセスを自動化するLangGraphエージェントを構築し、Streamlit UIも整えてきた。エージェントは顧客対応・価格計算・時間枠提案・予約確定までを一気通貫で処理する。

問題はストレージだ。これまでの実装には2つのインメモリオブジェクトしか存在しなかった。

LangGraphチェックポインター（MemorySaver）：グラフの状態スナップショットを保存し、会話のターンをまたいで継続させる（※LangGraphチェックポインターとは、エージェントの会話状態をthread_id単位でシリアライズ・保存する仕組みで、マルチターン対話の実現に不可欠なコンポーネント）
Pythonリスト（InMemoryBookingRepository）：確定済み予約をスレッドロック付きのリストで管理する

class InMemoryBookingRepository:
    def __init__(self) -> None:
        self._lock = threading.RLock()
        self.technicians = {...}  # ハードコードされた担当者
        self._bookings: list[Booking] = []

この構造の致命的な欠点は明快だ。プロセスを再起動すると会話履歴も予約データも消える。さらに、セッションAはセッションBが作成した予約を参照できないため、二重予約のチェックが機能しない。Streamlit UIを被せると製品らしく見えるが、ストレージの挙動はノートブックのカーネルと変わらない。

設計の核心：Protocolで差し替え可能にする

移行の肝は、グラフのノードをPostgres固有の実装に依存させないことだ。そのためにPythonのProtocolでインターフェースを定義する。

from typing import Protocol

class BookingRepository(Protocol):
    @property
    def technicians(self) -> dict[str, Technician]: ...
    def list_bookings(self) -> list[Booking]: ...
    def create_booking(
        self, option: TimeOption, details: BookingDetails, price: float
    ) -> Booking: ...

このプロトコルを満たす実装として、InMemoryBookingRepository（テスト・ローカルデモ用）とPostgresBookingRepository（本番用）を用意する。どちらを使うかはアプリ起動時に決定される。環境変数DATABASE_URLが設定されていれば予約リポジトリとLangGraphチェックポインターの両方がPostgresを使い、未設定ならインメモリにフォールバックする。

def build_graph(
    llm: BaseChatModel,
    *,
    repository: BookingRepository | None = None,
    checkpointer: Any | None = None,
) -> Any:
    repository = repository or InMemoryBookingRepository()
    graph = StateGraph(AgentState)
    # ...

グラフのノードはrepositoryを通じてDBと対話する。ノード側のコードはPostgresかインメモリかを意識しない。

DBにアクセスするノードは2つだけ

アーキテクチャとして整理しておきたいのが、ブッキングリポジトリに触れるノードは全体のうち2つだけという点だ。

generate_schedule_options_node（読み取り）：既存の予約と担当者情報をロードし、空き時間枠を計算して上位3件を返す。内部では「次の7日間、日曜除く、固定開始時間、移動距離スコアリング」という決定的なルールを適用する
confirm_booking_node（書き込み）：選択されたスロットで予約をDBに挿入する

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

その他のノードはAgentState（グラフの作業メモリ）を読み書きするだけでDBには触れない。LangGraphはノードが返す部分的な更新をAgentStateにマージし、チェックポインターがそのスナップショットをthread_idに紐付けてPostgresに保存する。

全体の構成変化

移行前後のアーキテクチャをSonerは図で示している（詳細は元記事の図を参照）。

移行前：StreamlitはPythonプロセス内のインメモリオブジェクトと直接対話する

移行後：チェックポインターと予約エンジンがPostgresを介してデータを永続化し、StreamlitやWhatsAppなど複数のフロントエンドが同じバックエンドを共有できる

PostgresBookingRepositoryの実装はpostgres.pyに約100行にわたって定義されており、techniciansテーブルとbookingsテーブルの2テーブル構成を取る。ソースコードはGitHubのcustomer-service-agentリポジトリで公開されている。

次回記事では、DockerでこのPostgres構成を実行・検証する方法と、ホスト型Postgresインスタンスへの接続方法を解説すると予告している。

詳細はBuilding a Proper Backend for My LangGraph AI Agentを参照していただきたい。