Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/cache/keys.py: 42%

1from __future__ import annotations 

2 

3import hashlib 

4from typing import Any 

5 

6from lexigram.serialization import dumps 

7 

8 

9class CacheKeyBuilder: 

10 """Builds consistent cache keys for different RAG components.""" 

11 

12 @staticmethod 

13 def build_embedding_key(text: str, model: str, prefix: str = "rag:") -> str: 

14 """Build cache key for embeddings.""" 

15 data = dumps({"text": text, "model": model}, sort_keys=True) 

16 hash_val = hashlib.sha256(data).hexdigest()[:16] 

17 return f"{prefix}embedding:{hash_val}" 

18 

19 @staticmethod 

20 def build_retrieval_key( 

21 query: str, 

22 params: dict[str, Any] | None = None, 

23 prefix: str = "rag:", 

24 ) -> str: 

25 """Build cache key for retrieval results.""" 

26 params = params or {} 

27 sorted_params = {k: params[k] for k in sorted(params.keys())} 

28 data = dumps({"query": query, "params": sorted_params}, sort_keys=True) 

29 hash_val = hashlib.sha256(data).hexdigest()[:16] 

30 return f"{prefix}retrieval:{hash_val}" 

31 

32 @staticmethod 

33 def build_document_key( 

34 doc_id: str, 

35 config: dict[str, Any] | None = None, 

36 prefix: str = "rag:", 

37 ) -> str: 

38 """Build cache key for preprocessed documents.""" 

39 config = config or {} 

40 sorted_config = {k: config[k] for k in sorted(config.keys())} 

41 data = dumps({"doc_id": doc_id, "config": sorted_config}, sort_keys=True) 

42 hash_val = hashlib.sha256(data).hexdigest()[:16] 

43 return f"{prefix}document:{hash_val}" 

44 

45 @staticmethod 

46 def build_reranking_key( 

47 query: str, 

48 document_ids: list[str], 

49 model: str, 

50 prefix: str = "rag:", 

51 ) -> str: 

52 """Build cache key for reranking results.""" 

53 data = dumps( 

54 { 

55 "query": query, 

56 "document_ids": sorted(document_ids), 

57 "model": model, 

58 }, 

59 sort_keys=True, 

60 ) 

61 hash_val = hashlib.sha256(data).hexdigest()[:16] 

62 return f"{prefix}reranking:{hash_val}" 

63 

64 @staticmethod 

65 def build_query_transformation_key( 

66 query: str, 

67 transformation_type: str, 

68 params: dict[str, Any] | None = None, 

69 prefix: str = "rag:", 

70 ) -> str: 

71 """Build cache key for query transformations.""" 

72 params = params or {} 

73 sorted_params = {k: params[k] for k in sorted(params.keys())} 

74 data = dumps( 

75 { 

76 "query": query, 

77 "type": transformation_type, 

78 "params": sorted_params, 

79 }, 

80 sort_keys=True, 

81 ) 

82 hash_val = hashlib.sha256(data).hexdigest()[:16] 

83 return f"{prefix}query_transform:{hash_val}"