AI Guardrail'lerin Koda Entegrasyonu

LLM güvenliği başlangıçta modelin kendisine bırakılmıştı. Sağlayıcılar modelleri alignment (RLHF vb.) ile eğitiyor, uygulama tarafında da system prompt’a kurallar yazılıyordu. Jailbreak’ler ve özellikle prompt injection bu yaklaşımın sınırını gösterdi, çünkü model talimat ile işlediği veriyi aynı context içinde ayırt edemiyor. Tool kullanımı ve agent’larla birlikte modele giren güvenilmeyen içerik ve modelin tetikleyebildiği aksiyonlar arttıkça kontrolü modelin dışına, çalışma anında input ve output’u denetleyen ayrı bir katmana taşımak gerekti. Bu katmana guardrail diyoruz.

Bu yazıda guardrail’in uygulama akışında nereye yerleştiğini, hangi adımda hangi modellerin çalıştığını ve entegrasyon yöntemlerini Lakera Guard üzerinden, kod örnekleriyle anlatıyoruz. Yöntemlerin çoğu diğer guardrail ürünlerine de uygulanabilir.

Guardrail nedir?

Guardrail, LLM’in önünde ve arkasında çalışan, metni sınıflandırıp her tehdit kategorisi için bir skor döndüren bağımsız bir servistir.

• Generative değil, discriminative modellerden oluşur. Metni girdi olarak alır ama metin üretmez, çıktı olarak kategori ve güven skoru verir. Tipik olarak encoder-based classifier’lar, NER modelleri, regex ve checksum doğrulaması birlikte kullanılır.
• LLM’in context’i dışında çalışır. Karar, korunan modelin talimatlara uyma davranışına bağlı değildir. System prompt’a “kötü niyetli istekleri reddet” yazmak, saldırganın hedef aldığı mekanizmanın kendisine güvenmek demektir.
• Düşük latency için tasarlanır. Her istekte senkron çağrılabilmesi için tek forward pass ile sonuç veren küçük modeller kullanılır. Büyük bir LLM’i hakem olarak kullanmak (LLM-as-a-judge) bu latency bütçesine genelde sığmaz.
• Policy decision point’tir, enforcement point değildir. Guardrail skorları policy’deki threshold’larla karşılaştırıp flagged kararını döner. Engelleme, maskeleme veya sadece loglama aksiyonu uygulamanın sorumluluğundadır.

Kontrol noktaları

Basit bir chatbot’ta kullanıcı input’u ve model output’u olmak üzere iki kontrol noktası vardır. Tool kullanan agent mimarilerinde modelin tetiklediği tool call’lar ve tool’lardan dönen içerik de kontrol noktasına dönüşür.

flowchart LR
    K([Kullanıcı]) -->|"① input"| APP[Uygulama]
    APP <-->|prompt / response| LLM[LLM]
    APP -->|"② tool call"| TOOL[Tools / API / MCP]
    TOOL -->|"③ tool result"| APP
    APP -->|"④ output"| K

    G{{Guardrail}}
    APP -. her noktada .-> G

Genel kural olarak modelin context’ine giren ve modelin ürettiği her içerik bir kontrol noktasıdır.

Adımlar ve devreye giren modeller

Guardrail tek bir modelden oluşmaz. Her tehdit kategorisi için ayrı bir detector vardır ve hangi detector’ların hangi threshold ile çalışacağını policy belirler. Lakera Guard’daki detector ailelerini aşağıdaki tabloda bulabilirsiniz.

Bir Guard çağrısı kabaca şu akışla işlenir.

flowchart TB
    REQ["POST /v2/guard<br/>messages + project_id"] --> POL[Projeye bağlı policy'yi yükle]
    POL --> CH[Uzun metni chunk'lara böl<br/>paralel işle]
    CH --> D1[Prompt Defense<br/>classifier]
    CH --> D2[PII<br/>ML + validation + pattern]
    CH --> D3[Moderation<br/>classifier'lar]
    CH --> D4[Link kontrolü<br/>domain listeleri]
    CH --> D5[Custom kurallar]
    D1 & D2 & D3 & D4 & D5 --> TH[Skorları policy'deki<br/>threshold'larla karşılaştır]
    TH --> RES["flagged: true/false<br/>+ breakdown / payload"]

Burada dikkat edilmesi gereken iki nokta var.

• Mesajın rolü hangi detector’ların çalışacağını etkiler. Policy’de detector’lar input (user, tool) ve output (assistant) için ayrı tanımlanabilir. Örneğin content moderation sadece output’ta, PII her iki yönde çalışabilir. system mesajları context olarak kullanılır.
• Hassasiyet policy’de tanımlanır. L1 (en düşük false positive) ile L4 (en düşük false negative) arasında seçilir ve kod değişikliği gerektirmez.

Guard API çağrısı

Aşağıdaki tüm entegrasyon yöntemleri aynı çağrıyı kullanır.

import os, requests

GUARD_URL = os.getenv("GUARD_URL", "https://api.lakera.ai/v2/guard")
HEADERS = {"Authorization": f"Bearer {os.getenv('LAKERA_GUARD_API_KEY')}"}
PROJECT_ID = "project-ornek-asistan"

def guard(messages, **opts):
    r = requests.post(GUARD_URL, headers=HEADERS, timeout=3, json={
        "messages": messages,
        "project_id": PROJECT_ID,
        "breakdown": True,   # detector bazında sonuç
        **opts,
    })
    r.raise_for_status()
    return r.json()

Guard’ın döndüğü yanıt şuna benzer.

{
  "flagged": true,
  "metadata": { "request_uuid": "01a15f8e-b7c9-7bad-a351-09d380117fa0" },
  "breakdown": [
    { "detector_type": "prompt_attack", "detected": true,  "result": "l1_confident" },
    { "detector_type": "pii/email",     "detected": false, "result": "l5_unlikely" }
  ]
}

• project_id uygulanacak policy’yi belirler. Uygulama ve ortam başına ayrı proje tanımlamak policy’leri birbirinden izole eder.
• flagged policy kararıdır, breakdown ise her detector’ün sonucunu ve güven seviyesini içerir.
• request_uuid isteğin Guard tarafındaki ID’sidir. Uygulama loguna yazılırsa Guard logları ve SIEM kayıtlarıyla korelasyon kurulabilir.
• System prompt ayrı bir system mesajı olarak gönderilmelidir. Kullanıcı içeriğini şablonlara gömmek yerine modelin gördüğü ham mesajları göndermek en doğru sonucu verir.

Entegrasyon yöntemleri

1. Asenkron izleme (Detect modu)

İlk aşamada Guard çağrısı critical path’in dışında, asenkron yapılır ve sonuç loglanır. Proje Detect modunda çalıştığı için detector sonuçları breakdown‘da döner ama flagged her zaman false olur. Kullanıcı akışına latency eklenmez.

sequenceDiagram
    participant K as Kullanıcı
    participant U as Uygulama
    participant G as Guard
    participant L as LLM
    K->>U: Mesaj
    U-)G: input kontrolü (async)
    U->>L: Mesaj
    L-->>U: Yanıt
    U-)G: output kontrolü (async)
    U-->>K: Yanıt
    G--)U: sonuç → log / SIEM

from concurrent.futures import ThreadPoolExecutor
pool = ThreadPoolExecutor(max_workers=8)

def guard_log(stage, messages):
    try:
        res = guard(messages)
        levels = {b["detector_type"]: b["result"] for b in res["breakdown"]}
        log.info("guard stage=%s uuid=%s results=%s", stage, res["metadata"]["request_uuid"], levels)
    except Exception as e:
        log.warning("guard stage=%s hata=%s", stage, e)

def chat(user_text):
    msgs = [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_text}]
    pool.submit(guard_log, "input", msgs)
    answer = call_llm(msgs)
    pool.submit(guard_log, "output", msgs + [{"role": "assistant", "content": answer}])
    return answer

Bu aşamada toplanan veriyle false positive’ler analiz edilir ve threshold’lar kalibre edilir. Lakera dokümantasyonuna göre kalibrasyon için yaklaşık 6.000 temsili örnek genelde yeterlidir.

2. Engelleme (Enforce modu)

Kalibrasyondan sonra aynı çağrılar senkron yapılır ve proje Enforce moduna alınır. Kodda değişen tek şey sonuca göre akışın kesilmesidir.

sequenceDiagram
    participant K as Kullanıcı
    participant U as Uygulama
    participant G as Guard
    participant L as LLM
    K->>U: Mesaj
    U->>G: input kontrolü
    alt flagged
        G-->>U: true
        U-->>K: Ret mesajı
    else temiz
        G-->>U: false
        U->>L: Mesaj
        L-->>U: Yanıt
        U->>G: output kontrolü
        G-->>U: true / false
        U-->>K: Yanıt veya ret mesajı
    end

BLOCKED = "Bu konuda yardımcı olamıyorum."

def chat(user_text):
    msgs = [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_text}]
    if guard(msgs)["flagged"]:
        return BLOCKED
    answer = call_llm(msgs)
    if guard(msgs + [{"role": "assistant", "content": answer}])["flagged"]:
        return BLOCKED
    return answer

Bu aşamada iki tasarım kararı verilmesi gerekir.

• Ret mesajı tespit edilen kategoriyi açık etmemelidir. Saldırgana hangi payload’ın yakalandığı bilgisini vermemek için genel bir mesaj kullanılır.
• Guard’a erişilemediğinde uygulamanın nasıl davranacağı tanımlanmalıdır. Fail-open erişilebilirliği, fail-closed güvenliği önceliklendirir. Timeout değeri ve seçilen davranış uygulamanın risk profiline göre belirlenmelidir.

3. Input kontrolünü LLM ile paralel çalıştırma

Guard yanıt süresi tipik olarak LLM inference süresinden çok daha kısadır. Input kontrolü LLM çağrısıyla paralel başlatılıp yanıt kullanıcıya dönmeden önce sonucu beklenirse, eklenen latency pratikte sıfıra yaklaşır.

import asyncio

async def chat(user_text):
    msgs = [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_text}]
    guard_task = asyncio.create_task(guard_async(msgs))
    llm_task = asyncio.create_task(call_llm_async(msgs))

    if (await guard_task)["flagged"]:
        llm_task.cancel()
        return BLOCKED
    answer = await llm_task
    if (await guard_async(msgs + [{"role": "assistant", "content": answer}]))["flagged"]:
        return BLOCKED
    return answer

Bu yöntemde flag’lenen input da LLM’e ulaşmış olur. PII’nin üçüncü taraf bir LLM sağlayıcısına gitmemesi gerekiyorsa input kontrolü sıralı yapılmalıdır.

4. Tek çağrıyla etkileşim taraması

Lakera’nın önerdiği minimum entegrasyon, LLM yanıtından sonra kullanıcı mesajı ve yanıtı tek çağrıda göndermektir. Guard son etkileşimi (son user mesajı ve sonrasını) tarar, önceki mesajları context olarak kullanır.

def chat(history, user_text):
    msgs = history + [{"role": "user", "content": user_text}]
    answer = call_llm(msgs)
    if guard(msgs + [{"role": "assistant", "content": answer}])["flagged"]:
        return BLOCKED
    return answer

Avantajı tek çağrı ve tam context, dezavantajı ise input’un LLM’e ulaştıktan sonra kontrol edilmesidir. Self-hosted kurulumlarda son etkileşim davranışı sürüme ve yapılandırmaya göre farklılık gösterebileceği için kullanıcı mesajının da tarandığı bir test isteğiyle doğrulanmalıdır.

5. Agent’lar ve tool call’lar

Agent mimarisinde Guard agent loop’un her adımında çağrılır. Kullanıcı isteği, agent’ın ürettiği tool call (çalıştırılmadan önce), tool sonucu ve nihai yanıt ayrı ayrı kontrol edilir. Tool tanımları da (örneğin MCP handshake sırasında) tool poisoning’e karşı taranabilir.

sequenceDiagram
    participant U as Uygulama / Agent
    participant G as Guard
    participant L as LLM
    participant T as Tool
    U->>G: kullanıcı mesajı
    U->>L: mesaj + tool tanımları
    L-->>U: tool_call (ör. export_records)
    U->>G: assistant tool_call + tools
    G-->>U: tehlikeli / yetki dışı mı?
    U->>T: tool call'u çalıştır (temizse)
    T-->>U: tool sonucu
    U->>G: tool sonucu (indirect injection?)
    U->>L: tool sonucu
    L-->>U: nihai yanıt
    U->>G: assistant yanıtı

def run_tool_step(msgs, tool_call, tools):
    # 1) Çalıştırmadan önce tool call'u kontrol et
    step = msgs + [{"role": "assistant", "content": "", "tool_calls": [tool_call]}]
    if guard(step, tools=tools)["flagged"]:
        return {"role": "tool", "tool_call_id": tool_call["id"], "content": "Bu işlem engellendi."}

    # 2) Çalıştır, sonucu güvenilmeyen içerik olarak tara
    result = execute(tool_call)
    tool_msg = {"role": "tool", "tool_call_id": tool_call["id"],
                "content": [{"type": "text", "text": result}]}
    if guard(step + [tool_msg])["flagged"]:
        return {"role": "tool", "tool_call_id": tool_call["id"], "content": "Tool sonucu güvenilir değil."}
    return tool_msg

Bu adımda Lakera tarafında Agent Behavior Defense çalışır. Tool allow/deny list deterministiktir. Dangerous Deviation detector’ü ise tool call’u konuşma geçmişi ve tools tanımlarıyla birlikte değerlendirir ve user veya system mesajlarında gerekçesi olmayan tehlikeli aksiyonları (data exfiltration, privilege escalation, kalıcı kaynak silme) flag’ler. tool mesajlarından gelen talimatlar gerekçe olarak kabul edilmez.

Canlıya alma adımları

flowchart LR
    A[Detect modu<br/>async, result seviyeleri loglanır] --> B[Seviye dağılımından<br/>threshold seçimi]
    B --> C[Enforce<br/>belirlenen threshold ile]
    C --> D[İzleme ve<br/>gerekirse threshold güncelleme]

Kalibrasyonda kullanılacak asıl veri breakdown içindeki result alanıdır. Guard her detector için policy’deki threshold’dan bağımsız olarak metnin gerçek seviyesini (l1_confident, l2_very_likely, l3_likely, l4_less_likely, l5_unlikely) döndürür. Bir threshold kendi seviyesini ve daha güçlü seviyeleri flag’lediği için, örneğin L3 threshold’u l1, l2 ve l3 sonuçlarını yakalar. Detect modunda bu seviyeler loglanırsa her threshold’un gerçek trafikte ne kadar flag üreteceği Enforce’a geçmeden hesaplanabilir.

LEVELS = ["l1_confident", "l2_very_likely", "l3_likely", "l4_less_likely"]

def flagged_at(results, threshold):
    """results: Detect loglarından bir detector'ün result değerleri"""
    allowed = LEVELS[: LEVELS.index(threshold) + 1]
    return sum(1 for r in results if r in allowed)

for t in LEVELS:
    print(t, flagged_at(results, t))

1. Detect modunda başlanır ve her istekte detector bazında result seviyeleri loglanır.
2. Flag’lenen örnekler seviyelerine göre incelenir, her threshold’daki false positive ve false negative oranı çıkarılır. Türkçe gibi dillerde ve kısa mesajlarda davranış bu aşamada kendi verinizle test edilmelidir.
3. Enforce, veriyle belirlenen threshold ile açılır.
4. Enforce sonrasında flag’ler izlenir ve threshold gerekirse güncellenir.

Detect verisi azsa veya gerçek kullanımı yansıtmıyorsa Enforce’u en düşük hassasiyetle (L1) açıp threshold’u kademeli olarak yükseltmek daha güvenli bir alternatiftir.

Mod ve threshold policy tarafında yönetildiği için bu adımlar kod değişikliği gerektirmez. Bu nedenle entegrasyon baştan output kontrolünü de içerecek şekilde kurulmalıdır.