الفصل 2: التخزين المؤقت والأداء

أنواع التخزين المؤقت

نوع التخزين	ما يُخزَّن	الفائدة
Embedding	embeddings الاستعلامات	تقليل استدعاءات API
دلالي	نتائج استعلامات مشابهة	تجاوز الاسترجاع
إجابة	إجابات RAG الكاملة	الأسرع

تخزين Embeddings مؤقتًا

from functools import lru_cache
import hashlib

class CachedEmbeddings:
    def __init__(self, embeddings, cache_size=10000):
        self.embeddings = embeddings
        self._cache = {}
    
    def embed_query(self, text):
        key = hashlib.md5(text.encode()).hexdigest()
        if key not in self._cache:
            self._cache[key] = self.embeddings.embed_query(text)
        return self._cache[key]

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

تخزين مؤقت بناءً على تشابه الاستعلام:

from langchain.cache import GPTCache
import gptcache

def get_similar_query(query):
    """Find cached result for similar query"""
    query_embedding = embed(query)
    
    # Search cache vector store
    results = cache_store.similarity_search(
        query_embedding, 
        threshold=0.95  # Very similar queries
    )
    
    if results:
        return results[0].metadata["response"]
    return None

تخزين الإجابات مؤقتًا باستخدام Redis

import redis
import json

r = redis.Redis()
CACHE_TTL = 3600  # 1 hour

def cached_rag(query, rag_chain):
    # Check cache
    cached = r.get(f"rag:{query}")
    if cached:
        return json.loads(cached)
    
    # Generate and cache
    response = rag_chain.invoke(query)
    r.setex(f"rag:{query}", CACHE_TTL, json.dumps(response))
    
    return response

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

def invalidate_on_update(document_id):
    """Invalidate caches when documents change"""
    # Clear related response caches
    r.delete_pattern(f"rag:*doc:{document_id}*")
    
    # Remove from semantic cache
    semantic_cache.delete(document_id)

مقارنة الأداء

السيناريو	زمن الاستجابة	التكلفة
بلا تخزين مؤقت	2-5s	0.01 دولار/استعلام
تخزين embedding	1-3s	0.005 دولار/استعلام
إصابة تخزين دلالي	<100ms	~0 دولار
إصابة تخزين إجابة	<50ms	~0 دولار