backend/rag_pipeline/context_cache.py

import hashlib
import json
from typing import Optional, List, Dict, Any

# Prosta implementacja Semantic/Context Caching w pamięci In-Memory.
# W pełnej skali zintegrowane np. z Redis lub Google Context Caching dla Gemini API.


class ContextCache:
    def __init__(self):
        self._cache = {}

    def _generate_key(self, query: str, filters: Dict[str, Any]) -> str:
        cache_data = {"query": query.lower().strip(), "filters": filters or {}}
        return hashlib.md5(json.dumps(cache_data, sort_keys=True).encode()).hexdigest()

    def get(self, query: str, filters: Dict[str, Any] = None) -> Optional[List[Dict]]:
        key = self._generate_key(query, filters)
        if key in self._cache:
            print("CACHE HIT! Zaoszczędzono czas i koszty zapytań (Context Caching).")
            return self._cache[key]
        return None

    def set(self, query: str, results: List[Dict], filters: Dict[str, Any] = None):
        key = self._generate_key(query, filters)
        self._cache[key] = results


# Globalna instancja systemu (można przepisać na Redisa)
semantic_cache = ContextCache()