Spaces:
Paused
Paused
File size: 5,758 Bytes
0d3f7cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """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
@property
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))
|