Memahami State Management untuk AI Agent di LangChain

Pendahuluan

Bayangkan kamu sedang membangun asisten virtual untuk aplikasi e-commerce seperti Tokopedia. Pengguna bertanya: “Apa rekomendasi laptop gaming di bawah 10 juta?” Agentmu menjawab dengan baik. Lalu pengguna bertanya lagi: “Yang ada di atas, bisa dicicil tidak?” — dan tiba-tiba agentmu bingung, karena ia tidak ingat percakapan sebelumnya.

Inilah masalah klasik yang diselesaikan oleh state management dalam AI Agent.

State management adalah kemampuan agent untuk menyimpan, mengakses, dan memperbarui konteks sepanjang sesi percakapan. Tanpa ini, setiap interaksi terasa seperti awal yang baru — agent tidak tahu apa yang sudah dibahas sebelumnya.

LangChain, sebagai framework populer untuk membangun AI Agent, menyediakan berbagai mekanisme state management yang powerful. Artikel ini akan membahas konsep dasarnya, pola yang umum digunakan, dan bagaimana mengimplementasikannya secara praktis. Jika kamu belum familiar dengan cara memanggil API LLM, disarankan membaca Cara Memanggil API LLM dengan Python: Panduan Lengkap Pemula terlebih dahulu.

Konsep Dasar

Apa Itu State dalam Konteks AI Agent?

State adalah kumpulan data yang mewakili “ingatan” agent pada suatu titik waktu. State bisa mencakup:

• Riwayat percakapan — pesan apa saja yang sudah dikirim dan diterima
• Variabel konteks — informasi yang dikumpulkan selama sesi (nama user, preferensi, hasil tool)
• Status tugas — apakah agent sedang mengerjakan sesuatu, sudah selesai, atau menemui hambatan

State Management di LangGraph

LangChain modern menggunakan LangGraph sebagai engine untuk membangun stateful agent. LangGraph merepresentasikan alur kerja agent sebagai graph — node adalah langkah pemrosesan, dan edge adalah aliran data antar langkah.

Setiap node membaca state saat ini, melakukan sesuatu, lalu mengembalikan patch (perubahan) ke state. LangGraph akan menggabungkan perubahan tersebut secara otomatis.

Struktur state didefinisikan menggunakan TypedDict atau dataclass:

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    # Riwayat pesan — operator.add berarti "tambahkan ke list"
    messages: Annotated[Sequence[BaseMessage], operator.add]
    # Konteks tambahan
    user_name: str
    current_task: str
    iteration_count: int

Tipe-Tipe State Reducer

LangGraph menggunakan konsep reducer untuk menentukan bagaimana state diperbarui ketika node mengembalikan nilai baru:

Contoh Kode

Setup Dasar

Pertama, install dependensi yang diperlukan:

pip install langgraph langchain-openai langchain-core

Contoh 1: Agent Sederhana dengan State Management

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import operator
import os

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# 1. Definisikan struktur state
class ChatState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    total_tokens_used: int

# 2. Inisialisasi LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 3. Node: fungsi yang memproses state
def chat_node(state: ChatState) -> dict:
    """Node utama yang menjalankan LLM dan memperbarui state."""
    response = llm.invoke(state["messages"])
    
    # Estimasi token (contoh sederhana)
    token_estimate = len(str(state["messages"])) // 4
    
    return {
        "messages": [response],  # Ditambahkan ke list berkat operator.add
        "total_tokens_used": state["total_tokens_used"] + token_estimate
    }

def should_continue(state: ChatState) -> str:
    """Menentukan apakah percakapan berlanjut atau selesai."""
    last_message = state["messages"][-1]
    # Jika user mengetik 'keluar', akhiri sesi
    if hasattr(last_message, 'content') and 'keluar' in last_message.content.lower():
        return "end"
    return "continue"

# 4. Bangun graph
builder = StateGraph(ChatState)
builder.add_node("chat", chat_node)
builder.set_entry_point("chat")
builder.add_conditional_edges(
    "chat",
    should_continue,
    {"continue": END, "end": END}
)

graph = builder.compile()

# 5. Jalankan percakapan
def run_conversation():
    state = {
        "messages": [],
        "total_tokens_used": 0
    }
    
    print("Asisten AI siap! Ketik 'keluar' untuk mengakhiri.\n")
    
    while True:
        user_input = input("Kamu: ")
        if user_input.lower() == 'keluar':
            break
        
        # Tambahkan pesan user ke state
        state["messages"] = list(state["messages"]) + [HumanMessage(content=user_input)]
        
        # Jalankan graph
        result = graph.invoke(state)
        state = result
        
        # Tampilkan respons terakhir
        last_message = state["messages"][-1]
        print(f"Asisten: {last_message.content}")
        print(f"(Total token: ~{state['total_tokens_used']})\n")

run_conversation()

Contoh 2: State Management dengan Memory Jangka Panjang

Untuk agent yang lebih canggih — misalnya layanan customer support seperti yang digunakan aplikasi fintech — kita perlu menyimpan state antar sesi berbeda. LangGraph menyediakan checkpointer untuk ini:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, Sequence
import operator

class SupportState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    customer_id: str
    issue_resolved: bool
    escalation_needed: bool

llm = ChatOpenAI(model="gpt-4o-mini")

SYSTEM_PROMPT = """Kamu adalah agen customer support yang ramah dan profesional.
Selalu ingat konteks percakapan sebelumnya dan bantu pelanggan menyelesaikan masalah mereka."""

def support_agent(state: SupportState) -> dict:
    from langchain_core.messages import SystemMessage
    
    # Tambahkan system prompt jika ini pesan pertama
    messages_with_system = [SystemMessage(content=SYSTEM_PROMPT)] + list(state["messages"])
    
    response = llm.invoke(messages_with_system)
    
    # Deteksi apakah perlu eskalasi
    needs_escalation = any(
        keyword in response.content.lower() 
        for keyword in ["manajer", "supervisor", "tidak bisa membantu"]
    )
    
    return {
        "messages": [response],
        "escalation_needed": needs_escalation,
        "issue_resolved": "terima kasih" in response.content.lower()
    }

# Inisialisasi checkpointer untuk persistensi state
memory = MemorySaver()

builder = StateGraph(SupportState)
builder.add_node("support", support_agent)
builder.set_entry_point("support")
builder.add_edge("support", END)

# Kompilasi dengan checkpointer
graph = builder.compile(checkpointer=memory)

# Setiap thread_id merepresentasikan satu sesi pelanggan
def chat_with_customer(customer_id: str, message: str, thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    
    result = graph.invoke(
        {
            "messages": [HumanMessage(content=message)],
            "customer_id": customer_id,
            "issue_resolved": False,
            "escalation_needed": False
        },
        config=config
    )
    
    last_msg = result["messages"][-1]
    print(f"Agent: {last_msg.content}")
    
    if result["escalation_needed"]:
        print("[SISTEM] Percakapan ini memerlukan eskalasi ke tim manusia.")
    
    return result

# Simulasi percakapan multi-turn
session_id = "session-customer-001"
customer = "CUST-12345"

print("=== Sesi 1 ===")
chat_with_customer(customer, "Halo, pesanan saya belum sampai sudah 5 hari", session_id)

print("\n=== Sesi 2 (konteks tersimpan) ===")
chat_with_customer(customer, "Nomor pesanannya adalah ORD-789012", session_id)

Contoh 3: Custom State Reducer

Kadang kita butuh logika yang lebih spesifik. Misalnya, menyimpan hanya N pesan terakhir untuk menghemat token:

from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage

def keep_last_n_messages(n: int = 10):
    """Factory function untuk membuat reducer yang menyimpan N pesan terakhir."""
    def reducer(existing: list, new: list) -> list:
        combined = list(existing) + list(new)
        # Selalu pertahankan system message jika ada
        system_msgs = [m for m in combined if hasattr(m, 'type') and m.type == 'system']
        non_system = [m for m in combined if not (hasattr(m, 'type') and m.type == 'system')]
        
        # Ambil N pesan terakhir dari non-system
        trimmed = non_system[-n:] if len(non_system) > n else non_system
        return system_msgs + trimmed
    
    return reducer

class TrimmedChatState(TypedDict):
    # Gunakan Annotated dengan custom reducer
    messages: Annotated[list[BaseMessage], keep_last_n_messages(10)]
    context_summary: str  # Ringkasan percakapan yang di-trim

# Penggunaan state ini sama seperti sebelumnya
# Tapi LangGraph akan otomatis memangkas pesan lama

Pola ini sangat berguna jika kamu mengembangkan aplikasi berbiaya rendah, karena sejalan dengan prinsip zero-shot prompting yang dibahas di Menguasai Zero-shot dan Few-shot Prompting untuk LLM.

Troubleshooting: Error yang Sering Muncul

InvalidUpdateError: Expected dict, got <class 'str'>

Penyebab: Node mengembalikan tipe data yang salah. LangGraph mengharapkan setiap node mengembalikan dict yang berisi key sesuai field di state.

Solusi:

# SALAH - mengembalikan string langsung
def bad_node(state: AgentState) -> str:
    return "Ini respons saya"

# BENAR - mengembalikan dict dengan key yang sesuai
def good_node(state: AgentState) -> dict:
    return {"messages": [AIMessage(content="Ini respons saya")]}

KeyError saat Mengakses State yang Belum Diinisialisasi

Penyebab: Field di TypedDict tidak diberi nilai awal saat pertama kali memanggil graph.invoke().

Solusi:

# Selalu inisialisasi SEMUA field saat invoke pertama
result = graph.invoke({
    "messages": [],           # List kosong, bukan None
    "customer_id": "",        # String kosong
    "issue_resolved": False,  # Boolean default
    "escalation_needed": False
})

# Atau gunakan nilai default di TypedDict (Python 3.11+)
from typing import TypedDict
class SafeState(TypedDict, total=False):
    messages: list
    customer_id: str

State Tidak Tersimpan Antar Invokasi (Checkpointer Tidak Berfungsi)

Penyebab: thread_id dalam config tidak konsisten antara panggilan, atau graph dikompilasi ulang tanpa checkpointer yang sama.

Solusi:

# Pastikan config menggunakan thread_id yang SAMA
config = {"configurable": {"thread_id": "sesi-user-001"}}

# SALAH - thread_id berbeda setiap panggilan
config1 = {"configurable": {"thread_id": f"sesi-{random.random()}"}}

# BENAR - gunakan identifier yang stabil
user_id = "usr_12345"
config = {"configurable": {"thread_id": f"chat-{user_id}"}}

result1 = graph.invoke(input1, config=config)  # State tersimpan
result2 = graph.invoke(input2, config=config)  # State dilanjutkan

RecursionError atau Loop Tak Terbatas

Penyebab: Kondisi conditional_edges tidak pernah mengarah ke END, menyebabkan graph berputar terus.

Solusi:

from langgraph.graph import END

def router(state: AgentState) -> str:
    # Selalu ada kondisi yang menuju END
    if state["iteration_count"] >= 10:
        return "end"  # Safeguard wajib!
    
    last_msg = state["messages"][-1]
    if "selesai" in last_msg.content.lower():
        return "end"
    
    return "continue"

builder.add_conditional_edges(
    "agent_node",
    router,
    {
        "continue": "agent_node",  # Loop kembali
        "end": END                 # Terminasi
    }
)

Pertanyaan yang Sering Diajukan

Apa perbedaan antara MemorySaver dan database eksternal untuk state?

MemorySaver menyimpan state di memori RAM — artinya data hilang saat program berhenti. Ini cocok untuk pengembangan dan testing. Untuk produksi, gunakan checkpointer berbasis database seperti PostgresSaver atau SqliteSaver dari paket langgraph-checkpoint-postgres agar state persisten antar restart server.

Bagaimana cara membatasi panjang riwayat pesan agar tidak overload token?

Gunakan custom reducer seperti contoh keep_last_n_messages di atas, atau gunakan fungsi trim_messages dari langchain_core.messages. Strategi lain adalah membuat node “summarizer” yang meringkas pesan lama menjadi satu pesan ringkasan sebelum context window penuh.

Apakah state bisa dibagi antar beberapa agent dalam satu graph?

Ya! Ini adalah kekuatan utama LangGraph. Kamu bisa mendefinisikan satu AgentState yang diakses oleh semua node. Setiap node hanya membaca field yang relevan dan hanya memperbarui field yang menjadi tanggung jawabnya. Ini memungkinkan arsitektur multi-agent yang terkoordinasi dengan baik.

Bagaimana cara men-debug state yang berubah di tengah eksekusi graph?

Gunakan callback on_chain_end atau aktifkan mode debug saat invoke:

# Mode debug — cetak setiap perubahan state
result = graph.invoke(
    initial_state,
    config={"recursion_limit": 10},
    debug=True  # Tampilkan state di setiap langkah
)

Mengapa state saya tidak ter-update meski node sudah mengembalikan nilai baru?

Pastikan tipe reducer sesuai. Jika field menggunakan operator.add (untuk list), nilai yang dikembalikan harus berupa list baru, bukan menimpa seluruh field. Jika reducer-nya default (overwrite), pastikan key yang dikembalikan persis sama dengan nama field di TypedDict.

Kesimpulan

State management adalah fondasi dari AI Agent yang benar-benar berguna. Tanpa kemampuan “mengingat” konteks, agent hanya akan terasa seperti chatbot biasa yang pelupa.

Dengan LangGraph, kita mendapatkan:

• Struktur state yang jelas menggunakan TypedDict
• Reducer yang fleksibel untuk berbagai kebutuhan pembaruan data
• Persistensi lintas sesi melalui checkpointer
• Kontrol alur dengan conditional edges yang aman

Kunci suksesnya adalah merancang struktur state dengan matang di awal — pikirkan data apa saja yang perlu “diingat” agent sepanjang siklus hidupnya, lalu pilih reducer yang tepat untuk setiap field.

Selamat bereksperimen dengan state management di LangChain! Dengan fondasi ini, kamu sudah siap membangun agent yang lebih cerdas dan responsif. Jangan ragu untuk terus eksplorasi artikel lainnya di KamusNgoding — masih banyak topik seru yang menunggumu!