RAG_FULL_APPLICATION_BACKEND/app/techniques/cache_incremental.py

from .base import BaseRAGTechnique
from ..services.cache_service import cache_service
from .hybrid_search import HybridSearch
from typing import List, Dict, Any

class CacheIncrementalRAG(BaseRAGTechnique):
    async def run(self, query: str, document_id: str, top_k: int = 5, **kwargs) -> Dict[str, Any]:
        technique_name = kwargs.get("underlying_technique", "hybrid")
        
        # 1. Cache Check
        await self.emit("CACHE_CHECK", "#6B7280", "Checking Redis cache for previous answer...")
        
        cached_result = cache_service.get(self.user_id, document_id, query, technique_name)
        if cached_result:
            await self.emit("CACHE_HIT", "#22C55E", "Cache hit! Returning stored answer (0ms).")
            return cached_result
            
        await self.emit("CACHE_MISS", "#8B5CF6", "Cache miss. Running full RAG pipeline...")
        
        # 2. Run Underlying Technique (e.g., Hybrid)
        # For simplicity, we use Hybrid as the default fallback
        underlying = HybridSearch(self.job_id, self.user_id)
        result = await underlying.run(query, document_id, top_k)
        
        # 3. Store in Cache
        cache_service.set(self.user_id, document_id, query, technique_name, result)
        
        await self.emit("DONE", "#22C55E", "Answer cached for future queries.")
        return result

    async def retrieve(self, query: str, document_id: str, top_k: int, **kwargs) -> List[Dict[str, Any]]:
        # Not used directly in Run override
        pass

    async def generate(self, query: str, chunks: List[Dict[str, Any]]) -> str:
        # Not used directly in Run override
        pass