import os import re import json from pathlib import Path from typing import Dict, Any, Set, Optional, Tuple import logging logger = logging.getLogger(__name__) class PromptManager: """Singleton class to manage prompt templates and JSON schemas""" _instance: Optional['PromptManager'] = None _initialized: bool = False def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self): if not self._initialized: self.prompts: Dict[str, str] = {} self.schemas: Dict[str, Dict[str, Any]] = {} self.prompt_variables: Dict[str, Set[str]] = {} self._load_all_prompts() PromptManager._initialized = True def _load_all_prompts(self): """Load all markdown files and corresponding JSON schemas from the prompts folder""" prompts_dir = Path(__file__).parent / 'prompts' if not prompts_dir.exists(): logger.warning(f"Prompts directory not found: {prompts_dir}") prompts_dir.mkdir(parents=True, exist_ok=True) return logger.info(f"Loading prompts and schemas from {prompts_dir}") # Load all .md files for md_file in prompts_dir.glob("*.md"): prompt_name = md_file.stem # filename without extension try: # Load prompt template with open(md_file, 'r', encoding='utf-8') as f: content = f.read().strip() # Extract variables from {{variable}} patterns variables = self._extract_variables(content) self.prompts[prompt_name] = content self.prompt_variables[prompt_name] = variables # Look for corresponding JSON schema file schema_file = md_file.with_suffix('.json') if schema_file.exists(): try: with open(schema_file, 'r', encoding='utf-8') as f: schema = json.load(f) self.schemas[prompt_name] = schema logger.debug(f"Loaded prompt '{prompt_name}' with schema and variables: {variables}") except json.JSONDecodeError as e: logger.error(f"Invalid JSON schema in {schema_file}: {e}") else: logger.debug(f"Loaded prompt '{prompt_name}' (no schema) with variables: {variables}") except Exception as e: logger.error(f"Error loading prompt file {md_file}: {e}") logger.info(f"Loaded {len(self.prompts)} prompt templates, {len(self.schemas)} with schemas") def _extract_variables(self, template: str) -> Set[str]: """Extract all {{variable}} placeholders from template""" pattern = r'\{\{(\w+)\}\}' variables = set(re.findall(pattern, template)) return variables def _validate_context(self, prompt_name: str, context: Dict[str, Any]) -> None: """Validate that all required variables are provided""" if prompt_name not in self.prompt_variables: raise ValueError(f"Unknown prompt: '{prompt_name}'") required_vars = self.prompt_variables[prompt_name] provided_vars = set(context.keys()) missing_vars = required_vars - provided_vars if missing_vars: raise ValueError( f"Missing required variables for prompt '{prompt_name}': {missing_vars}. " f"Required: {required_vars}, Provided: {provided_vars}" ) # Warn about extra variables (not an error, but might indicate mistakes) extra_vars = provided_vars - required_vars if extra_vars: logger.warning(f"Extra variables provided for prompt '{prompt_name}': {extra_vars}") def _fill_template(self, template: str, context: Dict[str, Any]) -> str: """Fill template with context variables""" result = template for key, value in context.items(): placeholder = f"{{{{{key}}}}}" # {{key}} result = result.replace(placeholder, str(value)) return result @classmethod def get_prompt(cls, prompt_name: str, context: Dict[str, Any] = None) -> str: """ Get a processed prompt with variables filled in Args: prompt_name: Name of the prompt template (filename without .md) context: Dictionary of variables to fill in the template Returns: Processed prompt string Raises: ValueError: If prompt doesn't exist or required variables are missing """ instance = cls() if prompt_name not in instance.prompts: available_prompts = list(instance.prompts.keys()) raise ValueError(f"Prompt '{prompt_name}' not found. Available prompts: {available_prompts}") context = context or {} # Validate that all required variables are provided instance._validate_context(prompt_name, context) # Fill the template template = instance.prompts[prompt_name] processed_prompt = instance._fill_template(template, context) return processed_prompt @classmethod def get_schema(cls, prompt_name: str) -> Optional[Dict[str, Any]]: """ Get the JSON schema for a prompt if it exists Args: prompt_name: Name of the prompt template Returns: JSON schema dictionary or None if no schema exists """ instance = cls() if prompt_name not in instance.prompts: raise ValueError(f"Prompt '{prompt_name}' not found") return instance.schemas.get(prompt_name) @classmethod def has_schema(cls, prompt_name: str) -> bool: """Check if a prompt has a JSON schema""" instance = cls() return prompt_name in instance.schemas @classmethod def get_prompt_with_schema(cls, prompt_name: str, context: Dict[str, Any] = None) -> Tuple[str, Optional[Dict[str, Any]]]: """ Get both the processed prompt and its schema (if available) Returns: Tuple of (prompt_string, schema_dict_or_None) """ prompt = cls.get_prompt(prompt_name, context) schema = cls.get_schema(prompt_name) return prompt, schema @classmethod def list_prompts(cls) -> Dict[str, Dict[str, Any]]: """ List all available prompts with their info Returns: Dictionary mapping prompt names to their info (variables, has_schema) """ instance = cls() result = {} for prompt_name in instance.prompts: result[prompt_name] = { 'variables': instance.prompt_variables[prompt_name], 'has_schema': prompt_name in instance.schemas, 'variable_count': len(instance.prompt_variables[prompt_name]) } return result @classmethod def reload_prompts(cls): """Reload all prompt templates and schemas (useful for development)""" if cls._instance: cls._instance._load_all_prompts() logger.info("Prompts and schemas reloaded") @classmethod def get_prompt_info(cls, prompt_name: str) -> Dict[str, Any]: """ Get detailed information about a specific prompt Returns: Dictionary with prompt template, schema, and required variables """ instance = cls() if prompt_name not in instance.prompts: raise ValueError(f"Prompt '{prompt_name}' not found") info = { 'name': prompt_name, 'template': instance.prompts[prompt_name], 'variables': instance.prompt_variables[prompt_name], 'variable_count': len(instance.prompt_variables[prompt_name]), 'has_schema': prompt_name in instance.schemas } if prompt_name in instance.schemas: info['schema'] = instance.schemas[prompt_name] return info # Updated convenience functions def get_prompt(prompt_name: str, context: Dict[str, Any] = None) -> str: """Convenience function to get a processed prompt""" return PromptManager.get_prompt(prompt_name, context) def get_prompt_with_schema(prompt_name: str, context: Dict[str, Any] = None) -> Tuple[str, Optional[Dict[str, Any]]]: """Convenience function to get prompt and schema together""" return PromptManager.get_prompt_with_schema(prompt_name, context) def get_schema(prompt_name: str) -> Optional[Dict[str, Any]]: """Convenience function to get just the schema""" return PromptManager.get_schema(prompt_name) def has_schema(prompt_name: str) -> bool: """Check if a prompt has structured output schema""" return PromptManager.has_schema(prompt_name)