"""GEPA Self-Evolution pipeline — orchestrates the full evolution workflow. Usage: pipeline = EvolutionPipeline() result = await pipeline.run( skill_name="erp-integration", base_content="...", dataset_path="self-evolution/eval-datasets/erp-integration.json", ) """ from __future__ import annotations import json import logging from datetime import UTC, datetime from pathlib import Path from typing import Any from hermes.evolution.constraint_gates import ConstraintGates from hermes.evolution.eval_suite import EvalSuite from hermes.evolution.optimizer import GEPAOptimizer logger = logging.getLogger(__name__) class EvolutionPipeline: """Main GEPA evolution pipeline orchestrator. Runs the full evolution cycle: 1. Initialize population from base content 2. For each generation: evaluate, select, crossover, mutate 3. Apply constraint gates 4. Track best-performing variants 5. Return evolved content """ def __init__( self, evolution_rounds: int = 10, population_size: int = 20, mutation_rate: float = 0.15, crossover_rate: float = 0.7, elite_ratio: float = 0.2, early_stop_rounds: int = 3, output_dir: str = "self-evolution/evolved-skills", ) -> None: self.optimizer = GEPAOptimizer( population_size=population_size, mutation_rate=mutation_rate, crossover_rate=crossover_rate, elite_ratio=elite_ratio, early_stop_rounds=early_stop_rounds, ) self.constraint_gates = ConstraintGates() self.eval_suite = EvalSuite() self.output_dir = Path(output_dir) self.evolution_rounds = evolution_rounds self._history: list[dict[str, Any]] = [] async def run( self, skill_name: str, base_content: str, dataset_path: str | Path | None = None, examples: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """Run the full evolution pipeline for a skill.""" logger.info(f"Starting evolution for '{skill_name}'") if examples is None and dataset_path: examples = self.eval_suite.load_dataset(dataset_path) examples = examples or [] population = self.optimizer.initialize_population(base_content) logger.info(f"Initialized population with {len(population)} individuals") best_content = base_content best_score: float = 0.0 generation_results: list[dict[str, Any]] = [] for round_num in range(1, self.evolution_rounds + 1): logger.info(f"Evolution round {round_num}/{self.evolution_rounds}") if self.optimizer.should_stop(): logger.info(f"Early stopping at round {round_num} (no improvement)") break fitness_scores = await self._evaluate_population( population, skill_name, base_content, examples ) avg_fitness = sum(fitness_scores) / len(fitness_scores) if fitness_scores else 0.0 max_fitness = max(fitness_scores) if fitness_scores else 0.0 best_idx = fitness_scores.index(max_fitness) if fitness_scores else 0 round_best = population[best_idx] gates_result = await self.constraint_gates.check_all( original=base_content, evolved=round_best, test_cases=examples, ) round_record = { "round": round_num, "avg_fitness": avg_fitness, "max_fitness": max_fitness, "best_content": round_best[:200], "gates_passed": gates_result["passed"], "failed_gates": gates_result["failed_gates"], } generation_results.append(round_record) if max_fitness > best_score and gates_result["passed"]: best_score = max_fitness best_content = round_best population = self.optimizer.evolve(population, fitness_scores) final_eval = await self.eval_suite.evaluate( skill_name=skill_name, examples=examples or [], evolved_content=best_content, ) result = { "skill_name": skill_name, "status": "completed", "generations_run": self.optimizer.generation, "best_score": best_score, "improvement": best_score - self._baseline_score(examples), "final_evaluation": final_eval, "generation_results": generation_results, "evolved_content": best_content, "timestamp": datetime.now(UTC).isoformat(), } self._history.append(result) await self._save_result(skill_name, result) logger.info(f"Evolution completed for '{skill_name}' with score {best_score:.3f}") return result async def _evaluate_population( self, population: list[str], skill_name: str, base_content: str, examples: list[dict[str, Any]], ) -> list[float]: """Evaluate fitness of each individual in the population.""" fitness_scores: list[float] = [] for individual in population: gates_result = await self.constraint_gates.check_all( original=base_content, evolved=individual, test_cases=examples, ) if not gates_result["passed"]: fitness_scores.append(0.0) continue eval_result = await self.eval_suite.evaluate( skill_name=skill_name, examples=examples or [], evolved_content=individual, ) score = eval_result.get("overall_score", 0.0) fitness_scores.append(score) return fitness_scores def _baseline_score(self, examples: list[dict[str, Any]]) -> float: """Compute baseline score from example metrics.""" if not examples: return 0.0 scores = [] for ex in examples: metrics = ex.get("metrics", {}) overall = ( metrics.get("accuracy", 0) * 0.35 + (1.0 - metrics.get("efficiency", 5000) / 10000) * 0.20 + metrics.get("coherence", 0) * 0.20 + metrics.get("safety", 0) * 0.25 ) scores.append(overall) return sum(scores) / len(scores) if scores else 0.0 async def _save_result(self, skill_name: str, result: dict[str, Any]) -> None: """Save evolution result to disk.""" try: self.output_dir.mkdir(parents=True, exist_ok=True) path = self.output_dir / f"{skill_name}_v{self.optimizer.generation}.json" with open(path, "w", encoding="utf-8") as f: json.dump(result, f, indent=2, default=str) logger.info(f"Saved evolution result to {path}") except Exception as e: logger.warning(f"Could not save evolution result: {e}") def get_history(self) -> list[dict[str, Any]]: """Get evolution history.""" return list(self._history)