Spaces:
Running
Running
File size: 1,655 Bytes
2ecc4a7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | 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
|