Runtime-monitoring & anomaliedetectie

LLM-applicaties in productie monitoren op anomalieën in tokengebruik, detectie van uitvoerpatronen, gedragsdrift, en het gebruik van tools zoals Langfuse, Helicone en aangepaste logging.

Runtime-monitoring is de verdedigingslaag die opvangt wat realtime-filters missen. Terwijl invoer- en uitvoerfilters per verzoek binaire blokkeren/toestaan-beslissingen nemen, analyseert monitoring patronen over tijd, gebruikers en sessies om langzaam smeulende aanvallen, geautomatiseerd sonderen en post-exploitatieactiviteit te detecteren.

Wat te monitoren

Anomalieën in tokengebruik

Ongebruikelijke tokenpatronen zijn sterke indicatoren van aanvalsactiviteit:

# Detectie van tokenanomalieën
class TokenAnomalyDetector:
    def __init__(self, window_size: int = 100):
        self.input_lengths: list[int] = []
        self.output_lengths: list[int] = []
        self.window_size = window_size
 
    def check(self, input_tokens: int, output_tokens: int) -> list[str]:
        alerts = []
        if len(self.input_lengths) >= self.window_size:
            avg_input = sum(self.input_lengths[-self.window_size:]) / self.window_size
            if input_tokens > avg_input * 5:
                alerts.append(f"Input length anomaly: {input_tokens} vs avg {avg_input:.0f}")
 
            avg_output = sum(self.output_lengths[-self.window_size:]) / self.window_size
            if output_tokens > avg_output * 10:
                alerts.append(f"Output length anomaly: {output_tokens} vs avg {avg_output:.0f}")
 
        self.input_lengths.append(input_tokens)
        self.output_lengths.append(output_tokens)
        return alerts

Detectie van uitvoerpatronen

Monitor de uitvoer van het model op patronen die wijzen op succesvolle uitbuiting:

Gedragsdrift

Gedragsdrift treedt op wanneer de responspatronen van het model veranderen zonder opzettelijke updates:

• Onderwerpdrift -- het model begint onderwerpen te bespreken buiten zijn bedoelde scope
• Toonverschuiving -- het model wordt na verloop van tijd meegaander of agressiever
• Capaciteitsuitbreiding -- het model begint tools of formaten te gebruiken die het eerder vermeed
• Veiligheidsdegradatie -- het weigeringspercentage daalt binnen een sessie of over sessies heen

Monitoringtools

Langfuse is een open-source LLM-observability-platform dat traces, metrieken en evaluaties vastlegt.

from langfuse import Langfuse
 
langfuse = Langfuse()
 
# Trace elke LLM-interactie
trace = langfuse.trace(name="chat-completion", user_id=user_id)
generation = trace.generation(
    name="main-llm",
    model="gpt-4o",
    input=user_message,
    output=response,
    metadata={
        "input_tokens": usage.prompt_tokens,
        "output_tokens": usage.completion_tokens,
        "refusal_detected": "sorry" in response.lower()[:50],
    },
)
 
# Score voor veiligheidsevaluatie
trace.score(name="safety", value=safety_score, comment="automated check")

Beveiligingsrelevante functies:

• Metrieken op traceniveau voor analyse per verzoek
• Sessie-tracking voor monitoring van gedrag over meerdere beurten
• Aangepaste scores voor veiligheidsevaluatiepijplijnen
• Dashboard voor het visualiseren van anomalieën in de tijd

Helicone biedt logging en analyse van verzoeken met een op proxy gebaseerde integratie die minimale codewijzigingen vereist.

from openai import OpenAI
 
# Route via de Helicone-proxy voor automatische logging
client = OpenAI(
    base_url="https://oai.helicone.ai/v1",
    default_headers={
        "Helicone-Auth": f"Bearer {HELICONE_API_KEY}",
        "Helicone-User-Id": user_id,
        "Helicone-Session-Id": session_id,
        "Helicone-Property-SecurityCheck": str(passed_security),
    },
)

Beveiligingsrelevante functies:

• Automatische logging van verzoek/respons zonder codewijzigingen
• Aangepaste eigenschappen voor het taggen van beveiligingsrelevante metadata
• Kostenbeheer (afwijkende kosten = mogelijke uitbuiting)
• Analyse op gebruikersniveau voor het identificeren van misbruikende accounts

Bouw voor maximale controle aangepaste logging-pijplijnen:

import json
import hashlib
from datetime import datetime
 
class SecurityLogger:
    def __init__(self, log_sink):
        self.sink = log_sink
 
    def log_interaction(
        self,
        user_id: str,
        session_id: str,
        input_text: str,
        output_text: str,
        metadata: dict,
    ):
        record = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "session_id": session_id,
            "input_hash": hashlib.sha256(input_text.encode()).hexdigest()[:16],
            "input_tokens": metadata.get("input_tokens"),
            "output_tokens": metadata.get("output_tokens"),
            "input_flagged": metadata.get("input_flagged", False),
            "output_flagged": metadata.get("output_flagged", False),
            "refusal_detected": metadata.get("refusal_detected", False),
            "latency_ms": metadata.get("latency_ms"),
        }
        self.sink.write(json.dumps(record))

Wanneer te kiezen voor aangepaste logging:

• Je moet loggen naar een specifieke SIEM (Splunk, Elastic, Datadog)
• Compliance vereist specifieke gegevensverwerking (geen logging door derden)
• Je hebt realtime-alarmintegratie nodig met bestaande incidentrespons

Alarmeringsstrategie

Effectieve alarmen moeten gevoeligheid (aanvallen opvangen) in balans brengen met specificiteit (operators niet overweldigen met valse alarmen).

Getrapt alarmmodel

Alarmafstemming

# Voorbeeld: P1-alarm voor aanhoudende injectiepogingen
class InjectionRateAlert:
    def __init__(self, threshold: int = 5, window_minutes: int = 10):
        self.threshold = threshold
        self.window_minutes = window_minutes
        self.injection_attempts: dict[str, list[datetime]] = {}
 
    def record_attempt(self, user_id: str) -> bool:
        now = datetime.utcnow()
        cutoff = now - timedelta(minutes=self.window_minutes)
        attempts = self.injection_attempts.get(user_id, [])
        attempts = [a for a in attempts if a > cutoff]
        attempts.append(now)
        self.injection_attempts[user_id] = attempts
 
        if len(attempts) >= self.threshold:
            return True  # Activeer P1-alarm
        return False

Implicaties voor red teams

Begrip van monitoring helpt red teamers detectie te vermijden:

1. Varieer payloads -- verstuur niet herhaaldelijk dezelfde aanvalsstring; monitoring zal ze clusteren
2. Doseer verzoeken -- stem af op normale gebruikersverkeerspatronen; bursts triggeren snelheidsalarmen
3. Gebruik verschillende sessies -- spreid pogingen over sessies om escalatiedetectie per sessie te vermijden
4. Let op tokengebruik -- houd invoer- en uitvoerlengtes waar mogelijk binnen normale bereiken
5. Roteer identiteiten -- gebruik indien mogelijk verschillende API-sleutels of accounts om markering op gebruikersniveau te vermijden

Verder lezen

• Defense-in-depth voor LLM-apps -- waar monitoring past in de verdedigingsstack
• Rate limiting, sandboxing & uitvoeringscontroles -- aanvullende infrastructuurcontroles
• Het AI-verdedigingslandschap -- breder overzicht van verdedigingstools

Verwante onderwerpen

• Defense-in-depth voor LLM-apps - Waar monitoring past in de verdedigingsstack
• Rate limiting, sandboxing & uitvoeringscontroles - Aanvullende infrastructuurcontroles
• Het AI-verdedigingslandschap - Breder overzicht van verdedigingstools, inclusief monitoringleveranciers
• Denken als een verdediger - Begrijpen hoe verdedigers alarmen en monitoring prioriteren

Referenties

• "Langfuse Documentation: LLM Observability" - Langfuse (2025) - Reference documentation for the open-source LLM observability platform
• "Helicone: LLM Monitoring and Analytics" - Helicone (2025) - Documentation for proxy-based LLM request logging and analytics
• "Detecting Adversarial Attacks on Neural Network Policies with Visual Foresight" - Various researchers (2024) - Research on anomaly detection approaches applicable to LLM monitoring
• "MITRE ATT&CK for Enterprise: Detection" - MITRE (2025) - Detection methodology framework adapted for AI system monitoring