Spaces:
Paused
Paused
| """GEPA optimizer — Genetic algorithm with Pareto optimization for skill evolution.""" | |
| from __future__ import annotations | |
| import logging | |
| import random | |
| logger = logging.getLogger(__name__) | |
| class GEPAOptimizer: | |
| """Genetic-Pareto Prompt Evolution optimizer. | |
| Uses tournament selection, crossover, mutation, and Pareto-based | |
| multi-objective optimization to evolve skills, tool descriptions, | |
| and system prompts. | |
| """ | |
| def __init__( | |
| self, | |
| population_size: int = 20, | |
| tournament_size: int = 3, | |
| mutation_rate: float = 0.15, | |
| crossover_rate: float = 0.7, | |
| elite_ratio: float = 0.2, | |
| early_stop_rounds: int = 3, | |
| ) -> None: | |
| self.population_size = population_size | |
| self.tournament_size = tournament_size | |
| self.mutation_rate = mutation_rate | |
| self.crossover_rate = crossover_rate | |
| self.elite_ratio = elite_ratio | |
| self.early_stop_rounds = early_stop_rounds | |
| self._generation = 0 | |
| self._no_improvement = 0 | |
| self._best_fitness: float | None = None | |
| def initialize_population(self, base_prompt: str, num_variants: int | None = None) -> list[str]: | |
| """Create initial population from a base prompt with random variants.""" | |
| size = num_variants or self.population_size | |
| population = [base_prompt] | |
| for _ in range(size - 1): | |
| variant = self._create_variant(base_prompt) | |
| population.append(variant) | |
| return population | |
| def tournament_select(self, population: list[str], fitness_scores: list[float]) -> str: | |
| """Select an individual using tournament selection.""" | |
| indices = random.sample(range(len(population)), self.tournament_size) | |
| best_idx = max(indices, key=lambda i: fitness_scores[i]) | |
| return population[best_idx] | |
| def crossover(self, parent1: str, parent2: str) -> tuple[str, str]: | |
| """Perform single-point crossover between two parents.""" | |
| if random.random() > self.crossover_rate: | |
| return parent1, parent2 | |
| words1 = parent1.split() | |
| words2 = parent2.split() | |
| if len(words1) < 3 or len(words2) < 3: | |
| return parent1, parent2 | |
| point1 = random.randint(1, len(words1) - 1) | |
| point2 = random.randint(1, len(words2) - 1) | |
| child1_words = words1[:point1] + words2[point2:] | |
| child2_words = words2[:point2] + words1[point1:] | |
| return " ".join(child1_words), " ".join(child2_words) | |
| def mutate(self, individual: str) -> str: | |
| """Mutate an individual by replacing random words.""" | |
| if random.random() > self.mutation_rate: | |
| return individual | |
| words = individual.split() | |
| if len(words) < 5: | |
| return individual | |
| num_mutations = max(1, len(words) // 20) | |
| synonyms = { | |
| "execute": ["run", "perform", "carry out", "invoke"], | |
| "create": ["generate", "build", "construct", "produce"], | |
| "search": ["find", "look up", "query", "retrieve"], | |
| "analyze": ["examine", "inspect", "review", "evaluate"], | |
| "process": ["handle", "manage", "deal with", "work on"], | |
| "return": ["provide", "give back", "output", "deliver"], | |
| "validate": ["verify", "check", "confirm", "ensure"], | |
| "implement": ["build", "develop", "code", "realize"], | |
| } | |
| for _ in range(num_mutations): | |
| idx = random.randint(0, len(words) - 1) | |
| word = words[idx].lower().strip(",.!?;:") | |
| if word in synonyms: | |
| words[idx] = random.choice(synonyms[word]) | |
| return " ".join(words) | |
| def evolve( | |
| self, | |
| population: list[str], | |
| fitness_scores: list[float], | |
| ) -> list[str]: | |
| """Run one generation of evolution.""" | |
| self._generation += 1 | |
| population_size = len(population) | |
| elite_count = max(1, int(population_size * self.elite_ratio)) | |
| elite_indices = sorted( | |
| range(len(fitness_scores)), | |
| key=lambda i: fitness_scores[i], | |
| reverse=True, | |
| )[:elite_count] | |
| elites = [population[i] for i in elite_indices] | |
| best = max(fitness_scores) | |
| if self._best_fitness is not None: | |
| if best <= self._best_fitness: | |
| self._no_improvement += 1 | |
| else: | |
| self._no_improvement = 0 | |
| self._best_fitness = best | |
| new_population = list(elites) | |
| while len(new_population) < population_size: | |
| parent1 = self.tournament_select(population, fitness_scores) | |
| parent2 = self.tournament_select(population, fitness_scores) | |
| child1, child2 = self.crossover(parent1, parent2) | |
| child1 = self.mutate(child1) | |
| child2 = self.mutate(child2) | |
| new_population.append(child1) | |
| if len(new_population) < population_size: | |
| new_population.append(child2) | |
| return new_population[:population_size] | |
| def should_stop(self) -> bool: | |
| """Check if evolution should stop.""" | |
| if self._best_fitness is None: | |
| return False | |
| if self._generation <= 1: | |
| return False | |
| return self._no_improvement >= self.early_stop_rounds | |
| def generation(self) -> int: | |
| """Current generation number.""" | |
| return self._generation | |
| def _create_variant(self, text: str) -> str: | |
| """Create a variant of text by rephrasing.""" | |
| sentences = text.replace("! ", ".\n").replace("? ", ".\n").split("\n") | |
| if len(sentences) <= 1: | |
| return self.mutate(text) | |
| random.shuffle(sentences) | |
| return self.mutate(" ".join(sentences)) | |