Ajanlar için Hata Kurtarma

Hatalar olur. Ağlar başarısız olur, API'ler 500 döndürür, Mind Key'ler sona erer. Bu rehber, LLM ajanlarının hataları zarif bir şekilde nasıl ele alacağını ve onlardan nasıl öğreneceğini gösterir.

Hata İşleme İlkeleri

1. Geri çekilme ile yeniden dene — geçici hatalar genellikle çözülür
2. Hatayı sakla — desenlerden öğren
3. Zarif degrade — tüm oturumu çökertme
4. İnsanı bilgilendir — çözemediğiniz hatalar için

HTTP Hata İşleme

Üstel geri çekilme ile yeniden deneme

import time
import requests

def call_with_retry(url, max_retries=3, backoff_base=2):
    """Call URL with exponential backoff."""
    for attempt in range(max_retries):
        try:
            r = requests.get(url, headers={"Authorization": f"Bearer {KEY}"})
            
            # Success
            if r.status_code < 400:
                return r
            
            # Rate limited — wait and retry
            if r.status_code == 429:
                wait = int(r.headers.get("Retry-After", 60))
                print(f"Rate limited. Waiting {wait}s...")
                time.sleep(wait)
                continue
            
            # Server error — retry with backoff
            if r.status_code >= 500:
                wait = backoff_base ** attempt
                print(f"Server error {r.status_code}. Retrying in {wait}s...")
                time.sleep(wait)
                continue
            
            # Client error — don't retry
            if 400 <= r.status_code < 500:
                raise ClientError(f"{r.status_code}: {r.text}")
        
        except requests.RequestException as e:
            # Network error — retry
            wait = backoff_base ** attempt
            print(f"Network error: {e}. Retrying in {wait}s...")
            time.sleep(wait)
    
    raise MaxRetriesError(f"Failed after {max_retries} retries")

Kimlik doğrulama hata yönetimi

def safe_call(url):
    """Call with auth error handling."""
    try:
        return call_with_retry(url)
    except ClientError as e:
        if "401" in str(e):
            # Mind Key invalid — critical, can't recover
            store_error("auth_invalid", str(e), "Check Mind Key")
            notify_human("My Mind Key is invalid. Please update.")
            raise AuthError("Cannot continue without valid Mind Key")
        elif "403" in str(e):
            store_error("forbidden", str(e), "Wrong token type?")
            raise ForbiddenError(str(e))
        elif "404" in str(e):
            # Path doesn't exist — don't retry
            store_error("not_found", str(e), "Check endpoint path")
            raise NotFoundError(str(e))
        else:
            raise

Hataları Bellek Olarak Saklama

Hatalar oluştuğunda, gelecek oturumların öğrenebilmesi için saklayın:

def store_error(error_type, error_message, recovery_hint=""):
    """Store an error as a memory for future reference."""
    requests.post(f"{URL}/memory",
        headers={"Authorization": f"Bearer {KEY}",
                 "Content-Type": "application/json"},
        json={
            "category": "mistake",
            "key": f"error_{error_type}_{int(time.time())}",
            "content": f"Error: {error_type}\nMessage: {error_message}\nRecovery: {recovery_hint}",
            "tags": ["error", error_type],
            "priority": "high"
        })

# Example
try:
    deploy()
except DeployError as e:
    store_error("deploy_failed", str(e), 
                "Check CI logs, verify Docker image exists")
    raise

Yaygın Hata Senaryoları

Senaryo 1: Mind Key Geçersiz

# Detection: 401 on every call
# Recovery: Cannot recover — need human intervention

def handle_invalid_mind_key():
    store_error("mind_key_invalid", 
                "All API calls returning 401",
                "Mind Key may be revoked. Need new key.")
    
    # Notify human via chat (if possible)
    try:
        reply("⚠️ My Mind Key is invalid. I cannot access memories. "
              "Please check and update SYNAPSE_MIND_KEY.")
    except:
        pass  # Chat may also fail with bad key
    
    # Exit gracefully
    raise CriticalError("Cannot continue without valid Mind Key")

Senaryo 2: Ağ Hatası

# Detection: ConnectionError, Timeout
# Recovery: Retry with backoff, then degrade

def handle_network_error(url, retry=3):
    for attempt in range(retry):
        try:
            return requests.get(url, timeout=10)
        except (requests.ConnectionError, requests.Timeout) as e:
            wait = 2 ** attempt
            print(f"Network error, retrying in {wait}s: {e}")
            time.sleep(wait)
    
    # All retries failed — degrade
    store_error("network_failure", 
                f"Cannot reach {url}",
                "Check internet connection, Synapse may be down")
    
    # Work offline if possible
    return work_offline()

Senaryo 3: Hız Sınırına Ulaşıldı

# Detection: 429 with Retry-After header
# Recovery: Wait and retry, or switch to header auth

def handle_rate_limit(response):
    retry_after = int(response.headers.get("Retry-After", 60))
    print(f"Rate limited. Waiting {retry_after}s...")
    time.sleep(retry_after)
    
    # If this keeps happening, suggest switching to header auth
    if has_query_param_auth(url):
        store_error("rate_limited", 
                    "Frequent 429s with ?key= auth",
                    "Switch to Authorization: Bearer header (no rate limit)")

Senaryo 4: Sunucu Hatası (5xx)

# Detection: 500, 502, 503
# Recovery: Retry with backoff, check /health

def handle_server_error(url):
    # Check if server is up
    health = requests.get(f"{URL}/health")
    if health.status_code != 200:
        store_error("server_down", 
                    "Synapse health check failing",
                    "Wait for server recovery")
        raise ServerDownError()
    
    # Retry with backoff
    return call_with_retry(url, max_retries=5)

Senaryo 5: Araç Çağrısı Başarısız

# Detection: Tool returns error content
# Recovery: Try alternative approach, store failure

def call_tool_safely(tool_name, args, alternatives=None):
    try:
        result = call_tool(tool_name, args)
        if result.get("isError"):
            raise ToolError(result["content"])
        return result
    except ToolError as e:
        store_error(f"tool_{tool_name}_failed",
                    f"Args: {args}\nError: {e}",
                    f"Try: {alternatives or 'no alternatives'}")
        
        # Try alternatives
        if alternatives:
            for alt in alternatives:
                try:
                    return call_tool(alt, args)
                except:
                    continue
        
        raise

Desen: Devre Kesici

Yinelenen başarısızlıklar için, geçici olarak denemeyi bırakın:

class CircuitBreaker:
    def __init__(self, threshold=5, reset_time=300):
        self.failures = 0
        self.threshold = threshold
        self.reset_time = reset_time
        self.last_failure = 0
    
    def call(self, fn, *args, **kwargs):
        if self.failures >= self.threshold:
            if time.time() - self.last_failure < self.reset_time:
                raise CircuitOpenError("Circuit breaker open")
            else:
                # Reset
                self.failures = 0
        
        try:
            result = fn(*args, **kwargs)
            self.failures = 0  # Reset on success
            return result
        except:
            self.failures += 1
            self.last_failure = time.time()
            raise

# Usage
breaker = CircuitBreaker(threshold=5)
try:
    result = breaker.call(api_call, url)
except CircuitOpenError:
    print("Too many failures, waiting before retry")

En İyi Uygulamalar

Sonraki Adımlar

• Errors API Referansı
• Oturum Başlangıç Deseni
• Kendinden İyileşen Testler