LLM 應用之縱深防禦

# 速率限制中介層範例
from datetime import datetime, timedelta
 
class RateLimiter:
    def __init__(self, max_requests: int = 60, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = timedelta(seconds=window_seconds)
        self.requests: dict[str, list[datetime]] = {}
 
    def check(self, user_id: str) -> bool:
        now = datetime.utcnow()
        user_requests = self.requests.get(user_id, [])
        # 移除過期項
        user_requests = [r for r in user_requests if now - r < self.window]
        if len(user_requests) >= self.max_requests:
            return False
        user_requests.append(now)
        self.requests[user_id] = user_requests
        return True

class InputProcessor:
    def __init__(self):
        self.prompt_shield = PromptShield()
        self.content_safety = ContentSafetyClient()
        self.max_input_tokens = 4096
 
    def process(self, user_input: str, session: Session) -> ProcessResult:
        # 大小檢查
        if count_tokens(user_input) > self.max_input_tokens:
            return ProcessResult.blocked("Input exceeds maximum length")
 
        # 提示注入偵測
        if self.prompt_shield.is_injection(user_input):
            return ProcessResult.blocked("Input flagged as injection")
 
        # 內容安全檢查
        safety = self.content_safety.analyze(user_input)
        if safety.max_severity > THRESHOLD:
            return ProcessResult.blocked(f"Content safety: {safety.category}")
 
        # Session 層行為分析
        session.add_message(user_input)
        if session.escalation_score > ESCALATION_THRESHOLD:
            return ProcessResult.blocked("Behavioral escalation detected")
 
        return ProcessResult.allowed(user_input)