執行時監控與異常偵測

# Token 異常偵測
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

from langfuse import Langfuse
 
langfuse = Langfuse()
 
# 追蹤每次 LLM 互動
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],
    },
)
 
# 為安全評估評分
trace.score(name="safety", value=safety_score, comment="automated check")

from openai import OpenAI
 
# 經 Helicone 代理路由以自動記錄
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),
    },
)

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))

# 範例：為持續注入嘗試之 P1 警報
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  # 觸發 P1 警報
        return False