5 Mar 2026

Ajan Hata Kurtarma: Üretim Güvenilirliği için 5 Desen

Ajanın test ortamında mükemmel çalışıyordu. Sonra üretimde, çok adımlı bir iş akışının 3. adımında hız sınırına çarptı, yakalanmamış bir istisna fırlattı ve sistemi tanımsız bir durumda bıraktı. Checkpoint yok. Yeniden deneme yok. Yedek plan yok. Sadece sessizlik — ve elle yeniden başlatılması gereken bozuk bir pipeline.

Ajan hata kurtarma, bir demo ile üretim sistemi arasındaki farktır. Bu makale üretim ajansal iş akışlarında kullanılan beş deseni kapsar: üstel geri çekilme, devre kesici, checkpoint-ve-sürdürme, yedek stratejiler ve yükseltme kuyrukları. Her desen Anthropic SDK ile uygulanmıştır.

Ön koşullar: Python’a hakim olmak ve Claude API’nin nasıl çalıştığını bilmek.

Ajanlar Neden Geleneksel Yazılımdan Farklı Başarısız Olur

Ajanlar daha karmaşık şekillerde başarısız olur:

Kısmi ilerleme: Bir ajan 8 adımlı iş akışının 1–4. adımlarını tamamlar, sonra başarısız olur. Kurtarma olmadan tüm ilerleme kaybedilir.
Belirsiz durum: Ajan bir araç çağırdı ama yanıt hatalı biçimliydi. Eylem gerçekleşti mi? Yeniden denenmelidir mi?
Basamaklı başarısızlık: Yavaş bir API çağrısı tüm akıl yürütme döngüsünü bloke eder.
Sessiz başarısızlık: LLM bir yanıt döndürür ama beklenen biçimi izlemez. Aşağı akış ayrıştırıcısı sessizce bozulur.

Desen 1: Titreme ile Üstel Geri Çekilme

Üstel geri çekilme, yeniden denemeler arasındaki bekleme süresini ikiye katlar. Titreme (jitter), birden fazla ajanın aynı anda yeniden denemesini önlemek için rastgelelik ekler.

import anthropic
import time
import random
from typing import Optional

client = anthropic.Anthropic()

def call_with_backoff(
    messages: list,
    model: str = "claude-sonnet-4-6",
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
) -> Optional[anthropic.types.Message]:
    """
    Geçici hatalarda üstel geri çekilme ile Claude API'yi çağırır.
    Hız sınırları ve sunucu hatalarında yeniden dener;
    istemci hatalarında hemen fırlatır.
    """
    for attempt in range(max_retries):
        try:
            return client.messages.create(
                model=model,
                max_tokens=1024,
                messages=messages,
            )
        except anthropic.RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay)
            print(f"Hız sınırı. {jitter:.1f}s içinde yeniden deniyor (deneme {attempt + 1}/{max_retries})")
            time.sleep(jitter)
        except anthropic.APIStatusError as e:
            if e.status_code < 500:
                raise
            if attempt == max_retries - 1:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay)
            print(f"Sunucu hatası {e.status_code}. {jitter:.1f}s içinde yeniden deniyor")
            time.sleep(jitter)
        except anthropic.APIConnectionError:
            if attempt == max_retries - 1:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            time.sleep(delay)
    return None

Ne zaman kullanılır: Ajan döngüsündeki her API çağrısında. Tüm LLM çağrıları için varsayılan sarıcı olmalıdır.

Desen 2: Devre Kesici

Devre kesici, başarısızlık oranlarını takip eder ve başarısız olan bir servisi çağırmayı geçici olarak durdurur. Üç durum: Kapalı (normal), Açık (başarısız) ve Yarı açık (toparlanıyor).

import time
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, TypeVar, Any

T = TypeVar("T")

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: float = 60.0
    success_threshold: int = 2

    _state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    _failure_count: int = field(default=0, init=False)
    _success_count: int = field(default=0, init=False)
    _last_failure_time: float = field(default=0.0, init=False)

    @property
    def state(self) -> CircuitState:
        if self._state == CircuitState.OPEN:
            if time.time() - self._last_failure_time > self.recovery_timeout:
                self._state = CircuitState.HALF_OPEN
                self._success_count = 0
        return self._state

    def call(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
        if self.state == CircuitState.OPEN:
            raise RuntimeError("Devre kesici AÇIK — servis kullanılamıyor")
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception:
            self._on_failure()
            raise

    def _on_success(self) -> None:
        self._failure_count = 0
        if self._state == CircuitState.HALF_OPEN:
            self._success_count += 1
            if self._success_count >= self.success_threshold:
                self._state = CircuitState.CLOSED

    def _on_failure(self) -> None:
        self._failure_count += 1
        self._last_failure_time = time.time()
        if self._failure_count >= self.failure_threshold:
            self._state = CircuitState.OPEN

Desen 3: Checkpoint ve Sürdürme

Uzun süre çalışan ajanlar, bir başarısızlık sonrasında son başarılı adımdan devam edebilmelidir.

import json
import os
from dataclasses import dataclass, asdict, field
from typing import Optional

@dataclass
class AgentCheckpoint:
    task_id: str
    current_step: int
    completed_steps: list[str] = field(default_factory=list)
    results: dict = field(default_factory=dict)
    messages: list[dict] = field(default_factory=list)

    def save(self, directory: str = ".checkpoints") -> None:
        os.makedirs(directory, exist_ok=True)
        path = os.path.join(directory, f"{self.task_id}.json")
        with open(path, "w") as f:
            json.dump(asdict(self), f, indent=2)

    @classmethod
    def load(cls, task_id: str, directory: str = ".checkpoints") -> Optional["AgentCheckpoint"]:
        path = os.path.join(directory, f"{task_id}.json")
        if not os.path.exists(path):
            return None
        with open(path) as f:
            data = json.load(f)
        return cls(**data)

    def clear(self, directory: str = ".checkpoints") -> None:
        path = os.path.join(directory, f"{self.task_id}.json")
        if os.path.exists(path):
            os.remove(path)

Bu desen LangGraph’ın yerleşik MemorySaver ve PostgresSaver checkpoint’leri ile doğal olarak bütünleşir. Bkz. LangGraph ile Durum Makineleri ve Ajanlar.

Desen 4: Yedek Stratejiler

Bazı başarısızlıklar yeniden denenemez. Ajanı ilerletmeye devam eden alternatif bir yola ihtiyacın var.

from typing import Callable

def with_fallback(
    primary: Callable[[], str],
    fallback: Callable[[], str],
    error_types: tuple = (Exception,),
) -> str:
    """Birincil fonksiyonu dener; belirtilen hatalarda ikincile geçer."""
    try:
        return primary()
    except error_types as e:
        print(f"Birincil başarısız ({type(e).__name__}): {e}. Yedek kullanılıyor.")
        return fallback()


def answer_question(question: str) -> str:
    def primary():
        # Birincil model: en yetenekli
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": question}],
        )
        return response.content[0].text

    def fallback():
        # Yedek: daha hafif ve hızlı model
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{"role": "user", "content": question}],
        )
        return f"[Yedek yanıt] {response.content[0].text}"

    return with_fallback(
        primary, fallback,
        error_types=(anthropic.RateLimitError, anthropic.APIStatusError),
    )

Desen 5: Yükseltme Kuyruğu

Bazı başarısızlıklar gerçekten otomatik olarak çözülemez. Yükseltme kuyruğu, bir insanın çözebilmesi için yeterli bağlamla birlikte başarısız görevleri yakalar.

import json
import uuid
from datetime import datetime
from dataclasses import dataclass, asdict
from enum import Enum

class EscalationReason(Enum):
    MAX_RETRIES_EXCEEDED = "max_retries_exceeded"
    AMBIGUOUS_STATE = "ambiguous_state"
    LOW_CONFIDENCE = "low_confidence"
    REQUIRES_HUMAN = "requires_human"
    IRREVERSIBLE_ACTION = "irreversible_action"

@dataclass
class EscalationRecord:
    id: str
    task_id: str
    reason: str
    error_message: str
    agent_state: dict
    context: str
    timestamp: str
    resolved: bool = False

    def save(self, queue_file: str = "escalation_queue.jsonl") -> None:
        with open(queue_file, "a") as f:
            f.write(json.dumps(asdict(self)) + "\n")


def escalate(
    task_id: str,
    reason: EscalationReason,
    error_message: str,
    agent_state: dict,
    context: str = "",
) -> EscalationRecord:
    record = EscalationRecord(
        id=str(uuid.uuid4()),
        task_id=task_id,
        reason=reason.value,
        error_message=error_message,
        agent_state=agent_state,
        context=context,
        timestamp=datetime.utcnow().isoformat(),
    )
    record.save()
    print(f"[YÜKSELTİLDİ] Görev {task_id}: {reason.value}")
    return record

Ne zaman kullanılır: Geri alınamaz eylemler veya düşük güven kararları içeren görevler için. Bkz. Çok Ajan Desenleri.

Yaygın Hatalar

Hata 1: Yeniden Denenemez Hataları Yeniden Denemek

Çözüm: Yeniden denemeden önce hataları sınıflandır.

# Yanlış: her şeyi yakalar
except Exception:
    retry()

# Doğru: yalnızca yeniden denenebilir hatalar
except (anthropic.RateLimitError, anthropic.APIConnectionError):
    retry()
except anthropic.APIStatusError as e:
    if e.status_code >= 500:
        retry()
    else:
        raise

Hata 2: Yeniden Denemede Durumu Kaybetmek

Çözüm: Denemeler arasında mesaj geçmişini koru. Yalnızca başarısız olan adımı yeniden dene.

Hata 3: Sınırsız Yeniden Deneme Döngüleri

Çözüm: Her zaman max_retries belirle. Tükendikten sonra yükselt veya hızla başarısız ol.

Üretim Kontrol Listesi

Tüm API çağrıları üstel geri çekilme ile sarıldı (taban 1s, maks 60s, titreme aktif)
Her dış bağımlılıkta devre kesici
Her önemli adımdan sonra checkpoint kalıcı depolamaya kaydedildi
Yedek yollar test edildi ve yanıt meta verilerinde işaretlendi
Yükseltme kuyruğu günlük izlendi ve gözden geçirildi
Yeniden deneme sayısı sınırlı (max_retries ≤ 5)

Sonraki Adımlar

Geri çekilme ile başla — Her API çağrısını call_with_backoff ile sar.
Checkpoint ekle 2–3’ten fazla adım içeren iş akışlarına.
Yükseltme kuyruğu oluştur geri alınamaz eylemler içeren görevler için.
Hata enjeksiyon testleri yaz — Her hata türünü kasıtlı olarak tetikleyerek kurtarma yollarını test et.

İlgili kılavuzlar: