Coverage for src / osiris_cli / performance.py: 0%

1#!/usr/bin/env python3 

2""" 

3Performance Optimizer for Osiris CLI 

4Caches responses, optimizes API calls, and manages resources 

5""" 

6 

7import hashlib 

8import json 

9from pathlib import Path 

10from typing import Optional, Dict, Any 

11from datetime import datetime, timedelta 

12from rich.console import Console 

13 

14console = Console() 

15 

16 

17class ResponseCache: 

18 """Cache AI responses to reduce API calls""" 

19 

20 def __init__(self, cache_dir: Path = None): 

21 if cache_dir is None: 

22 cache_dir = Path.home() / ".osiris" / "cache" 

23 

24 self.cache_dir = cache_dir 

25 self.cache_dir.mkdir(parents=True, exist_ok=True) 

26 self.cache_file = self.cache_dir / "response_cache.json" 

27 self.cache: Dict[str, Dict[str, Any]] = {} 

28 self.max_age_hours = 24 # Cache expires after 24 hours 

29 self.max_cache_size = 100 # Keep last 100 responses 

30 self.load() 

31 

32 def _get_cache_key(self, prompt: str, model: str, provider: str) -> str: 

33 """Generate cache key from prompt, model, and provider""" 

34 key_str = f"{provider}:{model}:{prompt}" 

35 return hashlib.sha256(key_str.encode()).hexdigest() 

36 

37 def load(self): 

38 """Load cache from disk""" 

39 if self.cache_file.exists(): 

40 try: 

41 with open(self.cache_file, 'r') as f: 

42 self.cache = json.load(f) 

43 except Exception: 

44 self.cache = {} 

45 

46 def save(self): 

47 """Save cache to disk""" 

48 try: 

49 # Keep only most recent entries 

50 if len(self.cache) > self.max_cache_size: 

51 # Sort by timestamp and keep newest 

52 sorted_items = sorted( 

53 self.cache.items(), 

54 key=lambda x: x[1].get('timestamp', ''), 

55 reverse=True 

56 ) 

57 self.cache = dict(sorted_items[:self.max_cache_size]) 

58 

59 with open(self.cache_file, 'w') as f: 

60 json.dump(self.cache, f, indent=2) 

61 except Exception as e: 

62 console.print(f"[yellow]Warning: Could not save cache: {e}[/yellow]") 

63 

64 def get(self, prompt: str, model: str, provider: str) -> Optional[str]: 

65 """Get cached response if available and not expired""" 

66 cache_key = self._get_cache_key(prompt, model, provider) 

67 

68 if cache_key in self.cache: 

69 entry = self.cache[cache_key] 

70 timestamp = datetime.fromisoformat(entry['timestamp']) 

71 age = datetime.now() - timestamp 

72 

73 # Check if cache is still valid 

74 if age < timedelta(hours=self.max_age_hours): 

75 console.print(f"[dim]Using cached response (age: {age.seconds // 60}m)[/dim]") 

76 return entry['response'] 

77 else: 

78 # Remove expired entry 

79 del self.cache[cache_key] 

80 

81 return None 

82 

83 def set(self, prompt: str, model: str, provider: str, response: str): 

84 """Cache a response""" 

85 cache_key = self._get_cache_key(prompt, model, provider) 

86 

87 self.cache[cache_key] = { 

88 'prompt': prompt[:100], # Store truncated prompt for reference 

89 'model': model, 

90 'provider': provider, 

91 'response': response, 

92 'timestamp': datetime.now().isoformat() 

93 } 

94 

95 self.save() 

96 

97 def clear(self): 

98 """Clear all cached responses""" 

99 self.cache = {} 

100 self.save() 

101 console.print("[green]✓[/green] Cache cleared") 

102 

103 def get_stats(self) -> Dict[str, Any]: 

104 """Get cache statistics""" 

105 if not self.cache: 

106 return { 

107 'entries': 0, 

108 'size_kb': 0, 

109 'oldest': None, 

110 'newest': None 

111 } 

112 

113 timestamps = [datetime.fromisoformat(e['timestamp']) for e in self.cache.values()] 

114 

115 return { 

116 'entries': len(self.cache), 

117 'size_kb': self.cache_file.stat().st_size / 1024 if self.cache_file.exists() else 0, 

118 'oldest': min(timestamps).isoformat() if timestamps else None, 

119 'newest': max(timestamps).isoformat() if timestamps else None 

120 } 

121 

122 

123class PerformanceOptimizer: 

124 """Optimize performance across the CLI""" 

125 

126 def __init__(self): 

127 self.response_cache = ResponseCache() 

128 self.stats = { 

129 'api_calls': 0, 

130 'cache_hits': 0, 

131 'cache_misses': 0, 

132 'total_tokens': 0, 

133 'total_time': 0.0 

134 } 

135 

136 def should_use_cache(self, prompt: str) -> bool: 

137 """Determine if caching should be used for this prompt""" 

138 # Don't cache commands 

139 if prompt.startswith('/'): 

140 return False 

141 

142 # Don't cache very short prompts 

143 if len(prompt) < 10: 

144 return False 

145 

146 # Don't cache prompts with time-sensitive keywords 

147 time_keywords = ['now', 'today', 'current', 'latest', 'recent'] 

148 if any(keyword in prompt.lower() for keyword in time_keywords): 

149 return False 

150 

151 return True 

152 

153 def get_cached_response(self, prompt: str, model: str, provider: str) -> Optional[str]: 

154 """Try to get cached response""" 

155 if not self.should_use_cache(prompt): 

156 return None 

157 

158 response = self.response_cache.get(prompt, model, provider) 

159 

160 if response: 

161 self.stats['cache_hits'] += 1 

162 else: 

163 self.stats['cache_misses'] += 1 

164 

165 return response 

166 

167 def cache_response(self, prompt: str, model: str, provider: str, response: str): 

168 """Cache a response""" 

169 if self.should_use_cache(prompt): 

170 self.response_cache.set(prompt, model, provider, response) 

171 

172 def record_api_call(self, tokens: int, time_seconds: float): 

173 """Record API call statistics""" 

174 self.stats['api_calls'] += 1 

175 self.stats['total_tokens'] += tokens 

176 self.stats['total_time'] += time_seconds 

177 

178 def get_stats(self) -> Dict[str, Any]: 

179 """Get performance statistics""" 

180 cache_stats = self.response_cache.get_stats() 

181 

182 total_requests = self.stats['cache_hits'] + self.stats['cache_misses'] 

183 cache_hit_rate = (self.stats['cache_hits'] / total_requests * 100) if total_requests > 0 else 0 

184 

185 avg_time = self.stats['total_time'] / self.stats['api_calls'] if self.stats['api_calls'] > 0 else 0 

186 

187 return { 

188 'api_calls': self.stats['api_calls'], 

189 'cache_hits': self.stats['cache_hits'], 

190 'cache_misses': self.stats['cache_misses'], 

191 'cache_hit_rate': f"{cache_hit_rate:.1f}%", 

192 'total_tokens': self.stats['total_tokens'], 

193 'avg_response_time': f"{avg_time:.2f}s", 

194 'cache_entries': cache_stats['entries'], 

195 'cache_size_kb': f"{cache_stats['size_kb']:.1f} KB" 

196 } 

197 

198 def clear_cache(self): 

199 """Clear response cache""" 

200 self.response_cache.clear() 

201 

202 

203# Global optimizer instance 

204performance_optimizer = PerformanceOptimizer()