5 Mar 2026

Durum Makineleri ve Ajanlar: LangGraph ile Güvenilir İş Akışları Oluşturma

Çoğu ajan eğitimi basit bir döngü gösterir: Claude’a sor, yanıtı çözümle, araç çağır, tekrarla. Bu demolar için işe yarar. Prodüksiyonda determinizm, hata kurtarma, insan onay kapıları ve denetlenebilirlik gereklidir.

LangGraph, durum makinelerini ajan iş akışlarına getirir. if ifadelerinden oluşan geçici bir döngü yerine, açık bir grafik elde edersiniz: adlandırılmış düğümler (mantık birimleri), türlendirilmiş kenarlar (geçişler) ve tüm yürütme boyunca akan paylaşılan durum şeması.

Ajanlar İçin Neden Durum Makineleri?

Geçici Döngü Sorunu

Tipik bir ajan döngüsü şöyle görünür:

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))

Bu iki veya üç araç için okunabilir. Beş araç, koşullu yollar, bir insan onay adımı ve yeniden deneme mantığı ekleyin — yüzlerce satır karmaşık kontrol akışı elde edersiniz.

Daha derin sorun örtük durumdur. Ajan hangi aşamadadır? Hangi verileri topladı? Her şey messages’ta yaşar — her düğümün okuduğu ve eklediği, şema dayatması olmayan tipesiz bir yapı.

Durum Makineleri Bir Çözüm Olarak

Durum makinesi örtük olanı açık hale getirir. Tanımlarsınız:

Düğümler — mevcut durumu alan, bir şey yapan ve durum güncellemeleri döndüren ayrık mantık birimleri.
Kenarlar — düğümler arasındaki geçişler, koşulsuz (A → B her zaman) veya koşullu.
Durum — grafik boyunca akan türlendirilmiş sözlük. Her adımda şema doğrulanır.

LangGraph Kullanılmaması Gereken Durumlar

Basit tek adımlı görevler için doğrudan bir API çağrısı daha hızlı ve nettir. LangGraph’ı şu durumlarda kullanın:

Sırayla çalışması gereken birden fazla farklı aşama
Ara sonuçlara dayalı koşullu dallanma
İnsan katılımı gerektiren adımlar
Hata kurtarma veya yeniden deneme mantığı
Denetlenebilirlik gereksinimleri

LangGraph Temelleri

Durumu Tanımlama

from typing import TypedDict

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

Düğümler

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"Bu konuyu kapsamlı biçimde araştır: {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"Bu notlara dayanarak bir taslak yaz:\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

Kenarlar ve Koşullu Yönlendirme

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"}
)

Grafik Oluşturma ve Çalıştırma

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": "MCP protokolü nedir?"})
print(result["draft"])

Belge İnceleme İş Akışı Oluşturma

Durumu Tasarlama

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

Düğümleri Uygulama

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": (
                "Bu belgeden tam olarak 5 anahtar terim çıkar. "
                "Bunları dize JSON dizisi olarak döndür.\n\n"
                f"Belge: {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": (
                "Bu belgede uyumluluk sorunlarını kontrol et. "
                "Ara: KKB (isimler, T.C. kimlik no, email, telefon), "
                "gizli işaretler, doğrulanmamış iddialar.\n\n"
                "Sorun açıklamalarının JSON dizisini döndür. "
                "Sorun bulunamazsa boş dizi [] döndür.\n\n"
                f"Belge: {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=== İNSAN İNCELEMESİ GEREKLİ ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "KKB redaksiyonu sonrası incelendi ve onaylandı",
        "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"Bu belgeyi tam olarak iki cümleyle özetle:\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

Grafiği Birleştirme

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": "Hasta Ahmet Yılmaz (T.C.: 12345678901) hipertansiyon hastasıdır.",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

İnsan Kontrol Noktaları ve Kesmeler

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("Sorunlar:", current_state.values["compliance_issues"])

graph.update_state(
    thread_config,
    {"human_feedback": "Manuel redaksiyon sonrası onaylandı", "is_approved": True}
)

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

Kalıcılık ve Prodüksiyon Dağıtımı

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
    ])

Yaygın Kalıplar ve Tuzaklar

Fan-Out / Fan-In: Send ile birden fazla düğümü paralel çalıştırın ve sonuçları birleştirin.
Aşırı koşullu yönlendirme: Çok fazla dallanma varsa alt grafiklere bölün.
Durum şişmesi: Büyük yapıtlar yerine referanslar (veritabanı ID’leri, S3 anahtarları) saklayın.
Düğümleri yalıtılmış test etme: Düğümler düz fonksiyonlar olduğundan doğrudan birim testi yapılabilir.

def test_extract_terms_node():
    state = {
        "document": "Yapay zeka modern toplumu hızla dönüştürmektedir.",
        "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, geçici döngülerin yerini açık, hata ayıklanabilir ve devam ettirilebilir iş akışlarıyla alır. Doğrusal bir grafikle başlayın (A → B → C → SON). Dallanma gerektiğinde koşullu bir kenar ekleyin. Onay gerektiğinde insan kontrol noktası ekleyin. Prodüksiyon ajan iş akışlarının çoğu tam olarak bu üç kalıba ihtiyaç duyar.