Database for storing and sampling programs during evolution The database implements a combination of MAP-Elites algorithm and island-based population model to maintain diversity during evolution. It also tracks the absolute best program separately to ensure it's never lost.
| 116 | |
| 117 | |
| 118 | class ProgramDatabase: |
| 119 | """ |
| 120 | Database for storing and sampling programs during evolution |
| 121 | |
| 122 | The database implements a combination of MAP-Elites algorithm and |
| 123 | island-based population model to maintain diversity during evolution. |
| 124 | It also tracks the absolute best program separately to ensure it's never lost. |
| 125 | """ |
| 126 | |
| 127 | def __init__(self, config: DatabaseConfig): |
| 128 | # Store configuration |
| 129 | self.config = config |
| 130 | |
| 131 | # In-memory program storage: ID -> Program object |
| 132 | self.programs: Dict[str, Program] = {} |
| 133 | |
| 134 | # Per-island feature grids for MAP-Elites (Key -> Program ID) |
| 135 | self.island_feature_maps: List[Dict[str, str]] = [{} for _ in range(config.num_islands)] |
| 136 | |
| 137 | # Handle both int and dict types for feature_bins configuration |
| 138 | if isinstance(config.feature_bins, int): |
| 139 | # Calculate dynamic bin size if int provided |
| 140 | self.feature_bins = max( |
| 141 | config.feature_bins, |
| 142 | int(pow(config.archive_size, 1 / len(config.feature_dimensions)) + 0.99), |
| 143 | ) |
| 144 | else: |
| 145 | # If dict, keep as is (we'll use feature_bins_per_dim instead) |
| 146 | self.feature_bins = 10 # Default fallback for backward compatibility |
| 147 | |
| 148 | # Island populations: Sets of Program IDs |
| 149 | self.islands: List[Set[str]] = [set() for _ in range(config.num_islands)] |
| 150 | |
| 151 | # Island management attributes |
| 152 | self.current_island: int = 0 |
| 153 | self.island_generations: List[int] = [0] * config.num_islands |
| 154 | self.last_migration_generation: int = 0 |
| 155 | self.migration_interval: int = getattr(config, "migration_interval", 10) # Default to 10 |
| 156 | self.migration_rate: float = getattr(config, "migration_rate", 0.1) # Default to 0.1 |
| 157 | |
| 158 | # Archive of elite programs (separate from islands, for global elites) |
| 159 | self.archive: Set[str] = set() |
| 160 | |
| 161 | # Track the absolute best program separately |
| 162 | self.best_program_id: Optional[str] = None |
| 163 | |
| 164 | # Track best program per island for proper island-based evolution |
| 165 | self.island_best_programs: List[Optional[str]] = [None] * config.num_islands |
| 166 | |
| 167 | # Track the last iteration number (for resuming) |
| 168 | self.last_iteration: int = 0 |
| 169 | |
| 170 | # Load database from disk if path is provided and exists |
| 171 | if config.db_path and os.path.exists(config.db_path): |
| 172 | self.load(config.db_path) |
| 173 | |
| 174 | # Prompt log storage |
| 175 | self.prompts_by_program: Dict[str, Dict[str, Dict[str, str]]] = None |