Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/caching/types.py: 77%

1"""Types and models for LLM caching.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from datetime import UTC, datetime 

7from typing import Any 

8 

9from lexigram.ai.llm.protocols import LLMCacheProtocol 

10from lexigram.domain import DomainModel 

11from lexigram.security.hashing import ambient as hashing 

12from lexigram.validation import Field 

13 

14__all__ = [ 

15 "CacheEntry", 

16 "CacheStats", 

17 "LLMCacheProtocol", 

18 "build_llm_cache_key", 

19] 

20 

21 

22@dataclass(init=False) 

23class CacheEntry(DomainModel): 

24 """Cache entry with metadata. 

25 

26 Attributes: 

27 key: Cache key. 

28 value: Cached value. 

29 created_at: When entry was created. 

30 expires_at: When entry expires (Unix timestamp). 

31 hits: Number of cache hits. 

32 size_bytes: Approximate size in bytes. 

33 """ 

34 

35 key: str = Field(..., description="Cache key") 

36 value: Any = Field(..., description="Cached value") 

37 created_at: datetime = Field( 

38 default_factory=lambda: datetime.now(UTC), 

39 description="Creation timestamp", 

40 ) 

41 expires_at: float = Field(..., description="Expiration timestamp (Unix)") 

42 hits: int = Field(default=0, description="Number of hits") 

43 size_bytes: int = Field(default=0, description="Approximate size in bytes") 

44 

45 

46@dataclass(init=False) 

47class CacheStats(DomainModel): 

48 """Cache statistics. 

49 

50 Attributes: 

51 hits: Number of cache hits. 

52 misses: Number of cache misses. 

53 evictions: Number of evictions. 

54 total_entries: Current number of entries. 

55 total_size_bytes: Total cache size in bytes. 

56 """ 

57 

58 hits: int = Field(default=0, description="Cache hits") 

59 misses: int = Field(default=0, description="Cache misses") 

60 evictions: int = Field(default=0, description="Cache evictions") 

61 total_entries: int = Field(default=0, description="Total entries") 

62 total_size_bytes: int = Field(default=0, description="Total size in bytes") 

63 

64 @property 

65 def hit_rate(self) -> float: 

66 """Calculate cache hit rate.""" 

67 total = self.hits + self.misses 

68 return self.hits / total if total > 0 else 0.0 

69 

70 

71def build_llm_cache_key( 

72 provider: str, 

73 model: str, 

74 prompt: str, 

75 model_revision: str | None = None, 

76 prompt_hash: bytes | None = None, 

77) -> str: 

78 """Build a deterministic cache key for LLM responses. 

79 

80 Includes ``model_revision`` and ``prompt_hash`` when available so that 

81 different revisions or prompts produce distinct cache entries. 

82 

83 Args: 

84 provider: Provider name (e.g. ``"openai"``). 

85 model: Model name (e.g. ``"gpt-4"``). 

86 prompt: The raw prompt text. 

87 model_revision: Optional model revision string. 

88 prompt_hash: Optional SHA-256 hash of the prompt. 

89 

90 Returns: 

91 A hex-encoded SHA-256 cache key. 

92 """ 

93 key_parts: list[str] = [provider, model, prompt] 

94 if model_revision: 

95 key_parts.append(model_revision) 

96 if prompt_hash: 

97 key_parts.append(prompt_hash.hex()) 

98 return hashing.hash_hex("|".join(key_parts))