Initialize the prompt context manager.
(self, prompt_names: Optional[List[str]] = None)
| 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 = {} |
| 83 | registry_prompt_configs: Dict[str, PromptConfig] = await self._load_from_registry() |
| 84 | prompt_configs.update(registry_prompt_configs) |
| 85 | |
| 86 | # Load prompts from JSON file |
| 87 | code_prompt_configs: Dict[str, PromptConfig] = await self._load_from_code() |
| 88 | |
| 89 | # Merge code configs with registry configs, only override if code version is strictly greater |
| 90 | for prompt_name, code_config in code_prompt_configs.items(): |
| 91 | if prompt_name in prompt_configs: |
| 92 | registry_config = prompt_configs[prompt_name] |
| 93 | # Compare versions: only override if code version is strictly greater |
| 94 | if version_manager.compare_versions(code_config.version, registry_config.version) > 0: |
| 95 | logger.info(f"| 🔄 Overriding prompt {prompt_name} from registry (v{registry_config.version}) with code version (v{code_config.version})") |
| 96 | prompt_configs[prompt_name] = code_config |
| 97 | else: |
| 98 | logger.info(f"| 📌 Keeping prompt {prompt_name} from registry (v{registry_config.version}), code version (v{code_config.version}) is not greater") |
| 99 | # If versions are equal, update the history with registry config (which has real class, not dynamic) |
| 100 | if version_manager.compare_versions(code_config.version, registry_config.version) == 0: |
| 101 | # Replace the code config in history with registry config to preserve real class reference |
| 102 | if prompt_name in self._prompt_history_versions: |
| 103 | self._prompt_history_versions[prompt_name][registry_config.version] = registry_config |
| 104 | else: |
| 105 | # New prompt from code, add it |
| 106 | prompt_configs[prompt_name] = code_config |
| 107 | |
| 108 | # Filter prompts by names if provided |
| 109 | # prompt_names are base names like "tool_calling", need to match both system_prompt and agent_message_prompt |
| 110 | if prompt_names is not None: |
| 111 | filtered_configs = {} |
| 112 | for base_name in prompt_names: |
| 113 | # Match system_prompt |
| 114 | system_prompt_name = f"{base_name}_system_prompt" |
| 115 | if system_prompt_name in prompt_configs: |
| 116 | filtered_configs[system_prompt_name] = prompt_configs[system_prompt_name] |
| 117 | # Match agent_message_prompt |
| 118 | agent_message_prompt_name = f"{base_name}_agent_message_prompt" |
| 119 | if agent_message_prompt_name in prompt_configs: |
| 120 | filtered_configs[agent_message_prompt_name] = prompt_configs[agent_message_prompt_name] |
| 121 | prompt_configs = filtered_configs |
| 122 | |
| 123 | # Store all prompts |
| 124 | for prompt_name, prompt_config in prompt_configs.items(): |
| 125 | self._prompt_configs[prompt_name] = prompt_config |
| 126 | |
| 127 | # Build all prompts concurrently with a concurrency limit |
| 128 | prompt_names = list(prompt_configs.keys()) |
| 129 | tasks = [ |
| 130 | self.build(prompt_configs[name]) for name in prompt_names |
| 131 | ] |
| 132 | results = await gather_with_concurrency(tasks, max_concurrency=10, return_exceptions=True) |
| 133 | |
| 134 | for prompt_name, result in zip(prompt_names, results): |
| 135 | if isinstance(result, Exception): |
| 136 | logger.error(f"| ❌ Failed to initialize prompt {prompt_name}: {result}") |
no test coverage detected