LangGraph: AI Agent Framework Production Rehberi 2026

1from langgraph.graph import StateGraph, END
2from typing import TypedDict, Annotated
3import operator
4 
5# State tanımı — grafik boyunca paylaşılan veri
6class AgentState(TypedDict):
7    messages: Annotated[list, operator.add]  # reducer: append
8    next_step: str
9    error_count: int
10    result: str | None

1from langgraph.graph import StateGraph, END
2from langchain_anthropic import ChatAnthropic
3 
4llm = ChatAnthropic(model="claude-sonnet-4-5-20250514")
5 
6def analyze_node(state: AgentState) -> AgentState:
7    """Kullanıcı isteğini analiz et ve plan oluştur."""
8    messages = state["messages"]
9    response = llm.invoke(messages + [
10        {"role": "system", "content": "Kullanıcı isteğini analiz et, plan yap."}
11    ])
12    return {
13        "messages": [response],
14        "next_step": "execute"
15    }
16 
17def execute_node(state: AgentState) -> AgentState:
18    """Planı uygula."""
19    # ... implementation
20    return {"result": "tamamlandı", "next_step": "review"}
21 
22def review_node(state: AgentState) -> AgentState:
23    """Sonucu gözden geçir."""
24    # ... implementation
25    return {"next_step": "end"}
26 
27# Graf oluştur
28builder = StateGraph(AgentState)
29builder.add_node("analyze", analyze_node)
30builder.add_node("execute", execute_node)
31builder.add_node("review", review_node)
32 
33# Entry point
34builder.set_entry_point("analyze")
35 
36# Edge"ler
37builder.add_edge("analyze", "execute")
38builder.add_edge("execute", "review")
39builder.add_edge("review", END)
40 
41graph = builder.compile()

1from langgraph.prebuilt import ToolNode
2from langchain_core.tools import tool
3 
4@tool
5def search_database(query: str) -> str:
6    """Veritabanında arama yap."""
7    # ... gerçek implementation
8    return f"Sonuç: {query} için 5 kayıt bulundu"
9 
10@tool
11def send_notification(message: str, channel: str) -> str:
12    """Bildirim gönder."""
13    # ... gerçek implementation
14    return f"Bildirim gönderildi: {channel}"
15 
16tools = [search_database, send_notification]
17tool_node = ToolNode(tools)
18 
19# LLM"i tool"larla bağla
20llm_with_tools = llm.bind_tools(tools)
21 
22def agent_node(state: AgentState) -> AgentState:
23    response = llm_with_tools.invoke(state["messages"])
24    return {"messages": [response]}
25 
26builder.add_node("agent", agent_node)
27builder.add_node("tools", tool_node)

1from langgraph.graph import MessagesState
2 
3# MessagesState: messages field"ını otomatik yönetir
4class MyState(MessagesState):
5    # Ekstra field"lar
6    task_complete: bool = False
7 
8# Bu pattern ReAct agent için ideal

1from typing import Literal
2 
3def route_after_agent(state: AgentState) -> Literal["tools", "review", "__end__"]:
4    """Agent çıktısına göre rota belirle."""
5    last_message = state["messages"][-1]
6 
7    # Tool çağrısı var mı?
8    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
9        return "tools"
10 
11    # Hata sayısı limitini aştı mı?
12    if state.get("error_count", 0) >= 3:
13        return "__end__"
14 
15    return "review"
16 
17builder.add_conditional_edges(
18    "agent",
19    route_after_agent,
20    {
21        "tools": "tools",
22        "review": "review",
23        "__end__": END,
24    }
25)
26 
27# Tools"tan agent"a geri dön (döngü!)
28builder.add_edge("tools", "agent")

1from langgraph.constants import Send
2 
3def fan_out_node(state: AgentState) -> list[Send]:
4    """Her belge için ayrı agent başlat."""
5    documents = state["documents"]
6    return [
7        Send("process_doc", {"doc": doc, "doc_id": i})
8        for i, doc in enumerate(documents)
9    ]
10 
11builder.add_conditional_edges("prepare", fan_out_node)

1from langgraph.checkpoint.sqlite import SqliteSaver
2 
3with SqliteSaver.from_conn_string("checkpoints.db") as saver:
4    graph = builder.compile(checkpointer=saver)
5 
6    config = {"configurable": {"thread_id": "user-session-123"}}
7 
8    # İlk çalıştırma
9    result = graph.invoke(
10        {"messages": [{"role": "user", "content": "Rapor oluştur"}]},
11        config=config
12    )
13 
14    # Aynı thread_id ile devam — checkpoint"ten kaldığı yer
15    result2 = graph.invoke(
16        {"messages": [{"role": "user", "content": "Devam et"}]},
17        config=config
18    )

1from langgraph.checkpoint.postgres import PostgresSaver
2import psycopg
3 
4# Connection string
5DB_URI = "postgresql://user:pass@localhost:5432/langgraph_checkpoints"
6 
7with PostgresSaver.from_conn_string(DB_URI) as saver:
8    # Schema oluştur (ilk kez)
9    saver.setup()
10 
11    graph = builder.compile(checkpointer=saver)
12 
13    config = {
14        "configurable": {
15            "thread_id": "prod-task-456",
16            "checkpoint_ns": "v1"  # versioning için
17        }
18    }
19 
20    # Async context"te
21    async with PostgresSaver.from_conn_string(DB_URI) as async_saver:
22        async_graph = builder.compile(checkpointer=async_saver)
23        result = await async_graph.ainvoke(initial_state, config=config)

1from langgraph.checkpoint.redis import RedisSaver
2 
3redis_saver = RedisSaver.from_conn_string("redis://localhost:6379")
4graph = builder.compile(checkpointer=redis_saver)
5 
6# TTL ile — 24 saat sonra otomatik sil
7config = {
8    "configurable": {
9        "thread_id": "ephemeral-task",
10        "ttl": 86400  # saniye
11    }
12}

1# Interrupt before bir node"dan önce dur
2graph = builder.compile(
3    checkpointer=saver,
4    interrupt_before=["execute_payment"],  # ödeme öncesi dur
5)
6 
7config = {"configurable": {"thread_id": "order-789"}}
8 
9# 1. Adım: Plan oluştur (otomatik)
10initial_result = graph.invoke(
11    {"messages": [{"role": "user", "content": "150 dolar ödeme yap"}]},
12    config=config
13)
14 
15# Grafik "execute_payment" node"undan önce durdu
16# Şu an state"i inceleyebilir, onaylayabilir veya iptal edebilirsin
17 
18print("Mevcut plan:", initial_result)
19user_approval = input("Onaylıyor musunuz? (y/n): ")
20 
21if user_approval == "y":
22    # Kaldığı yerden devam
23    final_result = graph.invoke(None, config=config)
24else:
25    # İptal et — farklı state ile yeniden başlat
26    final_result = graph.invoke(
27        {"messages": [{"role": "user", "content": "İptal et"}]},
28        config=config
29    )

1graph = builder.compile(
2    checkpointer=saver,
3    interrupt_after=["generate_report"],  # rapor oluştuktan sonra dur
4)
5 
6# Rapor oluştu, kullanıcı gözden geçiriyor
7result = graph.invoke(initial_state, config=config)
8report = result["report"]
9 
10# Kullanıcı düzenleme yapıyor
11edited_report = edit_report_ui(report)
12 
13# Düzenlenmiş raporu state"e koy ve devam
14graph.update_state(
15    config,
16    {"report": edited_report, "approved": True}
17)
18 
19final = graph.invoke(None, config=config)

1# Stream mode"ları: "values", "updates", "debug"
2async def stream_agent_response():
3    config = {"configurable": {"thread_id": "stream-demo"}}
4 
5    async for event in graph.astream_events(
6        {"messages": [{"role": "user", "content": "Analiz yap"}]},
7        config=config,
8        version="v2"
9    ):
10        kind = event["event"]
11 
12        if kind == "on_chat_model_stream":
13            # LLM token"ları gerçek zamanlı
14            chunk = event["data"]["chunk"]
15            if chunk.content:
16                print(chunk.content, end="", flush=True)
17 
18        elif kind == "on_tool_start":
19            # Tool çağrısı başladı
20            print(f"\n[Tool: {event['name']} başladı]")
21 
22        elif kind == "on_tool_end":
23            # Tool tamamlandı
24            print(f"[Tool: {event['name']} tamamlandı]")
25 
26        elif kind == "on_chain_end":
27            # Node tamamlandı
28            if event["name"] in ["analyze", "execute", "review"]:
29                print(f"\n[Adım: {event['name']} ✓]")

1from fastapi import FastAPI
2from fastapi.responses import StreamingResponse
3 
4app = FastAPI()
5 
6@app.post("/run-agent")
7async def run_agent(request: dict):
8    async def event_generator():
9        config = {"configurable": {"thread_id": request["thread_id"]}}
10 
11        async for event in graph.astream_events(
12            {"messages": [{"role": "user", "content": request["prompt"]}]},
13            config=config,
14            version="v2"
15        ):
16            if event["event"] == "on_chat_model_stream":
17                chunk = event["data"]["chunk"].content
18                if chunk:
19                    yield f"data: {chunk}\n\n"
20 
21        yield "data: [DONE]\n\n"
22 
23    return StreamingResponse(
24        event_generator(),
25        media_type="text/event-stream"
26    )

1from langchain_core.messages import HumanMessage
2 
3def supervisor_node(state):
4    """Hangi worker agent çalışacak?"""
5    system_prompt = """
6    Mevcut worker"lar: researcher, writer, reviewer
7    Göreve göre uygun worker"ı seç veya FINISH de.
8    """
9    response = llm.invoke([
10        {"role": "system", "content": system_prompt},
11        *state["messages"]
12    ])
13    return {"next": response.content.strip()}
14 
15def route_supervisor(state) -> str:
16    next_agent = state.get("next", "FINISH")
17    if next_agent == "FINISH":
18        return END
19    return next_agent
20 
21builder.add_node("supervisor", supervisor_node)
22builder.add_node("researcher", researcher_agent)
23builder.add_node("writer", writer_agent)
24builder.add_node("reviewer", reviewer_agent)
25 
26builder.set_entry_point("supervisor")
27for agent in ["researcher", "writer", "reviewer"]:
28    builder.add_edge(agent, "supervisor")
29 
30builder.add_conditional_edges("supervisor", route_supervisor)

1from langgraph_swarm import create_swarm, create_handoff_tool
2 
3# Her agent diğerine handoff yapabilir
4research_handoff = create_handoff_tool(
5    agent_name="writer",
6    description="Araştırma tamamlandı, yazara devret"
7)
8 
9researcher = create_react_agent(
10    llm,
11    tools=[web_search, research_handoff],
12    name="researcher"
13)
14 
15writer = create_react_agent(
16    llm,
17    tools=[text_editor, create_handoff_tool("reviewer")],
18    name="writer"
19)
20 
21swarm = create_swarm(
22    [researcher, writer, reviewer],
23    default_active_agent="researcher"
24)

1# Alt grafik oluştur (sub-graph)
2research_graph = create_research_subgraph()
3writing_graph = create_writing_subgraph()
4 
5# Ana grafik alt grafikleri node olarak kullanır
6main_builder = StateGraph(MainState)
7main_builder.add_node("research_team", research_graph)
8main_builder.add_node("writing_team", writing_graph)
9main_builder.add_edge("research_team", "writing_team")
10main_builder.add_edge("writing_team", END)
11 
12main_graph = main_builder.compile()

1from tenacity import retry, stop_after_attempt, wait_exponential
2from langgraph.errors import NodeInterrupt
3 
4def resilient_api_node(state: AgentState) -> AgentState:
5    """Hata toleranslı API çağrısı."""
6    try:
7        result = call_external_api(state["query"])
8        return {"result": result, "error_count": 0}
9    except RateLimitError:
10        # Rate limit — bekle ve devam et
11        error_count = state.get("error_count", 0) + 1
12        if error_count >= 3:
13            # İnsan müdahalesi gerekiyor
14            raise NodeInterrupt(
15                f"Rate limit 3 kez aşıldı. Manuel müdahale gerekiyor."
16            )
17        return {"error_count": error_count, "next_step": "wait"}
18    except Exception as e:
19        # Kritik hata — kayıt al ve devam et
20        print(f"Hata: {e}")
21        return {"error_count": state.get("error_count", 0) + 1}
22 
23 
24@retry(
25    stop=stop_after_attempt(3),
26    wait=wait_exponential(multiplier=1, min=4, max=10)
27)
28def call_external_api(query: str) -> str:
29    """Retry mantığı ile dış API çağrısı."""
30    response = requests.post("https://api.example.com/analyze", json={"q": query})
31    response.raise_for_status()
32    return response.json()["result"]

1def route_on_error(state: AgentState) -> str:
2    error_count = state.get("error_count", 0)
3    if error_count == 0:
4        return "success_path"
5    elif error_count < 3:
6        return "retry_path"
7    else:
8        return "fallback_path"  # Basit, hatasız alternatif
9 
10builder.add_conditional_edges(
11    "api_call",
12    route_on_error,
13    {
14        "success_path": "process_result",
15        "retry_path": "api_call",     # Döngü
16        "fallback_path": "simple_handler"
17    }
18)

1import os
2os.environ["LANGSMITH_API_KEY"] = "ls__your_key"
3os.environ["LANGSMITH_TRACING"] = "true"
4os.environ["LANGSMITH_PROJECT"] = "production-agents"
5 
6# Artık tüm LangGraph çalıştırmaları otomatik trace ediliyor
7# LangSmith UI"da görülebilecekler:
8# - Her node"un çalışma süresi
9# - Her LLM çağrısının token kullanımı
10# - Tool çağrıları ve sonuçları
11# - Hatalar ve stack trace
12# - State değişimleri
13 
14# Özel metadata ekle
15config = {
16    "configurable": {"thread_id": "task-123"},
17    "metadata": {
18        "user_id": "user-456",
19        "task_type": "report_generation",
20        "version": "v2.1"
21    }
22}

1from langsmith import Client
2 
3client = Client()
4 
5# Çalışma geçmişini sorgula
6runs = client.list_runs(
7    project_name="production-agents",
8    run_type="chain",
9    filter="and(eq(status, 'success'), gt(total_tokens, 5000))"
10)
11 
12for run in runs:
13    print(f"Thread: {run.extra.get('thread_id')}")
14    print(f"Total tokens: {run.total_tokens}")
15    print(f"Duration: {run.end_time - run.start_time}")
16    print(f"Cost estimate: ${run.total_tokens * 0.000003:.4f}")

1# langgraph.json — deployment config
2{
3  "dependencies": ["."],
4  "graphs": {
5    "agent": "./src/agent.py:graph"
6  },
7  "env": ".env.production"
8}

1# Deploy
2pip install langgraph-cli
3langgraph build -t my-agent:latest
4langgraph deploy --project my-prod-agent

1FROM python:3.12-slim
2 
3WORKDIR /app
4COPY requirements.txt .
5RUN pip install -r requirements.txt
6 
7COPY . .
8 
9# PostgreSQL için wait script
10COPY wait-for-db.sh .
11RUN chmod +x wait-for-db.sh
12 
13CMD ["./wait-for-db.sh", "python", "-m", "uvicorn", "main:app", "--host", "0.0.0.0"]

1# docker-compose.yml
2services:
3  agent:
4    build: .
5    environment:
6      - DATABASE_URL=postgresql://postgres:pass@db:5432/langgraph
7      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
8    depends_on:
9      - db
10      - redis
11 
12  db:
13    volumes:
14      - pgdata:/var/lib/postgresql/data
15 
16  redis:
17    command: redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru

1import asyncio
2from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
3 
4async def run_long_task(task_id: str, initial_state: dict):
5    """24 saat+ sürebilen görev."""
6    async with AsyncPostgresSaver.from_conn_string(DB_URI) as saver:
7        graph = builder.compile(checkpointer=saver)
8        config = {"configurable": {"thread_id": task_id}}
9 
10        try:
11            # Görev başlat
12            async for event in graph.astream_events(
13                initial_state,
14                config=config,
15                version="v2"
16            ):
17                # Progress kaydet
18                await update_task_progress(task_id, event)
19 
20                # Uzun beklemeler için heartbeat
21                if event["event"] == "on_chain_start":
22                    await update_task_heartbeat(task_id)
23 
24        except asyncio.CancelledError:
25            # Graceful shutdown — checkpoint kaydedildi, sonra devam edilebilir
26            print(f"Task {task_id} duraklatıldı, checkpoint kaydedildi")

1from typing import TypedDict, Annotated
2import operator
3from langgraph.graph import StateGraph, END
4from langgraph.checkpoint.postgres import PostgresSaver
5from langchain_anthropic import ChatAnthropic
6 
7class ResearchState(TypedDict):
8    topic: str
9    messages: Annotated[list, operator.add]
10    research_notes: list[str]
11    draft: str | None
12    revision_count: int
13    approved: bool
14 
15def research_node(state: ResearchState) -> ResearchState:
16    notes = []
17    for query in generate_search_queries(state["topic"]):
18        results = web_search(query)
19        notes.extend(parse_search_results(results))
20    return {"research_notes": notes}
21 
22def write_node(state: ResearchState) -> ResearchState:
23    prompt = f"""
24    Konu: {state["topic"]}
25    Notlar: {state["research_notes"]}
26    Taslak oluştur. Revision #{state["revision_count"] + 1}
27    """
28    response = llm.invoke([{"role": "user", "content": prompt}])
29    return {
30        "draft": response.content,
31        "revision_count": state["revision_count"] + 1
32    }
33 
34def quality_check_node(state: ResearchState) -> ResearchState:
35    # Kalite skoru hesapla
36    score = evaluate_draft_quality(state["draft"])
37    return {"approved": score >= 0.85}
38 
39def route_quality(state: ResearchState) -> str:
40    if state["approved"]:
41        return "publish"
42    if state["revision_count"] >= 3:
43        return "human_review"  # İnsan müdahalesi
44    return "write"  # Yeniden yaz
45 
46builder = StateGraph(ResearchState)
47builder.add_node("research", research_node)
48builder.add_node("write", write_node)
49builder.add_node("quality_check", quality_check_node)
50builder.add_node("publish", publish_node)
51builder.add_node("human_review", human_review_node)
52 
53builder.set_entry_point("research")
54builder.add_edge("research", "write")
55builder.add_edge("write", "quality_check")
56builder.add_conditional_edges("quality_check", route_quality)
57builder.add_edge("publish", END)
58builder.add_edge("human_review", END)
59 
60with PostgresSaver.from_conn_string(DB_URI) as saver:
61    graph = builder.compile(
62        checkpointer=saver,
63        interrupt_before=["human_review"]
64    )

1# Subgraph kendi state"ini tanımlıyor
2class SubgraphState(TypedDict):
3    messages: Annotated[list, operator.add]
4    sub_result: str  # Ana grafta yok
5 
6# Ana state"e map etmek için transform fonksiyonu
7def subgraph_input_transform(state: MainState) -> SubgraphState:
8    return {"messages": state["messages"][-5:]}  # Son 5 mesaj
9 
10def subgraph_output_transform(state: SubgraphState) -> dict:
11    return {"processed_data": state["sub_result"]}
12 
13main_builder.add_node(
14    "subgraph",
15    subgraph_graph,  # Derlenmiş alt grafik
16    # Input/output transform ile state izolasyonu
17)

1from celery import Celery
2from langgraph.checkpoint.redis import RedisSaver
3 
4celery_app = Celery("tasks", broker="redis://localhost:6379/0")
5 
6@celery_app.task(bind=True, max_retries=3)
7def run_agent_task(self, thread_id: str, initial_state: dict):
8    """Background"da agent çalıştır."""
9    try:
10        redis_saver = RedisSaver.from_conn_string("redis://localhost:6379")
11        graph = builder.compile(checkpointer=redis_saver)
12 
13        config = {"configurable": {"thread_id": thread_id}}
14        result = graph.invoke(initial_state, config=config)
15 
16        # Sonucu veritabanına kaydet
17        save_result_to_db(thread_id, result)
18        return {"status": "success", "thread_id": thread_id}
19 
20    except Exception as exc:
21        # Retry: exponential backoff
22        raise self.retry(exc=exc, countdown=2 ** self.request.retries)
23 
24# Kullanım
25task = run_agent_task.delay(
26    thread_id="bg-task-123",
27    initial_state={"topic": "AI trends 2026", "messages": []}
28)
29 
30# Sonuç sorgula
31result = celery_app.AsyncResult(task.id)
32print(result.status)  # PENDING / SUCCESS / FAILURE

1import uuid
2 
3# Kullanıcı oturumu bazlı
4thread_id = f"user-{user_id}-session-{session_id}"
5 
6# Görev bazlı (idempotent)
7thread_id = f"task-{task_type}-{content_hash}"
8 
9# Tek seferlik
10thread_id = str(uuid.uuid4())

1import signal
2 
3def shutdown_handler(signum, frame):
4    # Mevcut graph çalışmasını bekle
5    # Checkpoint kaydedildi — devam edilebilir
6    print("Graceful shutdown başlatıldı...")
7    shutdown_event.set()
8 
9signal.signal(signal.SIGTERM, shutdown_handler)