Bikin Agent Stateful pake LangGraph dan Trace pake Langfuse

Kebanyakan demo agent mati begitu user nanya pertanyaan lanjutan. Model lupa apa yang baru saja dibilang, tool call tabrakan satu sama lain, dan kamu nggak tahu step mana yang abisin budget. Solusinya: graph yang pegang state antar step, plus tracing yang nunjukin apa yang sebenernya terjadi.

Artikel ini bikin persis itu: agent LangGraph yang inget percakapan, manggil tool beneran, dan ngirim trace lengkap ke Langfuse. Hasilnya: satu file yang bisa langsung jalan.

Prerequisites

• Python 3.10 ke atas

• OpenAI API key (OPENAI_API_KEY) atau chat model lain yang bisa di-wrap pake LangChain

• Akun Langfuse gratis di cloud.langfuse.com buat dapetin LANGFUSE_PUBLIC_KEY dan LANGFUSE_SECRET_KEY

Install empat package yang beneran kamu butuhin:

pip install "langgraph>=0.2" "langchain-openai>=0.2" "langfuse>=3" python-dotenv

Pas artikel ini ditulis, langgraph ada di versi 0.6.x dan langfuse di 3.x. Pin di atas versi itu dan import path di artikel ini tetep valid.

Sebenernya Agen LangGraph Itu Apa

LangGraph modelin agent sebagai graph berarah. Tiap node itu function Python. State-nya typed dict yang ngalir lewat edge. Conditional edge milih node berikutnya berdasarkan state, dan di situlah keputusan "panggil tool", "tanya user", atau "selesai" dibuat.

Agent berguna yang paling minimal punya empat bagian:

• Schema state yang nyimpen riwayat pesan

• Node model yang manggil LLM

• Node tools yang jalanin fungsi yang dimintain model

• Conditional edge yang loop balik ke model sampai model berhenti manggil tool

Begitu empat ini ada, semua pola lain (human-in-the-loop, parallel tool call, subgraph) cuma variasi dari shape ini.

Step 1: Definisikan State

LangGraph pake type hint Python. Annotated dengan reducer ngomong ke graph gimana cara merge update.

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

Reducer add_messages nambah pesan baru dan overwrite berdasarkan id. Tanpa reducer ini, tiap node bakal replace seluruh list, dan agent kamu kehilangan konteks setelah satu turn. Ini bug paling sering di kode LangGraph pemula.

Step 2: Tool Beneran

Tool itu cuma function dengan docstring dan type hint. Model baca docstring buat tentuin kapan harus manggil.

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Return the current weather for a city. Use this when the user asks about weather."""
    # Stub. Ganti dengan API call beneran.
    lookup = {
        "jakarta": "31C, partly cloudy",
        "singapore": "29C, thunderstorms",
        "tokyo": "22C, clear",
    }
    return lookup.get(city.lower(), f"No data for {city}")

Keep docstring tool tetap pendek dan spesifik. Model yang nggak yakin harus manggil tool atau nggak bakal skip, dan kamu bakal buang sore buat debug prompt padahal masalahnya cuma deskripsi yang kabur.

Step 3: Hubungin Model dan Graph

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [get_weather]
model_with_tools = model.bind_tools(tools)

def call_model(state: State) -> dict:
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    return END

graph = StateGraph(State)
graph.add_node("model", call_model)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "model")
graph.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "model")

agent = graph.compile()

ToolNode adalah prebuilt node yang baca tool_calls dari pesan terakhir, jalanin tool yang tepat, dan return ToolMessage. Kamu bisa tulis ini manual, tapi versi prebuilt ngurusin parallel tool call dan error wrapping otomatis.

Fungsi should_continue adalah conditional edge. Dia return nama node berikutnya. Kalau model baru aja bikin tool call, kita ke tools. Kalau nggak, kita ke END dan graph return jawaban akhir.

Step 4: Jalanin

from langchain_core.messages import HumanMessage

result = agent.invoke({
    "messages": [HumanMessage(content="What is the weather in Jakarta and Singapore?")]
})

for msg in result["messages"]:
    msg.pretty_print()

Karena model bisa manggil tool secara paralel, kamu harusnya liat satu AIMessage dengan dua tool_calls, dua hasil ToolMessage, dan AIMessage akhir yang gabungin jawaban. Total step graph: model, tools, model. Tiga kunjungan node, satu round trip.

Step 5: Trace pake Langfuse

Di sinilah kebanyakan tim berhenti. Mereka deploy agent, nemu bug di production, dan nggak punya cara liat step mana yang rusak. Langfuse adalah tool tracing open source yang hook ke LangGraph cuma dengan satu baris.

Set tiga env var (taruh di .env):

LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_HOST=https://cloud.langfuse.com

Terus aktifin callback-nya:

from langfuse import get_client
from langfuse.langchain import CallbackHandler

langfuse = get_client()
handler = CallbackHandler()

result = agent.invoke(
    {"messages": [HumanMessage(content="What is the weather in Jakarta?")]},
    config={"callbacks": [handler]},
)

Buka dashboard Langfuse. Kamu bakal liat trace dengan tiga span: satu buat model call, satu buat eksekusi tool, satu buat model call terakhir. Tiap span nunjukin prompt, completion, token usage, latency, dan cost.

Satu trace ini ngubah "agent ngasih jawaban aneh" jadi "model call kedua halusinasi suhu." Data sama, tapi yang ini bisa kamu tindak lanjuti.

Step 6: Tambahin Evaluation Smoke Test

Tracing itu reaktif. Kamu juga butuh cek cepat bahwa graph masih behave bener setelah kamu ubah prompt atau ganti model. Langfuse punya setup dataset-and-runner buat ini.

from langfuse import get_client

langfuse = get_client()
dataset = langfuse.create_dataset(name="weather-agent-smoke")

dataset.create_item(
    input={"messages": [HumanMessage(content="Weather in Tokyo?")]},
    expected_output="Tokyo",
)

def run_agent(item):
    return agent.invoke(item.input, config={"callbacks": [handler]})

result = langfuse.run_evaluation(
    dataset_name="weather-agent-smoke",
    task=run_agent,
    evaluators=[],  # tambahin custom evaluator buat cek lebih ketat
)

Pasang ini di CI dan kamu nangkep regresi prompt sebelum nyampe ke user. Argumen evaluators=[] adalah list callable; isi dengan fungsi apa pun yang nerima tuple (input, output, expected_output) dan return skor.

Kapan Pake LangGraph vs Opsi Lain

LangGraph pilihan yang pas kalau kamu butuh kontrol eksplisit atas alur: loop, branch, step approval manusia, persistence. State object adalah source of truth, dan kamu bisa inspect di tengah eksekusi pake checkpointer.

Skip buat tool use single-turn yang simpel. create_tool_calling_agent dari LangChain lebih pendek dan jalanin model call yang sama tanpa boilerplate graph. Ke LangGraph pas demo-nya udah nggak muat di satu turn.

Buat sistem multi-agent di mana agent nge-handoff satu sama lain, pola subgraph LangGraph lebih bersih dibanding setup role-based CrewAI. Harganya lebih banyak kode; untungnya tiap handoff keliatan di trace.

Jebakan Umum

Lupa reducer. Kalau kamu deklarasiin messages: list[BaseMessage] tanpa Annotated[..., add_messages], tiap node replace list-nya. Graph tetep jalan. Agent-nya cuma lupa semua hal setelah step pertama.

Return shape yang salah dari node. Node harus return dict yang key-nya field di state. Return list polos atau satu pesan works buat beberapa node tapi silently break node lain. Selalu return {"messages": [...]}.

Cuma trace call pertama. Kalau kamu build agent sekali dan reuse lintas request, daftarin callback di dalam request handler, bukan di module load. Callback level module bisa bocor state antar user.

Campur async dan sync node. Kalau kamu manggil ainvoke di graph, semua fungsi node harus async. call_model sync di graph async bakal ngeblok event loop. Pilih satu style dan konsisten.

Langkah Selanjutnya

Empat arah yang worth it, urut dari dampak terbesar:

• Tambahin MemorySaver checkpointer biar agent inget lintas request dalam session yang sama.

• Ganti ChatOpenAI dengan model lokal lewat Ollama buat motong cost di query simpel.

• Sambungin trace Langfuse ke alerting stack kamu pas token cost per trace lewat threshold.

• Ganti stub tool dengan API beneran dan tambahin evaluator yang cek shape input tool.

  Bagikan artikel ini
  X

LinkedIn

Facebook

WhatsApp Salin link (function(){const lang = "id";

  const shareLabel = lang === 'en' ? 'Copied!' : 'Tersalin!';
  function copyArticleLink() {
    const btn = document.getElementById('copy-link-btn');
    const text = document.getElementById('copy-link-text');
    navigator.clipboard.writeText(window.location.href).then(() => {
      const orig = text.textContent;
      text.textContent = shareLabel;
      btn.classList.add('copied');
      setTimeout(() => {
        text.textContent = orig;
        btn.classList.remove('copied');
      }, 2000);
    });
  }
})();    

Butuh Bantuan Implementasi?

Saya membantu tim mendesain dan membangun infrastruktur cloud scalable, pipeline DevOps, dan sistem production-grade.

   Konsultasi Gratis