255 lines
8.7 KiB
Python
255 lines
8.7 KiB
Python
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)
|