Coverage for src/dataknobs_llm/prompts/implementations/config_library.py: 24%

1"""Configuration-based prompt library implementation. 

2 

3This module provides a prompt library that loads prompts from Python dictionaries. 

4Useful for programmatic prompt definition and testing. 

5 

6Example: 

7 config = { 

8 "system": { 

9 "analyze_code": { 

10 "template": "Analyze this {{language}} code:\\n{{code}}", 

11 "defaults": {"language": "python"}, 

12 "validation": { 

13 "level": "error", 

14 "required_params": ["code"] 

15 } 

16 } 

17 }, 

18 "user": { 

19 "code_question": { 

20 "template": "Please analyze the following code...", 

21 "rag_configs": [...] 

22 }, 

23 "followup_question": { 

24 "template": "Additionally, check for..." 

25 } 

26 } 

27 } 

28 

29 library = ConfigPromptLibrary(config) 

30 template = library.get_system_prompt("analyze_code") 

31""" 

32 

33import logging 

34from typing import Any, Dict, List, Optional 

35 

36from ..base import ( 

37 BasePromptLibrary, 

38 PromptTemplate, 

39 RAGConfig, 

40 MessageIndex, 

41 ValidationConfig, 

42 ValidationLevel, 

43) 

44 

45logger = logging.getLogger(__name__) 

46 

47 

48class ConfigPromptLibrary(BasePromptLibrary): 

49 """Prompt library that loads prompts from configuration dictionaries. 

50 

51 Features: 

52 - Direct in-memory configuration 

53 - No filesystem dependencies 

54 - Useful for testing and programmatic definition 

55 - Supports all prompt types (system, user, messages, RAG) 

56 

57 Example: 

58 >>> config = { 

59 ... "system": {"greet": {"template": "Hello {{name}}!"}}, 

60 ... "user": {"ask": {"template": "Tell me about {{topic}}"}} 

61 ... } 

62 >>> library = ConfigPromptLibrary(config) 

63 >>> template = library.get_system_prompt("greet") 

64 """ 

65 

66 def __init__(self, config: Optional[Dict[str, Any]] = None): 

67 """Initialize configuration-based prompt library. 

68 

69 Args: 

70 config: Configuration dictionary with structure: 

71 { 

72 "system": {name: PromptTemplate, ...}, 

73 "user": {name: PromptTemplate, ...}, 

74 "messages": {name: MessageIndex, ...}, 

75 "rag": {name: RAGConfig, ...} 

76 } 

77 """ 

78 super().__init__() 

79 self._config = config or {} 

80 

81 # Load prompts from config 

82 self._load_from_config() 

83 

84 def _load_from_config(self) -> None: 

85 """Load all prompts from the configuration dictionary.""" 

86 self._load_system_prompts() 

87 self._load_user_prompts() 

88 self._load_message_indexes() 

89 self._load_rag_configs() # Load standalone RAG configs 

90 

91 def _load_system_prompts(self) -> None: 

92 """Load system prompts from config.""" 

93 system_config = self._config.get("system", {}) 

94 

95 for name, data in system_config.items(): 

96 try: 

97 template = self._parse_prompt_template(data) 

98 self._cache_system_prompt(name, template) 

99 logger.debug(f"Loaded system prompt from config: {name}") 

100 except Exception as e: 

101 logger.error(f"Error loading system prompt {name}: {e}") 

102 

103 def _load_user_prompts(self) -> None: 

104 """Load user prompts from config.""" 

105 user_config = self._config.get("user", {}) 

106 

107 for name, data in user_config.items(): 

108 try: 

109 template = self._parse_prompt_template(data) 

110 self._cache_user_prompt(name, template) 

111 logger.debug(f"Loaded user prompt from config: {name}") 

112 except Exception as e: 

113 logger.error(f"Error loading user prompt {name}: {e}") 

114 

115 def _load_message_indexes(self) -> None: 

116 """Load message indexes from config.""" 

117 messages_config = self._config.get("messages", {}) 

118 

119 for name, data in messages_config.items(): 

120 try: 

121 message_index = self._parse_message_index(data) 

122 self._cache_message_index(name, message_index) 

123 logger.debug(f"Loaded message index from config: {name}") 

124 except Exception as e: 

125 logger.error(f"Error loading message index {name}: {e}") 

126 

127 def _load_rag_configs(self) -> None: 

128 """Load RAG configurations from config.""" 

129 rag_config = self._config.get("rag", {}) 

130 

131 for name, data in rag_config.items(): 

132 try: 

133 rag = self._parse_rag_config(data) 

134 self._cache_rag_config(name, rag) 

135 logger.debug(f"Loaded RAG config from config: {name}") 

136 except Exception as e: 

137 logger.error(f"Error loading RAG config {name}: {e}") 

138 

139 # Note: _parse_prompt_template(), _parse_validation_config(), and 

140 # _parse_rag_config() are now inherited from BasePromptLibrary 

141 

142 def _parse_message_index(self, data: Dict[str, Any]) -> MessageIndex: 

143 """Parse message index from config data. 

144 

145 Args: 

146 data: Message index data 

147 

148 Returns: 

149 MessageIndex dictionary 

150 """ 

151 message_index: MessageIndex = { 

152 "messages": data.get("messages", []), 

153 } 

154 

155 # Add optional fields 

156 if "rag_configs" in data: 

157 message_index["rag_configs"] = [ 

158 self._parse_rag_config(rag_data) 

159 for rag_data in data["rag_configs"] 

160 ] 

161 

162 if "metadata" in data: 

163 message_index["metadata"] = data["metadata"] 

164 

165 return message_index 

166 

167 def get_system_prompt(self, name: str, **kwargs) -> Optional[PromptTemplate]: 

168 """Get a system prompt by name. 

169 

170 Args: 

171 name: System prompt name 

172 **kwargs: Additional arguments (unused) 

173 

174 Returns: 

175 PromptTemplate if found, None otherwise 

176 """ 

177 return self._get_cached_system_prompt(name) 

178 

179 def get_user_prompt(self, name: str, **kwargs) -> Optional[PromptTemplate]: 

180 """Get a user prompt by name. 

181 

182 Args: 

183 name: User prompt name 

184 **kwargs: Additional arguments (unused) 

185 

186 Returns: 

187 PromptTemplate if found, None otherwise 

188 """ 

189 return self._get_cached_user_prompt(name) 

190 

191 def get_message_index(self, name: str, **kwargs) -> Optional[MessageIndex]: 

192 """Get a message index by name. 

193 

194 Args: 

195 name: Message index name 

196 **kwargs: Additional arguments (unused) 

197 

198 Returns: 

199 MessageIndex if found, None otherwise 

200 """ 

201 return self._get_cached_message_index(name) 

202 

203 def get_rag_config(self, name: str, **kwargs) -> Optional[RAGConfig]: 

204 """Get a standalone RAG configuration by name. 

205 

206 Args: 

207 name: RAG config name 

208 **kwargs: Additional arguments (unused) 

209 

210 Returns: 

211 RAGConfig if found, None otherwise 

212 """ 

213 return self._get_cached_rag_config(name) 

214 

215 def get_prompt_rag_configs( 

216 self, 

217 prompt_name: str, 

218 prompt_type: str = "user", 

219 **kwargs 

220 ) -> List[RAGConfig]: 

221 """Get RAG configurations for a specific prompt. 

222 

223 Resolves both inline RAG configs and references to standalone configs. 

224 

225 Args: 

226 prompt_name: Prompt name 

227 prompt_type: Type of prompt ("user" or "system") 

228 **kwargs: Additional arguments (unused) 

229 

230 Returns: 

231 List of RAGConfig (empty if none defined) 

232 """ 

233 # Get the prompt template 

234 if prompt_type == "system": 

235 template = self.get_system_prompt(prompt_name) 

236 else: 

237 template = self.get_user_prompt(prompt_name) 

238 

239 if template is None: 

240 return [] 

241 

242 configs = [] 

243 

244 # Get inline RAG configs from template 

245 if "rag_configs" in template: 

246 configs.extend(template["rag_configs"]) 

247 

248 # Resolve RAG config references 

249 if "rag_config_refs" in template: 

250 for ref_name in template["rag_config_refs"]: 

251 ref_config = self.get_rag_config(ref_name) 

252 if ref_config: 

253 configs.append(ref_config) 

254 else: 

255 logger.warning( 

256 f"RAG config reference '{ref_name}' not found " 

257 f"for prompt '{prompt_name}'" 

258 ) 

259 

260 return configs 

261 

262 def add_system_prompt(self, name: str, template: PromptTemplate) -> None: 

263 """Add or update a system prompt. 

264 

265 Args: 

266 name: System prompt name 

267 template: Prompt template to add 

268 """ 

269 self._cache_system_prompt(name, template) 

270 logger.debug(f"Added/updated system prompt: {name}") 

271 

272 def add_user_prompt(self, name: str, template: PromptTemplate) -> None: 

273 """Add or update a user prompt. 

274 

275 Args: 

276 name: User prompt name 

277 template: Prompt template to add 

278 """ 

279 self._cache_user_prompt(name, template) 

280 logger.debug(f"Added/updated user prompt: {name}") 

281 

282 def add_message_index(self, name: str, message_index: MessageIndex) -> None: 

283 """Add or update a message index. 

284 

285 Args: 

286 name: Message index name 

287 message_index: Message index to add 

288 """ 

289 self._cache_message_index(name, message_index) 

290 logger.debug(f"Added/updated message index: {name}") 

291 

292 def add_rag_config(self, name: str, rag_config: RAGConfig) -> None: 

293 """Add or update a RAG configuration. 

294 

295 Args: 

296 name: RAG config name 

297 rag_config: RAG configuration to add 

298 """ 

299 self._cache_rag_config(name, rag_config) 

300 logger.debug(f"Added/updated RAG config: {name}") 

301 

302 def list_system_prompts(self) -> List[str]: 

303 """List all available system prompt names. 

304 

305 Returns: 

306 List of system prompt identifiers 

307 """ 

308 return list(self._system_prompt_cache.keys()) 

309 

310 def list_user_prompts(self) -> List[str]: 

311 """List available user prompts. 

312 

313 Returns: 

314 List of user prompt names 

315 """ 

316 return list(self._user_prompt_cache.keys()) 

317 

318 def list_message_indexes(self) -> List[str]: 

319 """List all available message index names. 

320 

321 Returns: 

322 List of message index identifiers 

323 """ 

324 return list(self._message_index_cache.keys())