← Kembali ke Blog
Oct 24, 2026 · 10 menit baca

Agent Kamu Mati di Tengah Jalan. LangGraph dan Postgres yang Lanjutin

AI AgentsLangGraphPostgresPythonTutorial

Langkah approval itu momen di mana agent berhenti jadi demo. Agent checkout validasi order, panggil interrupt() buat minta persetujuan manusia atas charge 499 dolar, lalu worker-nya kena redeploy sebelum ada yang sempat klik approve. Pod hilang. Approval yang pending ikut hilang, karena cuma numpang di dict Python.

Customer coba lagi. Sekarang ada dua charge dan satu thread support yang bingung.

Itu bukan model yang gagal. Itu state yang gagal, dan obatnya membosankan: tulis run-nya ke Postgres di setiap langkah supaya proses lain bisa lanjutin. Di bawah ini kita bangun graph tiga node yang berhenti buat approval, matikan prosesnya, lanjutin di container baru, lalu bedah empat jebakan yang bikin pola ini rusak di production.

Yang perlu disiapkan

  • Docker buat container Postgres, atau Postgres 14+ apa saja
  • Python 3.10+
  • Sekitar 15 menit
pip install -U langgraph "langgraph-checkpoint-postgres" "psycopg[binary,pool]"

Semua contoh di bawah jalan di langgraph 1.2.16, versi yang dipasang pip saat artikel ini ditulis.

Langkah 1: Postgres yang umurnya lebih panjang dari proses kamu

docker run -d --name agent-pg -e POSTGRES_PASSWORD=postgres -p 5432:5432 postgres:16-alpine

Lalu export connection string-nya:

export DB_URI="postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

sslmode=disable aman buat container lokal. Buat koneksi remote, tetap pakai TLS.

Langkah 2: graph yang butuh manusia

Simpan sebagai demo.py. Tiga node: validate, charge, notify. Cuma charge yang interrupt.

import operator
import os
from typing import Annotated, TypedDict

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt

DB_URI = os.environ["DB_URI"]
THREAD = "order-7f3a"


class State(TypedDict):
    order_id: str
    total_cents: int
    log: Annotated[list[str], operator.add]
    approved: bool


def validate(state: State) -> dict:
    return {"log": [f"validated {state['order_id']}"]}


def charge(state: State) -> dict:
    decision = interrupt(
        {
            "question": "Charge this card?",
            "order_id": state["order_id"],
            "amount_cents": state["total_cents"],
        }
    )
    if not decision.get("approved"):
        return {"log": ["charge rejected by human"], "approved": False}
    return {"log": [f"charged {state['total_cents']} cents"], "approved": True}


def notify(state: State) -> dict:
    return {"log": ["customer notified"]}


def build(checkpointer):
    g = StateGraph(State)
    g.add_node("validate", validate)
    g.add_node("charge", charge)
    g.add_node("notify", notify)
    g.add_edge(START, "validate")
    g.add_edge("validate", "charge")
    g.add_edge("charge", "notify")
    g.add_edge("notify", END)
    return g.compile(checkpointer=checkpointer)


if __name__ == "__main__":
    config = {"configurable": {"thread_id": THREAD}}
    with PostgresSaver.from_conn_string(DB_URI) as cp:
        cp.setup()
        graph = build(cp)
        graph.invoke(
            {"order_id": "A-1042", "total_cents": 49900, "log": []},
            config,
            durability="sync",
        )
        snap = graph.get_state(config)
        print("next nodes:", snap.next)
        print("interrupt payload:", snap.tasks[0].interrupts[0].value)
        print("state log:", snap.values["log"])

Jalankan:

next nodes: ('charge',)
interrupt payload: {'question': 'Charge this card?', 'order_id': 'A-1042', 'amount_cents': 49900}
state log: ['validated A-1042']

Script-nya keluar padahal run-nya belum selesai. snap.next isinya ('charge',), artinya graph nunggu node itu dijalankan lagi, kali ini dengan jawaban.

setup() yang bikin empat tabel: checkpoints, checkpoint_blobs, checkpoint_writes, dan checkpoint_migrations. Panggil sekali aja sebagai langkah migrasi waktu deploy, bukan di tiap request.

Langkah 3: resume dari proses lain

Simpan sebagai resume.py:

import os

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command

from demo import THREAD, build

DB_URI = os.environ["DB_URI"]
config = {"configurable": {"thread_id": THREAD}}

with PostgresSaver.from_conn_string(DB_URI) as cp:
    graph = build(cp)
    before = graph.get_state(config)
    print("state before resume:", before.values, "next:", before.next)
    out = graph.invoke(Command(resume={"approved": True}), config, durability="sync")
    print("final log:", out["log"])
    print("checkpoints stored:", len(list(graph.get_state_history(config))))

Jalankan sebagai proses terpisah, bukan dari dalam demo.py:

state before resume: {'order_id': 'A-1042', 'total_cents': 49900, 'log': ['validated A-1042']} next: ('charge',)
final log: ['validated A-1042', 'charged 49900 cents', 'customer notified']
checkpoints stored: 5

Saya jalankan dua script itu sebagai dua container terpisah yang cuma berbagi instance Postgres. Proses kedua nggak punya memori apa pun soal proses pertama, tapi tetap tahu order id, nominalnya, dan bahwa validasi sudah lewat. Lima baris checkpoint buat run tiga node: satu buat input, lalu satu per super-step.

Langkah 4: taruh di belakang HTTP

Cerita restart cuma berguna kalau API kamu bisa nemuin run yang lagi pause. Dua hal penting: connection pool, bukan satu koneksi, dan setup() waktu boot.

import os

from fastapi import FastAPI
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool

from demo import build

DB_URI = os.environ["DB_URI"]

pool = ConnectionPool(
    conninfo=DB_URI,
    max_size=10,
    kwargs={"autocommit": True, "prepare_threshold": 0, "row_factory": dict_row},
)
pool.wait()

checkpointer = PostgresSaver(pool)
checkpointer.setup()
graph = build(checkpointer)

app = FastAPI()


def cfg(thread_id: str):
    return {"configurable": {"thread_id": thread_id}}


@app.get("/orders/{thread_id}")
def pending(thread_id: str):
    snap = graph.get_state(cfg(thread_id))
    return {"waiting_on": list(snap.next), "state": snap.values}


@app.post("/orders/{thread_id}")
def start(thread_id: str, body: dict):
    graph.invoke(body, cfg(thread_id), durability="sync")
    return pending(thread_id)


@app.post("/orders/{thread_id}/approve")
def approve(thread_id: str, body: dict):
    graph.invoke(Command(resume=body), cfg(thread_id), durability="sync")
    return pending(thread_id)

Route-nya def biasa biar FastAPI jalankan di threadpool, karena itu yang dibutuhin PostgresSaver versi sync. Kalau sisa aplikasi kamu async, pakai AsyncPostgresSaver dan await graph.ainvoke(...). Manggil saver sync dari handler async def bikin event loop ke-blok di setiap tulisan ke database, dan itu baru kelihatan pas traffic ramai.

Jebakan yang bikin duit hilang

Jebakan 1: node jalan ulang dari atas

Resume itu bukan lanjutin di tengah fungsi. Node-nya mulai lagi dari awal, dan interrupt() balikin nilai resume di titik yang tadi berhenti, tapi semua baris di atasnya jalan dua kali. Saya hitung pakai print di atas interrupt: sekali sebelum pause, sekali lagi pas resume.

Jadi ini bug yang tinggal nunggu waktu:

def charge(state: State) -> dict:
    receipt = stripe.PaymentIntent.create(amount=state["total_cents"])
    decision = interrupt({"receipt": receipt.id})

Taruh side effect di bawah interrupt, atau kirim idempotency key dari thread_id plus nama node biar panggilan kedua jadi no-op.

Jebakan 2: input baru bukan resume

Waktu graph lagi pause, invoke() dengan input baru nggak error. Input-nya diterima, digabung ke state, dan node yang interrupt jalan lagi. Saya tes persis itu: nggak ada exception, proses lanjut, dan string yang saya kirim muncul di log akhir.

Jadi cek state dulu sebelum mutusin:

snap = graph.get_state(cfg)
if snap.next:
    graph.invoke(Command(resume=answer), cfg, durability="sync")
else:
    graph.invoke(new_input, cfg, durability="sync")

Tanpa cek itu, user yang klik approve dua kali, atau retry dari queue, diam-diam nimpa payload kedua ke thread yang sama.

Jebakan 3: default-nya async

durability punya tiga nilai: exit, async, sync. Kalau nggak diisi, kamu dapat async, yang nulis checkpoint di background sambil langkah berikutnya jalan. Proses mati di jendela itu, langkah terakhirnya hilang. sync nulis sebelum node berikutnya mulai, dan itu yang kamu mau buat urusan duit, email, dan apa pun yang sudah di-approve manusia. Harganya satu round trip database per langkah, jadi agent 20 langkah bayar 20 kali.

Satu detail dari source: durability diabaikan, plus muncul warning, kalau graph-nya nggak punya checkpointer.

Jebakan 4: satu koneksi bukan pool

PostgresSaver.from_conn_string() buka satu koneksi dan kasih ke saver. Cukup buat script. Di server yang handle banyak run sekaligus, satu koneksi itu jadi bottleneck dan sumber kesalahan antar request. Bikin saver dari ConnectionPool seperti contoh FastAPI di atas.

Edit state sebelum resume

UI approval biasanya butuh lebih dari sekadar ya atau tidak. Misalnya manusia nurunin nominalnya. update_state nulis checkpoint baru, dan node yang di-resume lihat nilai yang sudah diedit:

graph.update_state(cfg, {"total_cents": 19900}, as_node="validate")
graph.invoke(Command(resume={"approved": True}), cfg, durability="sync")

Output dari run itu:

log after edited resume: ['validated A-1042', 'charged 19900 cents', 'customer notified']

as_node penting. Itu kasih tahu LangGraph node mana yang bikin edit, dan dari situ ditentuin apa yang jalan berikutnya. Salah isi, langkah bisa kelewat atau keulang.

Riwayat yang sama juga dipakai buat replay. get_state_history() balikin yang terbaru dulu, dan invoke pakai checkpoint_id lama akan jalanin ulang semua setelahnya. Interrupt ikut kepicu lagi waktu replay, dan itu berguna banget pas lagi debug alur approval.

Storage dan bersih-bersih

Empat tabel, satu baris per checkpoint, isi channel penuh di setiap langkah. Murah buat state kecil, nggak murah buat thread yang bawa riwayat chat panjang atau output tool besar. Hapus thread yang sudah selesai:

checkpointer.delete_thread(thread_id)

Jadwalkan itu, atau bikin job retensi di tabel checkpoint. Buat memori yang harus lintas thread, misalnya preferensi user, pakai Store di samping checkpointer, bukan dipaksa masuk ke graph state.

Ringkasan

  • Compile graph pakai checkpointer Postgres, dan kirim thread_id di setiap panggilan.
  • Panggil setup() sekali waktu deploy.
  • interrupt() pause dan simpan. Command(resume=...) lanjutin, bahkan dari proses lain.
  • Semua baris di atas interrupt() jalan dua kali. Pindahin side effect ke bawah, atau bikin idempotent.
  • Cek get_state().next sebelum invoke pakai input baru.
  • Pakai durability="sync" buat langkah yang nggak boleh hilang.
  • Bikin saver dari ConnectionPool, bukan satu koneksi.

Langkah lanjutan yang worth it: pindahin panggilan resume ke webhook dari UI approval kamu, lalu tambah retry policy di node charge.

Referensi

Butuh Bantuan Implementasi?

Saya membantu tim mendesain dan membangun infrastruktur cloud scalable, pipeline DevOps, dan sistem production-grade.

Konsultasi Gratis