Belajar LangChain - Agents & LangGraph Dasar

#Pendahuluan

Di episode 12 kalian memoles pipeline RAG dengan multi-query, hybrid retrieval, dan reranking — menghasilkan konteks yang jauh lebih berkualitas. Tapi ada batas yang tak bisa diatasi chain statis: chain tahu persis urutan langkahnya, padahal pertanyaan tidak selalu butuh urutan yang sama. Episode 13 menjawabnya dengan agent — program yang memutuskan sendiri langkah selanjutnya berdasarkan state — lalu menyusunnya secara eksplisit dengan LangGraph. Alur episode ini: konsep agent loop, membangun agent dengan create_agent, mengamatinya dengan astream_events, membongkar StateGraph (node, edge, conditional edge), dan menutup dengan interrupt/resume untuk human-in-the-loop.

#Dari Chain ke Agent: Menambah Kemampuan Bertindak

Chain yang kita bangun selama ini bersifat deterministik: prompt masuk, model memanggil, parser mengubah output. Agent menambahkan satu mesin baru yang disebut agent loop:

1. Model menerima riwayat pesan dan daftar tool yang tersedia.
2. Model memutuskan: jawab langsung, atau panggil satu atau lebih tool.
3. Jika memanggil tool, eksekusi tool dilakukan, hasilnya dikirim kembali ke model sebagai pesan baru.
4. Langkah 2 dan 3 diulang sampai model menjawab final atau mencapai batas (recursion limit).

Kunci konseptualnya: model punya kendali jalur — bukan kalian yang menetapkan "selalu retrieve dulu". Agent ini masih satu model, hanya saja kini dilengkapi kemampuan berpikir-panggil-ulang. Definisi tool-nya sama seperti episode 8 dengan dekorator @tool:

Definisi tool untuk agent

from langchain_core.tools import tool
 
@tool
def cek_cuaca(kota: str) -> str:
    "Mengembalikan suhu kota yang diminta."
    return f"28 derajat Celsius di {kota}"
 
@tool
def cari_berita(kueri: str) -> str:
    "Mencari berita terbaru berdasarkan kueri."
    return "Artikel demo: inflasi turun di kuartal kedua"

#Membangun Agent dengan create_agent

Daripada menyusun loop dengan tangan, LangChain v1 menyediakan create_agent — constructor prebuilt yang menerima model, daftar tool, dan prompt, lalu mengembalikan objek yang siap di-invoke:

Membuat agent tool-calling

from langchain.agents import create_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Kamu asisten yang ringkas. Gunakan tool hanya bila perlu."),
    ("placeholder", "{messages}"),
])
 
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    tools=[cek_cuaca, cari_berita],
    prompt=prompt,
)
result = agent.invoke({"messages": [("user", "Cuaca di Jakarta?")]})
print(result["messages"][-1].content)

Perhatikan input-nya: dict dengan key messages berisi daftar pesan — persis seperti ChatPromptTemplate dari episode 4. Outputnya juga berupa messages, dan pesan terakhir memuat jawaban final. Karena create_agent mengembalikan objek bertipe Runnable, semua metode LCEL dari episode 5 tetap berlaku: invoke, ainvoke, stream, hingga batch.

#Mengamati Agent Loop dengan astream_events

Saat agent sedang berpikir, memilih tool, dan memproses hasil, kalian tidak akan tahu apa yang terjadi jika hanya memakai invoke. Di produksi (dan saat debugging), gunakan astream_events — aliran event yang mengungkap setiap tahap: mulai dari eksekusi tool hingga token jawaban:

Streaming event agent

async for event in agent.astream_events(
    {"messages": [("user", "Cuaca di Jakarta dan berita ekonomi?")]},
    version="v1",
):
    kind = event["event"]
    if kind == "on_tool_start":
        print("TOOL:", event["name"], event["data"].get("input"))
    elif kind == "on_chat_model_stream":
        chunk = event["data"]["chunk"]
        if chunk.content:
            print(chunk.content, end="")

Filter pada event di atas memisahkan dua hal yang berbeda: on_tool_start memberi tahu tool mana yang dipanggil dan dengan input apa, sedangkan on_chat_model_stream memberi token-token jawaban secara real-time. Ini pola streaming yang sama seperti episode 5, hanya diterapkan pada agent — dan kombinasi keduanya adalah bahan baku UI "lihat agen berpikir".

#StateGraph Dasar: Node, Edge, START, dan END

Agent loop yang "digulung" oleh create_agent sebenarnya bisa ditulis ulang secara eksplisit sebagai graph: sekumpulan node (fungsi) yang saling terhubung oleh edge. Di LangGraph, graph itu bernama StateGraph, dan state-nya dideskripsikan dengan TypedDict. Contoh minimal dua node:

StateGraph minimal dengan dua node

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
 
class AgentState(TypedDict):
    messages: list
 
def node_a(state: AgentState):
    return {"messages": [("ai", "dari node A")]}
 
def node_b(state: AgentState):
    return {"messages": [("ai", "dari node B")]}
 
graph = StateGraph(AgentState)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_edge(START, "a")
graph.add_edge("a", "b")
graph.add_edge("b", END)
 
app = graph.compile()
result = app.invoke({"messages": []})

Setiap node adalah fungsi yang menerima state dan mengembalikan update parsial — LangGraph menggabungkannya ke state utama. START dan END adalah node khusus untuk masuk dan keluar graph. compile() mengubah graph menjadi executable, dan hasil invoke berisi state lengkap setelah semua node selesai. Inilah inti yang membedakan LangGraph dari chain: state eksplisit yang mengalir antar node, bukan pipe diam-diam.

#Conditional Edges: Memilih Jalan Berdasarkan State

Graph statis sudah berguna, tapi kekuatan LangGraph ada di conditional edges — edge yang memilih node tujuan berdasarkan isi state. Ini yang membuat routing dinamis seperti "jika hasil pencarian kosong, generate langsung; jika tidak, rangkum" menjadi mungkin:

Conditional edge dengan fungsi router

def router(state: AgentState):
    if not state.get("documents"):
        return "generate"
    return "summarize"
 
graph = StateGraph(AgentState)
graph.add_node("retrieve", node_retrieve)
graph.add_node("summarize", node_summarize)
graph.add_node("generate", node_generate)
graph.add_edge(START, "retrieve")
graph.add_conditional_edges(
    "retrieve",
    router,
    {"generate": "generate", "summarize": "summarize"},
)
graph.add_edge("summarize", "generate")
graph.add_edge("generate", END)

add_conditional_edges menerima tiga argumen: node asal, fungsi yang mengembalikan nama tujuan, dan pemetaan nama ke node. Fungsi router membaca state dan mengembalikan kunci tujuan — LangGraph lalu menelusuri pemetaan untuk menemukan node yang tepat. Dengan pola ini, kalian bisa meniru persis logika agent loop: node tools mengeksekusi panggilan tool, lalu conditional edge memutuskan apakah kembali ke model atau menuju node akhir.

#Interrupt & Resume: Human-in-the-Loop

Ada kalanya agent tidak boleh berjalan tanpa pengawasan: konfirmasi pembayaran, menjalankan perintah destruktif, atau menyetujui tool yang sensitif. LangGraph menyediakan interrupt — node yang menjeda eksekusi di tengah jalan dan menunggu keputusan manusia. Syaratnya: graph harus punya checkpointer agar state tersimpan saat dijeda.

Interrupt untuk persetujuan manusia

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
 
def execute_sensitive(state: AgentState):
    action = interrupt({"action": "hapus-file", "target": "/tmp/demo"})
    return {"messages": [("ai", f"aksi disetujui: {action}")]}
 
graph.add_node("execute", execute_sensitive)
graph.add_edge("execute", END)
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
 
config = {"configurable": {"thread_id": "thread-1"}}
app.invoke({"messages": []}, config=config)

Saat mencapai interrupt, eksekusi berhenti dan statusnya tersimpan di bawah thread_id. Untuk melanjutkan, kirim Command dengan nilai resume:

Melanjutkan setelah interrupt

app.invoke(Command(resume="setuju"), config=config)

Nilai "setuju" di atas menjadi nilai balik dari interrupt di dalam node execute. thread_id menghubungkan semua pemanggilan pada percakapan yang sama — konsep yang sudah kalian kenali dari checkpoints episode 7, kini dipakai untuk human-in-the-loop yang sesungguhnya.

#Penutup

Episode ini menandai lompatan konseptual dari pipeline statis ke sistem yang menentukan jalannya sendiri: memahami agent loop sebagai siklus panggil tool dan umpan balik, membangun agent dengan create_agent, memantau prosesnya dengan astream_events, menyusun logika secara eksplisit dengan StateGraph beserta conditional edges, dan menjedanya dengan interrupt/Command untuk human-in-the-loop. Agent kini bukan lagi sekadar "chain dengan tool", melainkan graph stateful yang bisa disetir, dijeda, dan dilanjutkan.

Inti yang harus dibawa pulang:

• Agent loop = model memutuskan, tool dieksekusi, hasil dikirim balik, berulang sampai jawaban final.
• create_agent menyatukan prompt, model, dan tools menjadi satu objek Runnable yang mendukung semua metode LCEL.
• astream_events membuka setiap tahap agent: panggilan tool dan token jawaban.
• StateGraph menyusun node dan edge; conditional edges membuat routing bergantung pada state.
• Interrupt menjeda graph di titik tertentu dan dilanjutkan dengan Command; wajib ada checkpointer.

Episode 13 membangun fondasi graph satu agen. Di episode 14 kita menaikkan levelnya: Deep Agents & Multi-Agent Patterns — subagent yang dibentuk otomatis, subagent asinkron, tool multi-modal untuk PDF/audio/video, dan backend state seperti StateBackend serta StoreBackend. Sampai jumpa di sana!