Belajar LangChain - LangGraph Lanjutan

#Pendahuluan

Di episode 13 kalian merakit create_agent dan menyusun logika agen sebagai StateGraph dengan node, edge, conditional edges, sampai interrupt untuk human-in-the-loop; di episode 7 kalian juga menyentuh checkpointing sederhana lewat MemorySaver dan thread_id. Dua fondasi itu menjadi pijakan untuk langkah berikutnya: grafik yang berkelanjutan di produksi. Episode ini menyelami LangGraph lebih dalam — mulai dari typed state dan reducers agar kontrak state eksplisit, checkpointing yang menyimpan snapshot di setiap langkah, time travel untuk meninjau, memutar ulang, dan menulis ulang riwayat, subgraphs untuk menyusun grafik besar dari grafik kecil, lalu persistence PostgreSQL dengan namespace dan TTL untuk mengendalikan siklus hidup penyimpanan.

#Typed State dan Reducers

StateGraph di episode 13 memakai dict biasa. Di produksi, state harus punya kontrak yang jelas — inilah gunanya typed state: mendeklarasikan skema state dengan TypedDict dan menandai tiap channel dengan aturan update lewat reducers.

Typed state dengan reducer

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph.message import add_messages
class BlogState(TypedDict):
    pesan: Annotated[list, add_messages]
    draf: str
    jumlah_update: Annotated[int, add]

Tanpa reducer, perilaku default setiap channel adalah last-write-wins: nilai terbaru yang ditulis sebuah node menimpa nilai lama. Reducer mengubah perilaku itu menjadi merge. add_messages dari langgraph.graph.message menggabungkan daftar pesan tanpa duplikat, operator.add menggabungkan nilai (cocok untuk list dan integer), dan reducer bisa berupa fungsi kustom apa pun yang menerima nilai lama dan nilai baru lalu mengembalikan hasil gabungan:

Reducer kustom

def ambil_terpanjang(lama: str, baru: str) -> str:
    return baru if len(baru) >= len(lama) else lama
 
class DraffState(TypedDict):
    draf: Annotated[str, ambil_terpanjang]

Reducer paling penting di dunia agent adalah add_messages — ia yang menjaga riwayat percakapan terus bertambah alih-alih tertimpa tiap giliran. Pilih reducer sesuai semantik channel: list yang menumpuk pakai operator.add, pesan pakai add_messages, dan skenario khusus tinggal menulis fungsi sendiri.

#Checkpointing: Snapshot di Setiap Langkah

Checkpoint adalah snapshot state thread pada setiap super-step — batas antar langkah eksekusi graph. Checkpointer menyimpan snapshot tersebut dan menjadi bahan bakar memory, human-in-the-loop, time travel, dan fault tolerance. Tanpa checkpointer, state hilang begitu invoke selesai.

Compile graph dengan checkpointer

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
 
def tulis_draf(state: BlogState):
    return {"pesan": [("user", "mulai menulis")], "draf": "Draf pertama.", "jumlah_update": 1}
builder = StateGraph(BlogState)
builder.add_node("tulis", tulis_draf)
builder.add_edge(START, "tulis")
builder.add_edge("tulis", END)
graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "thread-1"}}
graph.invoke({"pesan": [], "draf": "", "jumlah_update": 0}, config=config)

Setiap pemanggilan menyertakan thread_id pada bagian configurable — inilah kunci utama penyimpanan. invoke berikutnya dengan thread_id yang sama melanjutkan dari checkpoint terakhir. Untuk produksi, MemorySaver tidak cukup karena ia menyimpan di RAM dan hilang saat proses restart. Ganti dengan checkpointer persisten: SqliteSaver (file lokal, pip install langgraph-checkpoint-sqlite) untuk pengembangan, dan PostgresSaver untuk produksi — kita bahas di bagian akhir.

#Time Travel: Meninjau, Memutar Ulang, dan Menulis Ulang Riwayat

Karena setiap langkah tersimpan sebagai checkpoint, kalian bisa "berkeliling waktu". get_state mengambil snapshot terbaru, get_state_history mengembalikan seluruh riwayat (terbaru duluan), dan tiap snapshot membawa config berisi checkpoint_id.

Time travel dengan checkpoint_id

snapshot = graph.get_state(config)
print(snapshot.values)
print(snapshot.next)
history = list(graph.get_state_history(config))
titik_input = history[-1]
balik = {"configurable": {"thread_id": "thread-1", "checkpoint_id": titik_input.config["configurable"]["checkpoint_id"]}}
graph.invoke(None, config=balik)
graph.update_state(config, {"draf": "Draf revisi dari masa lalu"})

Menambahkan checkpoint_id pada config membuat invoke memutar ulang eksekusi dari titik itu: node sebelum checkpoint dilewati (hasilnya sudah tersimpan), node setelahnya dieksekusi ulang. update_state menulis nilai baru ke state dan membuat checkpoint baru tanpa mengubah yang lama — inilah cara menulis ulang sejarah atau melakukan fork menuju lintasan alternatif. Nilai update tetap melewati reducer, jadi channel seperti pesan akan menumpuk, bukan menimpa.

#Subgraphs: Grafik di Dalam Grafik

Subgraph adalah compiled graph yang dipasang sebagai node di graph lain. Ini pola komposisi: pecah workflow besar menjadi grafik kecil yang bisa diuji terpisah, lalu rakit menjadi satu.

Subgraph sebagai node

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class ChildState(TypedDict):
    tugas: str
    hasil: str
 
def kerjakan(state: ChildState):
    return {"hasil": "diproses: " + state["tugas"]}
child_graph = StateGraph(ChildState)
child_graph.add_node("kerjakan", kerjakan)
child_graph.add_edge(START, "kerjakan")
child_graph.add_edge("kerjakan", END)
child = child_graph.compile()
 
class ParentState(TypedDict):
    tugas: str
    hasil: str
 
def jalankan_child(state: ParentState):
    return child.invoke({"tugas": state["tugas"]})
parent = StateGraph(ParentState)
parent.add_node("child", jalankan_child)
parent.add_edge(START, "child")
parent.add_edge("child", END)
graph = parent.compile(checkpointer=checkpointer)

State dipindahkan antara induk dan anak dengan mencocokkan nama key: key yang ada di skema anak ikut dikirim, dan key yang ditulis anak lalu juga ada di skema induk dipetakan kembali. Setiap subgraph juga mengelola checkpoint namespace-nya sendiri (checkpoint_ns berisi nama node), jadi perubahan state di dalam subgraph belum tentu langsung terlihat di induk — untuk data yang harus lintas batas graph, pertimbangkan Store. Penting juga: hanya compile graph induk dengan checkpointer; subgraph yang di-compile dengan checkpointer sendiri memecah namespace dan menggelembungkan penyimpanan.

#Persistence Produksi dengan langgraph-checkpoint-postgres

Untuk produksi, memory dan SQLite tidak memadai: butuh checkpointer yang durable, mendukung async, dan dibagikan antar instance. langgraph-checkpoint-postgres menyediakan PostgresSaver dan AsyncPostgresSaver yang menyimpan checkpoint di PostgreSQL. Pasang lewat pip install langgraph-checkpoint-postgres.

Instalasi checkpointer PostgreSQLPostgresSaver dengan from_conn_string

pip install langgraph-checkpoint-postgres

from langgraph.checkpoint.postgres import PostgresSaver
 
DB_URI = "postgresql://langgraph:password@localhost:5432/langgraph"
 
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

setup() membuat tabel checkpoints beserta index-nya — jalankan sekali saat pertama kali memakai database baru. Setiap invoke dengan thread_id kini menulis checkpoint ke Postgres, tahan restart, dan bisa diakses dari instance mana pun. Untuk aplikasi async, pakai AsyncPostgresSaver dari langgraph.checkpoint.postgres.aio. Di beban tinggi, hindari koneksi tunggal; gunakan connection pool:

Connection pool untuk beban produksi

from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver
pool = ConnectionPool(DB_URI, max_size=10, kwargs={"autocommit": True, "row_factory": dict_row})
checkpointer = PostgresSaver(pool)
checkpointer.setup()

Dua pengaturan wajib di atas — autocommit dan row_factory berbentuk dict_row — dibutuhkan agar setup() bisa commit tabel dan agar PostgresSaver mengakses kolom memakai sintaks dict. Pola ini dipakai di deployment produksi karena koneksi dibagikan dan tidak timeout pada workflow yang lama.

#Namespace dan TTL: Isolasi dan Siklus Hidup Data

Setiap checkpoint punya namespace (checkpoint_ns). Nilai "" berarti graph induk; subgraph memakai nama node plus id unik, dan subgraph bertingkat digabung dengan pemisah |. Namespace inilah yang memisahkan checkpoint graph induk dari checkpoint subgraph dalam thread yang sama — dan isolasi ini penting untuk multi-tenant: desain thread_id dengan skema yang jelas (id pengguna, jenis workflow, timestamp) agar thread tidak saling tumpang tindih. PostgresSaver menyimpan thread_id pada kolom berbatas panjang, jadi jaga di bawah 255 karakter, misalnya dengan UUID atau hash. TTL (time-to-live) mengatur berapa lama checkpoint dipertahankan. Di LangGraph Platform, kebijakan retensi dikonfigurasi di langgraph.json:

{
  "graphs": {
    "agent": "./agent.py:graph"
  },
  "checkpointer": {
    "ttl": {
      "strategy": "delete",
      "sweep_interval_minutes": 60,
      "default_ttl": 43200
    }
  }
}

default_ttl dinyatakan dalam menit (43200 = 30 hari) dan dihitung dari aktivitas terakhir — setiap run baru mereset hitung mundur. strategy dengan nilai delete menghapus seluruh thread beserta checkpoint dan writes-nya, sedangkan keep_latest mempertahankan thread dan checkpoint terbaru sambil memangkas riwayat lama; sweep_interval_minutes mengontrol seberapa sering proses background memeriksa thread yang kedaluwarsa. TTL hanya berlaku untuk thread yang dibuat setelah konfigurasi diterapkan, bukan retroaktif. Jika kalian self-host dan memakai PostgresSaver langsung, TTL tidak otomatis: checkpoint menumpuk tanpa batas, jadi siapkan pruning terjadwal — cron job yang menghapus baris checkpoints yang lebih tua dari batas, atau memanggil adelete_thread untuk thread yang sudah usang. Buat kebijakan retensi sejak awal, karena data yang tak pernah dibersihkan memperlambat setiap get_state_history dan membengkakkan biaya storage.

#Penutup

Episode 17 mengubah StateGraph sederhana menjadi mesin yang bisa dioperasikan: typed state dan reducers membuat kontrak state eksplisit, checkpointing menyimpan snapshot tiap langkah, time travel membuka kemampuan meninjau, memutar ulang, dan menulis ulang riwayat, subgraphs memecah grafik besar menjadi komponen teruji, serta persistence PostgreSQL dengan namespace dan TTL memastikan state bertahan lama, terisolasi, dan tidak membengkak tanpa kendali.

Inti yang harus dibawa pulang:

• Typed state dengan TypedDict plus reducers (add_messages, operator.add, atau fungsi kustom) menentukan cara update digabungkan; defaultnya last-write-wins.
• Checkpointer menyimpan checkpoint per super-step dan selalu butuh thread_id; MemorySaver hanya untuk pengembangan, produksi butuh PostgresSaver atau AsyncPostgresSaver.
• Time travel memakai get_state, get_state_history, dan checkpoint_id untuk replay, plus update_state untuk menulis ulang atau memfork riwayat.
• Subgraph berbagi state lewat pencocokan key dan punya checkpoint namespace sendiri; compile hanya induk dengan checkpointer.
• TTL dikonfigurasi lewat checkpointer.ttl di langgraph.json pada platform; di self-host, pruning terjadwal wajib karena checkpoints menumpuk tanpa batas.

Checkpointing dan time travel menambah beban penyimpanan dan latensi yang tidak gratis. Di episode 18 kita menggarap Performance & Cost Optimization: streaming, batching, dan parallelism untuk menekan latensi, plus token usage tracking, prompt caching, model tiering, dan caching layer untuk menekan biaya. Sampai jumpa di sana!