Agent Guardrail'lar: Otonom Yapay Zeka Sistemlerini Güvenli Tutmanın Yolu

Guardrail'sız Agent Neden Yayına Çıkmamalı

Otonom AI agent'lar, klasik yazılımdan tamamen farklı bir dünya. Belirli kod yollarını izleyen geleneksel uygulamaların aksine, agent'lar karar veriyor, tool'lar çağırıyor ve her zaman öngöremeyeceğiniz çıktılar üretiyor. Bu öngörülemezlik onları güçlü kılıyor — ama aynı zamanda kontrolsüz bırakıldığında tehlikeli de yapıyor. Veritabanına, e-posta API'sine ve ödeme sistemine erişimi olan bir agent düşünün. Bir talimatı yanlış anlarsa ya da prompt injection saldırısına uğrarsa, saniyeler içinde ciddi hasar verebilir.

Guardrail'lar, agent davranışını kabul edilebilir sınırlar içinde tutan güvenlik mekanizmaları. Input, output, aksiyon ve sistem seviyesinde çalışıyorlar. Böylece alttaki model hata yapsa bile sistem bir bütün olarak güvenli şekilde çöküyor. Bir köprü düşünün: sadece çeliğin dayanıklılığına güvenmezsiniz. Yedek destekler, yük sensörleri ve acil kapatma mekanizmaları eklersiniz. AI agent'lar da aynı derinlemesine savunma yaklaşımına ihtiyaç duyuyor.

Şunu net söyleyeyim: guardrail'sız agent deploy ederseniz, bir şeylerin ters gidip gitmeyeceği değil, ne zaman gideceği sorusu oluyor. Production'da prompt injection ile veri sızıntısından, dakikalar içinde binlerce dolarlık fatura çıkaran kontrolsüz API çağrılarına kadar pek çok olay gördüm. Guardrail eklemenin maliyeti, bu arızaları temizlemenin maliyetinin yanında hiç kalır.

Guardrail'lar agent'ın yapabileceklerini kısıtlamaz — tam tersine, onları mümkün kılar. Güvenli çalışacağına güvenemediğiniz bir agent'a asla gerçek otonomi veremezsiniz.

Input Validation ve Prompt Shield'lar

İlk savunma hattınız input validation. Agent'ın context'ine giren her mesaj, belge veya veri kaynağı; kötü niyetli içerik, bozuk talimatlar ve policy ihlalleri açısından taranmalı. Prompt injection — saldırganın zararsız görünen içeriğin içine gizli talimatlar yerleştirmesi — hâlâ en büyük tehditlerden biri. Dışarıdan belge çeken bir RAG agent çalıştırıyorsanız özellikle dikkat edin: corpus'taki herhangi bir belge enjekte edilmiş yönergeler taşıyor olabilir.

Prompt shield'lar, gelen içeriği agent'ın muhakeme döngüsüne ulaşmadan önce sınıflandırarak çalışır. Modern yaklaşımlar, kural tabanlı pattern matching'i injection girişimlerini yakalamak için eğitilmiş hafif classifier model'lerle birleştirir. Kural tabanlı katman bilinen kalıpları yakalar: "ignore previous instructions" gibi ifadeler, base64 veya unicode encoding hileleri ve delimiter manipülasyonu. Classifier katmanı ise statik kuralları atlayan yeni saldırıları, input'un semantik amacını değerlendirerek ele alır.

Input'lara yapısal kısıtlamalar da uygulamanız gerekir. Agent bir kullanıcı sorgusu bekliyorsa, uzunluk limitleri, karakter encoding geçerliliği ve kontrol karakterlerinin yokluğu kontrol edilmeli. Tool sonuçları için schema validation yaparak harici verilerin beklenen tiplere ve aralıklara uyduğundan emin olun. Bu temel kontroller sıkıcı gelebilir ama şaşırtıcı sayıda production hatasını önlüyor.

import re
from dataclasses import dataclass
from enum import Enum

class ThreatLevel(Enum):
    SAFE = "safe"
    SUSPICIOUS = "suspicious"
    BLOCKED = "blocked"

@dataclass
class ValidationResult:
    threat_level: ThreatLevel
    reasons: list[str]
    sanitized_input: str | None

class PromptShield:
    """Ajan sistemleri için çok katmanlı girdi doğrulama."""

    # Bilinen enjeksiyon kalıpları
    INJECTION_PATTERNS = [
        r"ignore\s+(all\s+)?previous\s+instructions",
        r"you\s+are\s+now\s+(a|an)\s+",
        r"system\s*:\s*",
        r"\[INST\]|\[/INST\]|<<SYS>>",
        r"act\s+as\s+(if|though)?\s+",
    ]

    # İzin verilen maksimum girdi uzunluğu (karakter)
    MAX_INPUT_LENGTH = 10_000

    def __init__(self, classifier=None):
        self._patterns = [re.compile(p, re.IGNORECASE) for p in self.INJECTION_PATTERNS]
        self._classifier = classifier  # İsteğe bağlı ML tabanlı sınıflandırıcı

    def validate(self, user_input: str) -> ValidationResult:
        """Kullanıcı girdisini birden fazla savunma katmanı üzerinden doğrulayın."""
        reasons: list[str] = []

        # Katman 1: Uzunluk ve kodlama kontrolleri
        if len(user_input) > self.MAX_INPUT_LENGTH:
            return ValidationResult(
                threat_level=ThreatLevel.BLOCKED,
                reasons=["Girdi izin verilen maksimum uzunluğu aşıyor"],
                sanitized_input=None,
            )

        # Katman 2: Kalıp tabanlı enjeksiyon tespiti
        for pattern in self._patterns:
            if pattern.search(user_input):
                reasons.append(f"Enjeksiyon kalıbı eşleşti: {pattern.pattern}")

        # Katman 3: Unicode ve kodlama saldırısı tespiti
        decoded = self._check_encoding_attacks(user_input)
        if decoded != user_input:
            reasons.append("Kodlanmış içerik tespit edildi (olası gizleme)")

        # Katman 4: ML sınıflandırıcı (mevcutsa)
        if self._classifier and self._classifier.is_injection(user_input):
            reasons.append("ML sınıflandırıcı potansiyel enjeksiyon olarak işaretledi")

        # Tehdit seviyesini belirle
        if len(reasons) >= 2:
            return ValidationResult(ThreatLevel.BLOCKED, reasons, None)
        elif len(reasons) == 1:
            return ValidationResult(ThreatLevel.SUSPICIOUS, reasons, decoded)
        return ValidationResult(ThreatLevel.SAFE, [], user_input)

    def _check_encoding_attacks(self, text: str) -> str:
        """Yaygın kodlama tabanlı saldırıları tespit edin ve çözün."""
        import base64
        # Base64 kodlu segmentleri kontrol et
        b64_pattern = re.compile(r"[A-Za-z0-9+/]{20,}={0,2}")
        for match in b64_pattern.finditer(text):
            try:
                decoded = base64.b64decode(match.group()).decode("utf-8")
                # Çözülen metin enjeksiyon kalıpları içeriyorsa işaretle
                for pattern in self._patterns:
                    if pattern.search(decoded):
                        return decoded
            except Exception:
                pass
        return text

Output Filtering ve İçerik Moderasyonu

Input validation kusursuz olsa bile agent'ın ürettiği çıktıları da denetlemeniz şart. Dil modelleri policy'leri ihlal eden, hassas bilgileri sızdıran veya kullanıcıları yanıltan halüsinasyon içerikler üretebilir. Output filtering, herhangi bir yanıt kullanıcıya ulaşmadan veya downstream bir aksiyonu tetiklemeden önceki son kontrol noktası.

Content moderation classifier'ları agent çıktılarını birkaç boyutta değerlendirir: toksisite, kişisel bilgiler (PII), gizli veri kalıpları ve alana özgü policy ihlalleri. PII tespiti için named entity recognition'ı yapısal veriler için regex pattern'leriyle (kredi kartı numaraları, TC kimlik numaraları, e-posta adresleri) birleştirmek sağlam bir kapsam sağlar. Büyük mimari soru şu: tüm yanıtı mı engelleyeceksiniz, yoksa sadece sorunlu kısımları mı çıkaracaksınız? Redaction kullanışlı kısımları korur ama eksik temizleme riski taşır. Engelleme daha güvenli ama kullanıcı deneyimini bozar.

Halüsinasyon tespiti daha karmaşık bir konu. Factual grounding — agent'ın iddialarının gerçekten kaynak belgelerce desteklenip desteklenmediğini kontrol etmek — çıktı ifadelerini alınan context ile karşılaştırmayı gerektirir. Teknikler basit string overlap metriklerinden, kaynak metnin her iddiayı mantıksal olarak destekleyip desteklemediğini yargılayan entailment classifier'lara kadar uzanır. Sağlık veya hukuk gibi yüksek riskli alanlarda standart uygulama, her olgusal iddia için açık kaynak atıfı istemek ve atıfsız iddiaları insan incelemesine göndermek.

Format kısıtlamalarını da unutmayın. Agent'ınız JSON döndürmesi gerekiyorsa, çıktıyı downstream'e göndermeden önce schema'ya göre validate edin. Kod üretiyorsa, SQL injection veya command injection pattern'leri gibi bariz güvenlik açıklarını yakalamak için static analysis çalıştırın. Bu yapısal kontroller, content classifier'larının yaptığı semantik kontrolleri tamamlar.

Aksiyon Sınırları ve İzin Sistemleri

Agent'ınız gerçek dünyada sonuçlar doğuran işler yapabiliyorsa — e-posta gönderme, veritabanı değiştirme, ödeme işleme — izin sistemi en kritik guardrail'ınız oluyor. En az yetki ilkesi AI agent'lar için insan kullanıcılardan bile daha önemli. Çünkü agent'lar makine hızında hareket eder, bir insanın doğal olarak göstereceği tereddüt veya muhakeme olmadan.

İyi tasarlanmış bir izin sistemi her aksiyon için üç boyut tanımlar: kapsam, hız ve onay seviyesi. Kapsam, aksiyonun neyi etkileyebileceğini belirler — bir veritabanı tool'u belirli tablolarda sadece okuma erişimiyle sınırlandırılabilir. Rate limit'ler bir zaman penceresi içindeki aksiyon sayısını veya toplam değerini sınırlar — mesela e-posta tool'u saatte en fazla on mesaj gönderebilir. Onay seviyesi aksiyonun otonom çalışıp çalışamayacağını, async incelemeye mi gideceğini yoksa anlık insan onayı mı gerektireceğini belirler.

Production'da en iyi çalışan model kademeli yapı. Veri okuma veya hesaplama gibi düşük riskli aksiyonlar insan gözetimi olmadan çalışır. Bildirim gönderme veya kritik olmayan kayıtları güncelleme gibi orta riskli aksiyonlar loglama ve sonradan incelemeyle devam eder. Yüksek riskli aksiyonlar — finansal işlemler, veri silme, yeni alıcılara e-posta — çalışmadan önce açık insan onayı gerektirir. Önemli olan şu: hangi aksiyonun hangi kademeye düşeceğini bir güvenlik incelemesi belirlemeli, agent'ın kendisi değil.

from enum import Enum
from dataclasses import dataclass
from typing import Any, Callable, Awaitable
import time

class ApprovalLevel(Enum):
    AUTO = "auto"           # İnsan gözetimi olmadan yürüt
    LOG = "log"             # Yürüt ve inceleme için kaydet
    CONFIRM = "confirm"     # Yürütmeden önce insan onayı gerektirir

@dataclass
class ActionPolicy:
    approval_level: ApprovalLevel
    max_calls_per_hour: int
    max_value_per_call: float | None = None  # Finansal eylemler için
    allowed_scopes: list[str] | None = None  # Hedef kaynakları kısıtla

class ActionGuardrail:
    """Ajan eylemleri üzerinde izin sınırlarını uygulayın."""

    def __init__(self, policies: dict[str, ActionPolicy]):
        self._policies = policies
        self._call_log: dict[str, list[float]] = {}  # eylem -> zaman damgaları

    async def execute(
        self,
        action_name: str,
        parameters: dict[str, Any],
        executor: Callable[..., Awaitable[Any]],
        approval_callback: Callable[[str, dict], Awaitable[bool]] | None = None,
    ) -> dict[str, Any]:
        """Tam koruma bariyeri uygulamasıyla bir eylemi yürütün."""
        policy = self._policies.get(action_name)
        if policy is None:
            # Varsayılan olarak bilinmeyen eylemleri reddet
            return {"status": "denied", "reason": "Bu eylem için tanımlanmış politika yok"}

        # Kapsam kısıtlamalarını kontrol et
        if policy.allowed_scopes is not None:
            target = parameters.get("target", "")
            if not any(target.startswith(scope) for scope in policy.allowed_scopes):
                return {"status": "denied", "reason": f"Hedef '{target}' izin verilen kapsamlar dışında"}

        # Hız sınırlarını kontrol et
        now = time.time()
        recent_calls = [
            t for t in self._call_log.get(action_name, [])
            if now - t < 3600  # Son bir saat içinde
        ]
        if len(recent_calls) >= policy.max_calls_per_hour:
            return {"status": "denied", "reason": "Hız sınırı aşıldı"}

        # Finansal eylemler için değer sınırlarını kontrol et
        if policy.max_value_per_call is not None:
            value = parameters.get("amount", 0)
            if value > policy.max_value_per_call:
                return {"status": "denied", "reason": f"Tutar {value} sınırı aşıyor {policy.max_value_per_call}"}

        # Onay seviyesini kontrol et
        if policy.approval_level == ApprovalLevel.CONFIRM:
            if approval_callback is None:
                return {"status": "denied", "reason": "İnsan onayı gerekli ancak geri çağırma sağlanmadı"}
            approved = await approval_callback(action_name, parameters)
            if not approved:
                return {"status": "denied", "reason": "İnsan inceleyici eylemi reddetti"}

        # Eylemi yürüt
        try:
            result = await executor(**parameters)
            # Çağrıyı kaydet
            self._call_log.setdefault(action_name, []).append(now)
            return {"status": "success", "result": result}
        except Exception as e:
            return {"status": "error", "reason": str(e)}

# Her kullanılabilir eylem için politikaları tanımla
policies = {
    "read_database": ActionPolicy(
        approval_level=ApprovalLevel.AUTO,
        max_calls_per_hour=100,
        allowed_scopes=["users.profile", "products", "orders.status"],
    ),
    "send_email": ActionPolicy(
        approval_level=ApprovalLevel.LOG,
        max_calls_per_hour=10,
    ),
    "process_payment": ActionPolicy(
        approval_level=ApprovalLevel.CONFIRM,
        max_calls_per_hour=5,
        max_value_per_call=500.00,
    ),
    "delete_record": ActionPolicy(
        approval_level=ApprovalLevel.CONFIRM,
        max_calls_per_hour=3,
        allowed_scopes=["drafts", "temp_files"],
    ),
}

Monitoring ve Circuit Breaker'lar

Runtime monitoring, guardrail'larınızı statik kurallardan adaptif bir güvenlik sistemine dönüştürür. Dağıtık sistemlerden ödünç alınan circuit breaker pattern'i, anormal davranış tespit ettiğinde agent operasyonlarını otomatik olarak durdurur. Fikir basit: temel metrikleri gerçek zamanlı takip edin ve herhangi biri eşiği aşarsa breaker'ı devreye sokup agent'ı hasar birikmeden durdurun.

İzlemeniz gereken metrikler: token tüketim hızı, tool çağrı sıklığı, hata oranı, döngü tespiti (aynı aksiyonun tekrar tekrar yapılması) ve toplam çalışma süresi. Normal bir görevi tamamlayan sağlıklı bir agent, bu metriklerde öngörülebilir kalıplar gösterir. Sapmalar gördüğünüzde — tool çağrılarında ani artış veya ardı ardına aynı başarısız aksiyonlar — agent'ın bir döngüye girdiğini, beklenmedik input'la karıştığını veya enjekte edilmiş talimatlarla manipüle edildiğini anlarsınız.

Pratikte circuit breaker'ları birden fazla seviyede kurmanız gerekir. İstek başına breaker, tek bir agent çağrısının tüketebileceği toplam kaynağı sınırlar. Kullanıcı başına breaker, tek bir kullanıcının — veya ele geçirilmiş bir hesabın — sistem kaynaklarını tekelleştirmesini önler. Sistem genelinde breaker, tüm kullanıcılardaki toplam hata oranı kabul edilebilir eşiği aştığında devreye girer. Bu, model bozulması veya upstream API kesintisi gibi sistemik bir soruna işaret eder.

Observability altyapınız tam agent trajectory'lerini yakalamalı — her prompt, argümanları ve sonuçlarıyla her tool çağrısı, her çıktı — olay sonrası analizi destekleyen structured log'larda. Bir şeyler ters gittiğinde, arızaya yol açan tam olay dizisini tekrar oynatabilmek, kök nedenleri bulmak ve guardrail'ları güçlendirmek için çok değerli. PII alanlarını depolamadan önce hash'lemek gibi gizliliği koruyan teknikler kullanarak veri sorumluluğu yaratmadan detaylı observability elde edin.

Framework'ler: NeMo Guardrails, Guardrails AI ve Ötesi

Guardrail implementasyonunu kolaylaştırmak için her biri farklı felsefesiyle birkaç açık kaynak framework ortaya çıktı. NVIDIA'nın NeMo Guardrails'i diyalog yönetimi yaklaşımı benimsiyor. Colang adlı özel bir scripting dili kullanarak konuşma rail'larını tanımlıyor — hangi konuşma akışlarına izin verildiğini ve hangilerinin yasak olduğunu belirliyorsunuz. LangChain ile entegre oluyor ve input/output rail'lar, konu kısıtlamaları ve bilgi tabanlarına karşı fact-checking destekliyor. Colang soyutlaması mühendis olmayanlara erişilebilir kılıyor, ama karmaşık programatik guardrail'lar gerektiğinde kısıtlayıcı hissedilebilir.

Guardrails AI (guardrails-ai kütüphanesi) farklı bir açıdan yaklaşıyor: structured output validation. Beklenen çıktı schema'larını, validator'ları ve düzeltici aksiyonları tanımlamak için "Guards" adlı XML tabanlı spec'ler kullanır. Modelin çıktısı validation'dan geçemezse, framework otomatik olarak neyin yanlış gittiğine dair spesifik geri bildirimle yeniden prompt gönderir — buna "re-asking" deniyor. Agent çıktılarının API response format'ları veya veritabanı schema'ları gibi katı veri kontratlarına uyması gerektiğinde özellikle iyi çalışır.

Kendi guardrail stack'inizi oluşturuyorsanız, modüler bir middleware mimarisi en fazla esnekliği verir. Her guardrail, agent'ın input veya output'unu alan, kontrolünü yapan ve veriyi ya ileten, ya değiştiren ya da exception fırlatan bağımsız bir middleware fonksiyonu olarak yaşar. Bu yaklaşım guardrail'ları birleştirmenize, sıralamanızı değiştirmenize ve bağımsız test etmenize olanak tanır. Ayrıca production'da farklı konfigürasyonları A/B test ederek güvenlik ile kullanıcı deneyimi arasındaki ideal noktayı bulmanızı kolaylaştırır.

Önemli Çıkarımlar

1. Otonom agent'ları production'da guardrail'sız deploy edemezsiniz. Guardrail'lar olmadan hatalar sadece olası değil — kaçınılmaz ve potansiyel olarak felaket.
2. Input validation; prompt injection ve kötü niyetli içeriğe karşı pattern matching, encoding tespiti ve ML classifier'ları birleştirmeli.
3. Output filtering toksisite, PII sızıntısı, halüsinasyon tespiti ve format validation'ı kapsamalı. Sadece system prompt talimatlarına güvenmek yetmez.
4. Aksiyon izin sistemleri kapsam kısıtlamaları, rate limit'ler ve kademeli onay seviyeleri uygulamalı. Yüksek riskli aksiyonlar her zaman insan onayı gerektirmeli.
5. Dağıtık sistemlerden gelen circuit breaker'lar, anormal davranan agent'ları otomatik durdurarak runtime güvenliği sağlar.
6. NeMo Guardrails ve Guardrails AI gibi framework'ler implementasyonu hızlandırır, ama production sistemleri çoğu zaman özel konfigürasyonlar oluşturmanıza izin veren modüler middleware yaklaşımından daha çok fayda görür.
7. Guardrail'ları gerçek production trafiğine göre iterate etmek, tüm hata senaryolarını önceden tahmin etmeye çalışmaktan çok daha iyi sonuç verir.