Global context manager for all prompts with version management.
| 23 | from src.dynamic import dynamic_manager |
| 24 | |
| 25 | class PromptContextManager(BaseModel): |
| 26 | """Global context manager for all prompts with version management.""" |
| 27 | model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow") |
| 28 | |
| 29 | base_dir: str = Field(default=None, description="The base directory to use for the prompts") |
| 30 | save_path: str = Field(default=None, description="The path to save the prompts") |
| 31 | contract_path: str = Field(default=None, description="The path to save the prompt contract") |
| 32 | |
| 33 | def __init__(self, |
| 34 | base_dir: Optional[str] = None, |
| 35 | save_path: Optional[str] = None, |
| 36 | contract_path: Optional[str] = None, |
| 37 | **kwargs): |
| 38 | """Initialize the prompt context manager. |
| 39 | |
| 40 | Args: |
| 41 | base_dir: Base directory for storing prompt data |
| 42 | save_path: Path to save prompt configurations |
| 43 | contract_path: Path to save prompt contract |
| 44 | """ |
| 45 | super().__init__(**kwargs) |
| 46 | |
| 47 | # Set up paths |
| 48 | if base_dir is not None: |
| 49 | self.base_dir = assemble_project_path(base_dir) |
| 50 | else: |
| 51 | self.base_dir = assemble_project_path(os.path.join(config.workdir, "prompt")) |
| 52 | os.makedirs(self.base_dir, exist_ok=True) |
| 53 | |
| 54 | if save_path is not None: |
| 55 | self.save_path = assemble_project_path(save_path) |
| 56 | else: |
| 57 | self.save_path = os.path.join(self.base_dir, "prompt.json") |
| 58 | |
| 59 | if contract_path is not None: |
| 60 | self.contract_path = assemble_project_path(contract_path) |
| 61 | else: |
| 62 | self.contract_path = os.path.join(self.base_dir, "contract.md") |
| 63 | |
| 64 | logger.info(f"| 📁 Prompt context manager base directory: {self.base_dir} and save path: {self.save_path}") |
| 65 | logger.info(f"| 📁 Prompt context manager contract path: {self.contract_path}") |
| 66 | |
| 67 | self._prompt_configs: Dict[str, PromptConfig] = {} # Current active configs (latest version) |
| 68 | # Prompt version history, e.g., {"prompt_name": {"1.0.0": PromptConfig, "1.0.1": PromptConfig}} |
| 69 | self._prompt_history_versions: Dict[str, Dict[str, PromptConfig]] = {} |
| 70 | |
| 71 | self._cleanup_registered = False |
| 72 | self._variables_lock = asyncio.Lock() # Lock for get/set trainable variables |
| 73 | |
| 74 | # Register cleanup on exit |
| 75 | if not self._cleanup_registered: |
| 76 | atexit.register(self.cleanup) |
| 77 | self._cleanup_registered = True |
| 78 | |
| 79 | async def initialize(self, prompt_names: Optional[List[str]] = None): |
| 80 | """Initialize the prompt context manager.""" |
| 81 | # Load prompts from PROMPT registry |
| 82 | prompt_configs = {} |