Initialize prompts by names using prompt context manager with concurrent support. Args: prompt_names: List of prompt names to initialize. If None, initialize all registered prompts.
(self, prompt_names: Optional[List[str]] = None)
| 29 | self._registered_configs: Dict[str, PromptConfig] = {} # prompt_name -> PromptConfig |
| 30 | |
| 31 | async def initialize(self, prompt_names: Optional[List[str]] = None): |
| 32 | """Initialize prompts by names using prompt context manager with concurrent support. |
| 33 | |
| 34 | Args: |
| 35 | prompt_names: List of prompt names to initialize. If None, initialize all registered prompts. |
| 36 | """ |
| 37 | self.base_dir = assemble_project_path(os.path.join(config.workdir, "prompt")) |
| 38 | os.makedirs(self.base_dir, exist_ok=True) |
| 39 | self.save_path = os.path.join(self.base_dir, "prompt.json") |
| 40 | self.contract_path = os.path.join(self.base_dir, "contract.md") |
| 41 | logger.info(f"| 📁 Prompt Manager base directory: {self.base_dir} with save path: {self.save_path} and contract path: {self.contract_path}") |
| 42 | |
| 43 | # Initialize prompt context manager |
| 44 | self.prompt_context_manager = PromptContextManager( |
| 45 | base_dir=self.base_dir, |
| 46 | save_path=self.save_path, |
| 47 | contract_path=self.contract_path, |
| 48 | ) |
| 49 | await self.prompt_context_manager.initialize(prompt_names=prompt_names) |
| 50 | |
| 51 | logger.info("| ✅ Prompts initialization completed") |
| 52 | |
| 53 | async def register(self, prompt: Union[Prompt, Dict[str, Any]], *, override: bool = False, **kwargs: Any) -> PromptConfig: |
| 54 | """Register a prompt class or template dictionary asynchronously. |
no test coverage detected