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 instruction ile işlediği veriyi aynı context içinde ayırt edemiyor. Tool use ve agent’larla birlikte modele giren untrusted content ve modelin tetikleyebildiği aksiyonlar arttıkça kontrolü modelin dışına, runtime’da 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. Pattern’lerin çoğu diğer guardrail ürünlerine de uygulanabilir.

Guardrail nedir?

Guardrail, LLM’in önünde ve arkasında çalışan, metni classify edip her threat kategorisi için bir score döndüren bağımsız bir servistir.

• Generative değil, discriminative modellerden oluşur. Metni input olarak alır ama metin üretmez, output olarak kategori ve confidence score verir. Tipik olarak encoder-based classifier’lar, NER modelleri, regex ve checksum validation birlikte kullanılır.
• LLM’in context’i dışında çalışır. Karar, korunan modelin instruction following 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.
• Low 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 judge olarak kullanmak (LLM-as-a-judge) bu latency budget’a genelde sığmaz.
• Policy decision point’tir, enforcement point değildir. Guardrail score’ları policy’deki threshold’larla karşılaştırıp flagged kararını döner. Block, mask veya sadece log 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 threat 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[Score'ları 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 role’ü 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.
• Sensitivity 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üğü response ş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 environment başına ayrı proje tanımlamak policy’leri birbirinden izole eder.
• flagged policy kararıdır, breakdown ise her detector’ün sonucunu ve confidence 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 template’lere gömmek yerine modelin gördüğü raw mesajları göndermek en doğru sonucu verir.

Entegrasyon yöntemleri

1. Asenkron monitoring (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: Response
    U-)G: output kontrolü (async)
    U-->>K: Response
    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. Blocking (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: Response
        U->>G: output kontrolü
        G-->>U: true / false
        U-->>K: Response 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 generic bir mesaj kullanılır.
• Guard’a erişilemediğinde uygulamanın nasıl davranacağı tanımlanmalıdır. Fail-open availability’yi, 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 response 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 response 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 third-party bir LLM provider’a gitmemesi gerekiyorsa input kontrolü sıralı yapılmalıdır.

4. Tek çağrıyla interaction taraması

Lakera’nın önerdiği minimum entegrasyon, LLM response’undan sonra user mesajı ve response’u tek çağrıda göndermektir. Guard last interaction’ı (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 last interaction davranışı sürüme ve konfigürasyona göre farklılık gösterebileceği için user 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. User isteği, agent’ın ürettiği tool call (execute edilmeden önce), tool result ve final response ayrı ayrı kontrol edilir. Tool definition’ları 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: user mesajı
    U->>L: mesaj + tool definitions
    L-->>U: tool_call (ör. export_records)
    U->>G: assistant tool_call + tools
    G-->>U: tehlikeli / yetki dışı mı?
    U->>T: tool call'u execute et (temizse)
    T-->>U: tool result
    U->>G: tool result (indirect injection?)
    U->>L: tool result
    L-->>U: final response
    U->>G: assistant response

def run_tool_step(msgs, tool_call, tools):
    # 1) Execute etmeden ö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) Execute et, sonucu untrusted content 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 conversation history ve tools definition’larıyla birlikte değerlendirir ve user veya system mesajlarında gerekçesi olmayan tehlikeli aksiyonları (data exfiltration, privilege escalation, kalıcı resource silme) flag’ler. tool mesajlarından gelen instruction’lar 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[Monitoring 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 sensitivity ile (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.