Spaces:
Paused
Paused
| """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) | |