LangGraph ile yapay zeka ajanı kurmak: durum, düğüm ve döngü

LangGraph, durumu olan, uzun süre çalışabilen yapay zeka ajanlarını ve iş akışlarını kurmak için kullanılan düşük seviyeli bir orkestrasyon çerçevesi. LangChain Inc. geliştiriyor, MIT lisanslı ve LangChain olmadan da kullanılabiliyor. Dokümanın kendi tarifiyle istemleri ya da ajan mimarisini sizin yerinize soyutlamıyor. Bu hem gücü hem de zorluğu: kontrol sizde ama tasarım da sizde.

Bu yazıda LangGraph'ın temel modelini, çalışan küçük bir ajanın iskeletini ve üretime çıkarken kritik olan üç konuyu anlatıyorum: durumun kalıcı saklanması, insan onayı ve hata yönetimi. Kod örnekleri LangChain'in resmi dokümanındaki hızlı başlangıç yapısına dayanıyor. Kütüphane hızlı değişiyor, bu yüzden yazarken kullandığınız sürümün dokümanını açık tutun.

Ajan ile iş akışı arasındaki fark

LangGraph dokümanı bir ayrım yapıyor ve bu ayrım doğru araç seçimini belirliyor. İş akışları önceden belirlenmiş kod yollarını izler ve belli bir sırada çalışmak üzere tasarlanır. Ajanlar ise kendi süreçlerini ve araç kullanımlarını kendileri belirler.

Doküman şu desenleri sayıyor: istem zincirleme (her çağrı bir öncekinin çıktısını işler), paralelleştirme, yönlendirme (girdiyi sınıflayıp uzman bir akışa göndermek), orkestratör ve işçi, değerlendirici ve iyileştirici, ve ajan (bir modelin geri bildirim döngüsünde araç çağırması). Önerisi net: adımlar biliniyorsa ve sonuç tahmin edilebilirse iş akışıyla başlayın, çünkü kontrolü ve hata ayıklaması daha kolay. Problem açık uçluysa ve modelin hangi aracı kullanacağını kendisinin seçmesi gerekiyorsa ajana geçin.

Benim bu konudaki yaklaşımım da bu: müşteri sınıflama, belge okuma ya da rapor hazırlama gibi adımları belli işlerde önce düz bir iş akışı kurun. Serbest dolaşan bir ajanı yalnızca iş akışı yetersiz kaldığında devreye alın. Çoğu iş ikinciye ihtiyaç duymuyor.

Üç temel parça

LangGraph'ta bir graf üç şeyden oluşuyor.

Durum (State): düğümlerin okuyup yazdığı paylaşılan veri.
Düğümler (Nodes): durumu alıp iş yapan ve durum güncellemesi döndüren fonksiyonlar.
Kenarlar (Edges): sıradaki düğümü seçen sabit bağlantılar ya da koşullu fonksiyonlar.

Grafı StateGraph ile kurarsınız ve kullanmadan önce .compile() çağırmanız zorunlu. Derleme yapısal kontrolleri çalıştırıyor ve checkpointer gibi seçenekleri bağlıyor.

Durum şeması varsayılan olarak TypedDict. Varsayılan değer gerekiyorsa dataclass, özyinelemeli doğrulama için Pydantic BaseModel kullanılabiliyor (daha yavaş). Her anahtarın kendi indirgeyicisi (reducer) var. İndirgeyici yoksa yeni değer eskisinin yerini alır. operator.add listeleri biriktirir. add_messages yeni mesajları ekler, aynı kimlikli mesajı değiştirir. Hazır MessagesState bu mantıkla bir messages anahtarı taşıyor.

Bir düğümden birden çok kenar çıkarsa, sonraki adımda bunlar paralel çalışıyor. Send ile çalışma anında kaç dala ayrılacağınızı belirleyebiliyorsunuz (map-reduce için). Command ise durum güncellemesini ve sonraki düğümü (goto) tek dönüşte birleştiriyor, kesinti sonrası devam için resume de taşıyor.

Çalışan küçük bir ajan

Aşağıdaki örnek, dokümandaki hızlı başlangıcın yapısını izleyen bir stok sorgulama ajanı. Veri sahte, model adını kendi sağlayıcınıza göre değiştirin:

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langchain.chat_models import init_chat_model
from langchain.messages import AnyMessage, SystemMessage, ToolMessage, HumanMessage
from langchain.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

STOK = {"A100": 12, "B200": 0}

@tool
def stok_sorgula(urun_kodu: str) -> int:
    """Ürün koduna göre stok adedini döndürür."""
    return STOK.get(urun_kodu, 0)

tools = [stok_sorgula]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("claude-sonnet-5-5", temperature=0)
model_with_tools = model.bind_tools(tools)

class Durum(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

def model_dugumu(state: Durum):
    sistem = SystemMessage(content="Stok sorularını araçla yanıtla. Bilmediğini uydurma.")
    return {"messages": [model_with_tools.invoke([sistem] + state["messages"])]}

def arac_dugumu(state: Durum):
    sonuclar = []
    for cagri in state["messages"][-1].tool_calls:
        arac = tools_by_name[cagri["name"]]
        gozlem = arac.invoke(cagri["args"])
        sonuclar.append(ToolMessage(content=gozlem, tool_call_id=cagri["id"]))
    return {"messages": sonuclar}

def devam_mi(state: Durum) -> Literal["arac_dugumu", END]:
    if state["messages"][-1].tool_calls:
        return "arac_dugumu"
    return END

builder = StateGraph(Durum)
builder.add_node("model_dugumu", model_dugumu)
builder.add_node("arac_dugumu", arac_dugumu)
builder.add_edge(START, "model_dugumu")
builder.add_conditional_edges("model_dugumu", devam_mi, ["arac_dugumu", END])
builder.add_edge("arac_dugumu", "model_dugumu")

agent = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "musteri-1"}}
cevap = agent.invoke({"messages": [HumanMessage(content="A100 kodlu üründen kaç adet var?")]}, config)
print(cevap["messages"][-1].content)

Döngünün mantığı kısa: model bir araç çağırmak isterse araç düğümüne gidiyor, araç sonucu modele dönüyor, model araç istemeyene kadar bu sürüyor. Bu, dokümanın tarif ettiği temel ajan döngüsünün kendisi. Daha hızlı bir yol istiyorsanız LangChain'in create_agent işlevi aynı döngüyü hazır sunuyor (model, araçlar, sistem istemi, checkpointer ve ara katman parametreleriyle). Kendi grafınızı yazmak, akışın nerede dallanacağını ve nerede duracağını denetlemek istediğinizde anlam kazanıyor.

Grafın sonsuza gitmesini de sınırlar: varsayılan özyineleme sınırı 1000 süper adım (1.0.6 sürümünden itibaren), aşılırsa GraphRecursionError fırlatılıyor. Bu sınırı configurable içine değil, yapılandırmanın üst düzeyine recursion_limit anahtarıyla koyuyorsunuz.

Tasarım sırası

LangGraph dokümanındaki "LangGraph'ta düşünmek" sayfası beş adım öneriyor ve ben de bu sırayı izliyorum:

İş akışını ayrı adımlara bölün. Her adım bir düğüm olacak ve tek bir iş yapacak.
Her düğümün ne tür bir işlem yaptığını sınıflayın: model adımı (anlama, analiz, metin üretimi), veri adımı (arama, kayıt getirme), eylem adımı (yanıt gönderme, kayıt açma gibi dış etki), kullanıcı girdisi adımı (onay ya da eksik bilgi bekleme).
Durumu tasarlayın.
Düğümleri yazın.
Grafı bağlayın.

Durum tasarımında doküman iki ilke veriyor. Durumda yalnızca adımlar arasında kalıcı olması gereken veriyi tutun: özgün girdi, yeniden hesaplanması pahalı ara sonuçlar, hata ayıklama bilgisi. Hesaplanabilir olanı türetin. İkincisi, "ham veriyi saklayın, istemi gerektiğinde biçimlendirin." Biçimlendirilmiş metni durumda tutarsanız, istem şablonunu değiştirmek durum şemasını da değiştirmenizi gerektirir.

Veri adımlarında yeniden deneme ve önbellek işe yarıyor. Eylem adımlarını genelde önbelleğe almamalısınız.

Hata yönetimi

Doküman hataları türüne göre ele alıyor:

Geçici hatalar (ağ sorunları, hız sınırı): otomatik yeniden deneme politikası, genellikle zaman aşımıyla birlikte.
Modelin düzeltebileceği hatalar (araç hatası, ayrıştırma sorunu): hatayı duruma yazıp döngüyü geri çevirin, model yaklaşımını ayarlasın.
Kullanıcının düzeltebileceği hatalar (eksik bilgi): interrupt() ile durup bilgi gelince devam edin.
Denemeler tükendiğinde: geliştiricinin tanımladığı kurtarma ya da telafi dalına yönlendirin.
Beklenmeyen hatalar: geniş bir try/except ile yutmak yerine yukarı çıksın, hata ayıklayın.

En son maddeyi vurgulamak istiyorum. Ajan kodunda her şeyi yakalayıp "bir sorun oluştu" demek cazip, ama bu, hatanın nedenini görünmez kılıyor.

Kalıcılık: checkpointer ve thread_id

Bir ajan birden fazla mesaj boyunca durum tutacaksa ya da kesintiden sonra devam edecekse grafı bir checkpointer ile derlemeniz gerekiyor. Checkpointer, durumun her adımdaki anlık görüntüsünü bir iş parçacığı (thread) altında saklıyor. Aynı thread_id ile çağırdığınızda konuşma kaldığı yerden sürüyor.

Seçim üretimde önemli. InMemorySaver ve MemorySaver kayıtları bellekte tutuyor, süreç yeniden başladığında hepsi gidiyor. Geliştirme için SqliteSaver dosyada saklıyor. Üretim için PostgresSaver (eşzamansız kodda AsyncPostgresSaver) öneriliyor. Postgres ile thread_id 255 karakterin altında olmalı. Uzun konuşmalarda kontrol noktaları birikiyor, eskilerini düzenli budamak gerekiyor.

Kalıcılığın iki katmanı var: checkpointer iş parçacığı içi kısa vadeli belleği ve kurtarmayı sağlıyor, store ise iş parçacıkları arası veriyi (kullanıcı tercihleri gibi) saklıyor.

Onay adımı: interrupt

Ajan e-posta göndermek, kayıt silmek ya da ödeme başlatmak gibi geri alınması zor bir eylem yapacaksa, bu eylemden önce bir insan onayı koymanızı öneririm. LangGraph'ta bunun aracı interrupt(). Çağrıldığı yerde çalışmayı durduruyor, durumu kaydediyor ve argümanını (JSON'a çevrilebilir olmalı) çağırana döndürüyor. Command(resume=...) ile devam ediyorsunuz ve verdiğiniz değer interrupt() çağrısının dönüş değeri oluyor:

from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class State(TypedDict):
    approved: bool

def onay_dugumu(state: State):
    approved = interrupt("Bu işlemi onaylıyor musunuz?")
    return {"approved": approved}

builder = StateGraph(State)
builder.add_node("onay", onay_dugumu)
builder.add_edge(START, "onay")
builder.add_edge("onay", END)
graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "thread-1"}}
sonuc = graph.invoke({"approved": False}, config)
print(sonuc["__interrupt__"])
sonuc = graph.invoke(Command(resume=True), config)
print(sonuc["approved"])

Üç koşul var: grafın bir checkpointer'ı olmalı, devam ederken kesintidekiyle aynı thread_id kullanılmalı ve düğüm devam ederken baştan çalıştırıldığı için interrupt() öncesindeki kod tekrar çalışacak. Bu yüzden interrupt() çağrısından önce yan etkisi olan kod (e-posta gönderme gibi) bulunmamalı ya da o kod tekrar çalışsa zarar vermeyecek biçimde yazılmalı. Çağrıları çıplak try/except içine de sarmayın.

Nerede işe yarıyor

LangGraph'ın README'si kendini uzun süre çalışan, durumlu ajan ve iş akışları için konumlandırıyor ve kullanan şirketler arasında Klarna, Replit ve Elastic'i sayıyor. Uygun düşen işler belge sınıflama, bir müşteri talebini doğru ekibe yönlendirme ya da rutin raporu hazırlayıp insan onayına sunma gibi, adımları büyük ölçüde belli süreçler. Bu tür işlerde ajanın tek başına karar vermesinden çok, iş akışının belirli bir adımda modeli çağırması hem denetlenebilir hem hata ayıklaması kolay.

İki bağlantı kurmak istiyorum. Birincisi, bu ajanların veriyi çektiği yer şirket içi sistemler. Müşteri kayıtlarını ve süreçleri tutan bir yapı yoksa ajanın erişeceği şey de yok, bu yüzden özel CRM yazılımı yazısı bu işin öncülü. İkincisi, tekrar eden denetimleri bir graf olarak kurmak mümkün. Örneğin teknik SEO denetim listesindeki durum kodu ve robots.txt kontrolleri belirli adımlar, modeli yalnızca bulguları yorumlamak için çağırırsınız. Google'ın kendi yaklaşımı da ajanların siteyle etkileşimini, siteler için isteğe bağlı bir konu olarak sunuyor (yapay zeka aramaları yazısı). Böyle bir sistem kurmak isterseniz işler sayfasına bakın ya da iletişim üzerinden yazın.

Üretime çıkmadan önce kontrol listesi

İş, düz bir iş akışıyla çözülemiyor mu? Çözülüyorsa ajan kurmayın.
Durumda yalnızca kalıcı olması gereken ham veri mi var?
Üretimde PostgresSaver (ya da eşdeğeri) kullanılıyor mu, thread_id stratejisi belli mi?
Geri alınması zor her eylemden önce interrupt() ile insan onayı var mı?
interrupt() öncesindeki kod tekrar çalıştırılınca zarar vermiyor mu?
Özyineleme sınırı ve hata dalları (yeniden deneme, kurtarma) tanımlı mı?
Her çalıştırmanın izi (LangSmith ya da kendi günlüğünüz) saklanıyor mu?

Tek bir adım atacaksanız önce en küçük iş akışını düğüm ve kenarlarıyla kâğıda çizin. Çizimi bitiremiyorsanız kodu yazmaya hazır değilsiniz.