استراتيجيات التخزين المؤقت

التخزين المؤقت هو أحد أكثر الطرق فعالية لتقليل تكاليف LLM وزمن الاستجابة. التخزين المؤقت المصمم جيداً يمكن أن يقلل التكاليف بنسبة 40-60% مع تحسين أوقات الاستجابة.

أنواع التخزين المؤقت للذكاء الاصطناعي

1. التخزين المؤقت للمطابقة التامة

النهج الأبسط—تخزين الاستعلامات المتطابقة.

import hashlib
import redis

class ExactMatchCache:
    def __init__(self, redis_client, ttl_seconds=3600):
        self.redis = redis_client
        self.ttl = ttl_seconds

    def _hash_key(self, prompt: str) -> str:
        return hashlib.sha256(prompt.encode()).hexdigest()

    async def get(self, prompt: str) -> Optional[str]:
        key = self._hash_key(prompt)
        cached = await self.redis.get(f"llm_cache:{key}")
        return cached.decode() if cached else None

    async def set(self, prompt: str, response: str):
        key = self._hash_key(prompt)
        await self.redis.setex(f"llm_cache:{key}", self.ttl, response)

# الاستخدام
cache = ExactMatchCache(redis_client)
cached_response = await cache.get(prompt)
if cached_response:
    return cached_response

response = await llm.complete(prompt)
await cache.set(prompt, response)

معدل الإصابة: 20-40% في التطبيقات النموذجية.

2. التخزين المؤقت الدلالي

تخزين الاستعلامات المتشابهة، ليس فقط المتطابقة.

from numpy import dot
from numpy.linalg import norm

class SemanticCache:
    def __init__(self, embedding_model, vector_store, threshold=0.95):
        self.embedder = embedding_model
        self.store = vector_store
        self.threshold = threshold

    async def get(self, query: str) -> Optional[str]:
        query_embedding = await self.embedder.embed(query)

        # البحث عن استعلامات مخزنة مشابهة
        results = await self.store.search(
            embedding=query_embedding,
            top_k=1
        )

        if results and results[0].score > self.threshold:
            return results[0].metadata["response"]
        return None

    async def set(self, query: str, response: str):
        embedding = await self.embedder.embed(query)
        await self.store.insert(
            embedding=embedding,
            metadata={"query": query, "response": response}
        )

معدل الإصابة: 40-60%، لكن مع مقايضات:

العامل	المطابقة التامة	التخزين المؤقت الدلالي
معدل الإصابة	أقل	أعلى
عبء زمن الاستجابة	الحد الأدنى	تكلفة التضمين
الإيجابيات الخاطئة	لا يوجد	محتمل
تكلفة التخزين	أقل	أعلى

3. التخزين المؤقت السياقي

تخزين الاستجابات بناءً على تشابه السياق، ليس الاستعلام فقط.

class ContextualCache:
    def __init__(self, semantic_cache):
        self.cache = semantic_cache

    async def get(self, query: str, context: str) -> Optional[str]:
        # دمج الاستعلام والسياق للبحث في التخزين المؤقت
        combined = f"الاستعلام: {query}\nالسياق: {context[:500]}"
        return await self.cache.get(combined)

    async def set(self, query: str, context: str, response: str):
        combined = f"الاستعلام: {query}\nالسياق: {context[:500]}"
        await self.cache.set(combined, response)

إبطال التخزين المؤقت

أصعب مشكلة في علوم الحاسوب—الآن مع تعقيد الذكاء الاصطناعي.

class SmartCacheManager:
    def __init__(self, cache, ttl_config):
        self.cache = cache
        self.ttl_config = ttl_config

    async def set_with_smart_ttl(self, key: str, value: str, query_type: str):
        # TTL مختلفة بناءً على نوع المحتوى
        ttl = self.ttl_config.get(query_type, 3600)
        await self.cache.set(key, value, ttl=ttl)

    async def invalidate_by_topic(self, topic: str):
        # إبطال جميع الاستجابات المخزنة المتعلقة بموضوع
        # مفيد عند تغير البيانات الأساسية
        keys = await self.cache.search_keys(f"*{topic}*")
        for key in keys:
            await self.cache.delete(key)

# تكوين TTL
TTL_CONFIG = {
    "factual": 86400,      # 24 ساعة - الحقائق لا تتغير كثيراً
    "time_sensitive": 300,  # 5 دقائق - الأحداث الجارية
    "personalized": 3600,   # ساعة واحدة - خاص بالمستخدم
    "creative": 0           # لا تخزين مؤقت - كل استجابة يجب أن تكون فريدة
}

التخزين المؤقت متعدد الطبقات

┌─────────────────────────────────────────────┐
│              تدفق الطلب                      │
├─────────────────────────────────────────────┤
│                                             │
│   استعلام ──▶ L1: الذاكرة ──▶ إصابة؟ إرجاع │
│                   │                         │
│                   ▼ عدم إصابة               │
│            L2: Redis ──▶ إصابة؟ إرجاع      │
│                   │                         │
│                   ▼ عدم إصابة               │
│            L3: دلالي ──▶ إصابة؟ إرجاع      │
│                   │                         │
│                   ▼ عدم إصابة               │
│               استدعاء LLM                   │
│                   │                         │
│                   ▼                         │
│            تحديث جميع الطبقات              │
│                                             │
└─────────────────────────────────────────────┘

class MultiLayerCache:
    def __init__(self, l1_memory, l2_redis, l3_semantic):
        self.layers = [l1_memory, l2_redis, l3_semantic]

    async def get(self, query: str) -> tuple[Optional[str], str]:
        for i, layer in enumerate(self.layers):
            result = await layer.get(query)
            if result:
                # ملء الطبقات الأسرع
                for faster_layer in self.layers[:i]:
                    await faster_layer.set(query, result)
                return result, f"L{i+1}"
        return None, "MISS"

    async def set(self, query: str, response: str):
        for layer in self.layers:
            await layer.set(query, response)

مقاييس التخزين المؤقت

تتبع هذه لتحسين التخزين المؤقت:

class CacheMetrics:
    def __init__(self):
        self.hits = 0
        self.misses = 0
        self.hit_latency = []
        self.miss_latency = []

    def record_hit(self, latency_ms: float, layer: str):
        self.hits += 1
        self.hit_latency.append(latency_ms)
        # التسجيل للمراقبة
        metrics.increment("cache_hit", tags={"layer": layer})

    def record_miss(self, latency_ms: float):
        self.misses += 1
        self.miss_latency.append(latency_ms)
        metrics.increment("cache_miss")

    @property
    def hit_rate(self) -> float:
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0

نصيحة للمقابلة

عند مناقشة التخزين المؤقت في المقابلات، دائماً اذكر:

1. ماذا تخزن مؤقتاً (الاستعلامات، التضمينات، الاستجابات الكاملة)
2. كيف تتعامل مع الإبطال (TTL، مدفوع بالأحداث، يدوي)
3. المقايضات (معدل الإصابة مقابل الحداثة، الذاكرة مقابل زمن الاستجابة)

بعد ذلك، سنستكشف استراتيجيات تحسين التكلفة بخلاف التخزين المؤقت. :::