5 Mar 2026

State Machine dan Agen: Membangun Workflow Andal dengan LangGraph

Sebagian besar tutorial agen menunjukkan loop sederhana: tanya Claude, parse respons, panggil tool, ulangi. Ini berhasil untuk demo. Dalam produksi, Anda memerlukan determinisme, pemulihan kesalahan, gerbang persetujuan manusia, dan auditabilitas.

LangGraph membawa state machine ke dalam workflow agen. Alih-alih loop ad-hoc yang disatukan oleh pernyataan if, Anda mendapatkan graph eksplisit: node bernama (unit logika), edge bertipe (transisi), dan skema state bersama yang mengalir melalui seluruh eksekusi.

Mengapa State Machine untuk Agen?

Masalah Loop Ad-Hoc

Loop agen tipikal terlihat seperti ini:

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

Ini mudah dibaca untuk dua atau tiga tool. Tambahkan lima tool, jalur kondisional, langkah persetujuan manusia, dan logika retry — Anda mendapat ratusan baris kontrol alur yang kusut.

Masalah yang lebih dalam adalah state implisit. Di tahap mana agen berada? Data apa yang telah dikumpulkannya? Semuanya tinggal di messages — blob tak bertipe yang dibaca dan ditambahkan setiap node, tanpa skema yang dipaksakan.

State Machine sebagai Solusi

State machine membuat yang implisit menjadi eksplisit. Anda mendefinisikan:

Node — unit logika diskrit yang menerima state saat ini, melakukan satu hal, dan mengembalikan pembaruan state.
Edge — transisi antar node, baik tanpa syarat (A → B selalu) atau kondisional.
State — dictionary bertipe yang mengalir melalui seluruh graph. Skema divalidasi di setiap langkah.

Kapan TIDAK Menggunakan LangGraph

Untuk tugas sederhana satu langkah, panggilan API langsung lebih cepat dan jelas. Gunakan LangGraph ketika workflow Anda memiliki:

Beberapa tahap berbeda yang harus dijalankan secara berurutan
Percabangan kondisional berdasarkan hasil antara
Langkah dengan keterlibatan manusia
Logika pemulihan kesalahan atau retry
Persyaratan auditabilitas

Dasar-Dasar LangGraph

Mendefinisikan State

from typing import TypedDict

class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

Node

import anthropic

client = anthropic.Anthropic()

def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Teliti topik ini secara menyeluruh: {state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}

def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Tulis draf berdasarkan catatan ini:\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

Edge dan Routing Kondisional

from typing import Literal

def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"

workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

Membangun dan Menjalankan Graph

from langgraph.graph import StateGraph, END

workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)

graph = workflow.compile()
result = graph.invoke({"query": "Apa itu protokol MCP?"})
print(result["draft"])

Membangun Workflow Peninjauan Dokumen

Merancang State

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

Mengimplementasikan Node

import json

def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "Ekstrak tepat 5 istilah kunci dari dokumen ini. "
                "Kembalikan sebagai array JSON string.\n\n"
                f"Dokumen: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}


def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "Periksa dokumen ini untuk masalah kepatuhan. "
                "Cari: PII (nama, NIK, email, nomor telepon), "
                "tanda rahasia, klaim yang tidak terverifikasi.\n\n"
                "Kembalikan array JSON deskripsi masalah. "
                "Kembalikan array kosong [] jika tidak ada masalah.\n\n"
                f"Dokumen: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}


def human_review_node(state: DocumentState) -> dict:
    print("\n=== TINJAUAN MANUSIA DIPERLUKAN ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "Ditinjau dan disetujui setelah penghapusan PII",
        "is_approved": True
    }


def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"Ringkas dokumen ini dalam tepat dua kalimat:\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

Merakit Graph

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"

workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)

workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check", route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)

graph = workflow.compile()

result = graph.invoke({
    "document": "Pasien Budi Santoso (NIK: 3171234567890001) menderita hipertensi.",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

Human Checkpoint dan Interupsi

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)

thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)

current_state = graph.get_state(thread_config)
print("Masalah:", current_state.values["compliance_issues"])

graph.update_state(
    thread_config,
    {"human_feedback": "Disetujui setelah penghapusan PII manual", "is_approved": True}
)

final_result = graph.invoke(None, config=thread_config)

Persistensi dan Deployment Produksi

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)

import asyncio

async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[
        graph.ainvoke({"document": doc, "key_terms": [], "compliance_issues": [],
                       "summary": "", "human_feedback": "", "is_approved": False})
        for doc in documents
    ])

Pola Umum dan Jebakan

Fan-Out / Fan-In: Jalankan beberapa node secara paralel dengan Send dan gabungkan hasilnya.
Routing kondisional berlebihan: Jika fungsi routing memiliki terlalu banyak cabang, pecah menjadi subgraph.
State membengkak: Simpan referensi (ID database, kunci S3) daripada artefak besar.
Pengujian node secara terisolasi: Karena node adalah fungsi biasa, bisa diuji unit secara langsung.

def test_extract_terms_node():
    state = {
        "document": "Kecerdasan buatan sedang mengubah masyarakat modern dengan cepat.",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)

LangGraph menggantikan loop ad-hoc dengan workflow yang eksplisit, mudah di-debug, dan dapat dilanjutkan. Mulai dengan graph linear (A → B → C → AKHIR). Tambahkan satu edge kondisional saat perlu percabangan. Tambahkan human checkpoint saat perlu persetujuan. Sebagian besar workflow agen produksi hanya membutuhkan ketiga pola ini.