LangGraph State Machine — Bagian yang Tutorial Skip

Tutorial LangGraph selalu menunjukkan workflow sederhana: node A → node B → END. Di production, ini hampir tidak pernah cukup.

Yang tidak ditunjukkan tutorial: bagaimana menangani error di tengah graph, bagaimana rollback ke state sebelumnya, dan bagaimana membuat agent berhenti dan bertanya ke manusia sebelum mengambil keputusan kritis.

Kenapa LangGraph, Bukan LangChain Chain Biasa

LangChain chains bersifat linear: input → transform → output. Untuk workflow sederhana ini cukup. Tapi agentic workflow sering butuh:

Loop — agent perlu mencoba lagi dengan strategi berbeda jika gagal
Branching — keputusan berbeda berdasarkan kondisi yang dinamis
State persistence — ingat apa yang sudah dilakukan di langkah sebelumnya
Human checkpoint — berhenti dan minta konfirmasi sebelum aksi irreversible

LangGraph State Machine — Flow Complaint Handler

Mental Model LangGraph

LangGraph adalah state machine. State adalah "apa yang diketahui agent saat ini". Setiap node mengubah state. Edges menentukan ke mana state itu pergi selanjutnya berdasarkan kondisi.

Fitur	LangChain Chain	LangGraph
Flow pattern	Linear (A→B→C)	Graph (A→B, A→C, loop)
State management	Manual / tidak ada	TypedDict state otomatis
Conditional routing	△ Terbatas	✓ Penuh dengan conditional edges
Checkpoint / resume	✗ Tidak ada	✓ Memory / PostgreSQL saver
Human-in-the-loop	✗ Tidak ada	✓ interrupt() native
Debug visibility	△ Verbose logs	✓ Node-by-node state trace
Learning curve	Lebih mudah	△ Lebih steep

Struktur Dasar: State + Nodes + Edges

Python · LangGraph StateGraph

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

# 1. State — ini yang "diingat" agent sepanjang workflow
class AgentState(TypedDict):
    messages: Annotated[List, operator.add]
    image_analysis: str
    damage_type: str
    sla_policy: str
    ticket_id: str
    needs_more_info: bool  # flag untuk conditional routing

# 2. Nodes — fungsi yang mengubah state
def analyze_image_node(state: AgentState) -> dict:
    analysis = vision_api.analyze(state["image_url"])
    return {
        "image_analysis": analysis.description,
        "needs_more_info": analysis.confidence < 0.7
    }

def query_sla_node(state: AgentState) -> dict:
    return {"sla_policy": rag_retriever.invoke(state["damage_type"])}

def create_ticket_node(state: AgentState) -> dict:
    return {"ticket_id": crm_api.create_ticket(state).id}

# 3. Conditional edge — routing berdasarkan state
def route_after_analysis(state: AgentState) -> str:
    if state["needs_more_info"]:
        return "request_info"  # node yang minta foto ulang ke customer
    return "query_sla"        # lanjut normal

# 4. Build graph
workflow = StateGraph(AgentState)
workflow.add_node("analyze_image", analyze_image_node)
workflow.add_node("query_sla", query_sla_node)
workflow.add_node("create_ticket", create_ticket_node)
workflow.set_entry_point("analyze_image")
workflow.add_conditional_edges(
    "analyze_image", route_after_analysis,
    {"query_sla": "query_sla", "request_info": "request_info"}
)
workflow.add_edge("query_sla", "create_ticket")
workflow.add_edge("create_ticket", END)
graph = workflow.compile()

Checkpoint — State Persistence yang Sering Diabaikan

Tanpa checkpoint, kalau graph gagal di tengah jalan (network error, timeout), seluruh proses harus diulang dari awal. Dengan checkpoint, graph bisa dilanjutkan dari node terakhir yang sukses.

Python · LangGraph Checkpoint

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver

# Development: simpan di memory
checkpointer = MemorySaver()

# Production: simpan di PostgreSQL — persist across restarts
# checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)

graph = workflow.compile(checkpointer=checkpointer)

# thread_id mengidentifikasi "session" agent yang unik
config = {"configurable": {"thread_id": "complaint-TKT-2847"}}
result = graph.invoke(initial_state, config)

# Kalau error di tengah jalan, resume dari state terakhir:
# graph.invoke(None, config)  ← None = gunakan state dari checkpoint

Human-in-the-Loop — Kapan Agent Harus Berhenti

Untuk aksi yang irreversible (hapus data, kirim email blast, approve transaksi besar), agent harus berhenti dan minta konfirmasi manusia terlebih dulu.

Python · Human Interrupt

from langgraph.graph import interrupt

def approve_escalation_node(state: AgentState) -> dict:
    # Graph berhenti di sini dan menunggu input manusia
    approval = interrupt({
        "question": "Tiket ini akan di-escalate ke GM. Konfirmasi?",
        "ticket_id": state["ticket_id"]
    })
    if approval["approved"]:
        return escalate_ticket(state)
    return {"status": "rejected_by_human"}

# Sisi application (Telegram/WhatsApp bot):
# 1. Graph berhenti di interrupt node
# 2. Kirim notifikasi ke manager minta konfirmasi
# 3. Manager reply → resume graph:
# graph.invoke(Command(resume={"approved": True}), config)

Selalu gunakan TypedDict untuk State

TypedDict memberikan type hints yang membantu IDE dan mencegah typo field name saat mengakses state. Definisikan semua field yang mungkin dibutuhkan dari awal — lebih mudah daripada menambah belakangan.

MemorySaver untuk dev, PostgresSaver untuk production

MemorySaver tidak persist saat server restart. Di production, selalu gunakan PostgresSaver atau checkpointer external lainnya agar workflow bisa di-resume setelah crash atau deploy ulang.

thread_id harus unik per "konversasi"

thread_id adalah identifier untuk satu instance workflow. Gunakan ID yang meaningful — misalnya "complaint-{ticket_id}" atau "session-{user_id}-{timestamp}" — agar mudah di-debug di production.

💡 Interrupt vs Conditional Edge

Gunakan interrupt untuk berhenti dan menunggu input manusia secara asynchronous. Gunakan conditional edge untuk routing otomatis berdasarkan state yang sudah ada. Keduanya berbeda tujuan dan tidak bisa saling menggantikan.

⚠️ Jangan Gunakan LangGraph untuk Workflow Sederhana

LangGraph menambah complexity — TypedDict, graph compilation, checkpointer setup. Untuk workflow linear yang tidak butuh loop, branching kompleks, atau state persistence, LangChain chain biasa atau bahkan plain Python function lebih mudah di-maintain.

Kesimpulan

Tiga hal wajib dari hari pertama implementasi LangGraph: TypedDict state yang jelas, checkpoint untuk persistence, dan interrupt node untuk aksi kritis yang butuh approval manusia.

Desain state Anda dulu sebelum desain nodes. State yang jelas membuat seluruh graph lebih mudah di-debug dan di-maintain ketika ada yang tidak berjalan sesuai rencana.

Episode berikutnya: Prompt vs Context vs Harness Engineering — perbedaan yang sering dikacaukan dan mengapa masing-masing penting di tempat yang berbeda dalam system LLM.

#LangGraph #LangChain #AIAgent #Python #AgenticAI #BuildInPublic