add bounded parallel self-hosting controls

Author: Darreckk

openevolve/api.py

@@ -8,7 +8,7 @@
 import uuid
 import inspect
 from typing import Union, Callable, Optional, List, Dict, Any, Tuple
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 
 from openevolve.controller import OpenEvolve
@@ -25,6 +25,10 @@ class EvolutionResult:
     best_code: str
     metrics: Dict[str, Any]
     output_dir: Optional[str]
+    completion_reason: str = "unknown"
+    last_completed_iteration: Optional[int] = None
+    completed_iteration_count: int = 0
+    llm_usage: Dict[str, Any] = field(default_factory=dict)
 
     def __repr__(self):
         return f"EvolutionResult(best_score={self.best_score:.4f})"
@@ -178,12 +182,17 @@ async def _run_evolution_async(
                 if numeric_metrics:
                     best_score = sum(numeric_metrics) / len(numeric_metrics)
 
+        llm_usage = getattr(controller, "llm_usage", {})
         return EvolutionResult(
             best_program=best_program,
             best_score=best_score,
             best_code=best_code,
             metrics=metrics,
             output_dir=actual_output_dir if not cleanup else None,
+            completion_reason=controller.completion_reason,
+            last_completed_iteration=controller.last_completed_iteration,
+            completed_iteration_count=controller.completed_iteration_count,
+            llm_usage=dict(llm_usage) if isinstance(llm_usage, dict) else {},
         )
 
     finally:

openevolve/config.py

@@ -4,6 +4,7 @@
 
 import os
 import re
+import math
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
@@ -306,6 +307,20 @@ class PromptConfig:
     )
 
 
+@dataclass
+class ControllerSchedulerConfig:
+    """Optional observed-result island allocator for bounded live comparisons."""
+
+    enabled: bool = False
+    exploitation_weight: float = 0.0
+    underexplored_weight: float = 0.0
+    validity_weight: float = 0.0
+    diversity_weight: float = 0.0
+    token_efficiency_weight: float = 0.0
+    rejection_penalty: float = 0.0
+    minimum_calls: int = 1
+
+
 @dataclass
 class DatabaseConfig:
     """Configuration for the program database"""
@@ -326,6 +341,9 @@ class DatabaseConfig:
     elite_selection_ratio: float = 0.1
     exploration_ratio: float = 0.2
     exploitation_ratio: float = 0.7
+    controller_scheduler: ControllerSchedulerConfig = field(
+        default_factory=ControllerSchedulerConfig
+    )
     # Note: diversity_metric fixed to "edit_distance"
     diversity_metric: str = "edit_distance"  # Options: "edit_distance", "feature_based"
 
@@ -418,6 +436,10 @@ class Config:
 
     # General settings
     max_iterations: int = 10000
+    # Optional per-run proposal-model budgets. Calls are reserved before
+    # submission; provider-token totals come from completed unique receipts.
+    max_llm_calls: Optional[int] = None
+    max_total_provider_tokens: Optional[int] = None
     checkpoint_interval: int = 100
     log_level: str = "INFO"
     log_dir: Optional[str] = None
@@ -436,6 +458,9 @@ class Config:
     diff_based_evolution: bool = True
     max_code_length: int = 10000
     diff_pattern: str = r"<<<<<<< SEARCH\n(.*?)=======\n(.*?)>>>>>>> REPLACE"
+    strict_diff_application: bool = False
+    enforce_evolve_blocks: bool = False
+    max_diff_blocks: int = 32
 
     # Early stopping settings
     early_stopping_patience: Optional[int] = None
@@ -489,13 +514,67 @@ def from_dict(cls, config_dict: Dict[str, Any]) -> "Config":
         if config.database.random_seed is None and config.random_seed is not None:
             config.database.random_seed = config.random_seed
 
-        if config.prompt.programs_as_changes_description and not config.diff_based_evolution:
+        config.validate()
+        return config
+
+    def validate(self) -> None:
+        """Validate combinations used by both YAML and programmatic callers."""
+        for field_name in ("max_llm_calls", "max_total_provider_tokens"):
+            value = getattr(self, field_name)
+            if value is not None and (
+                isinstance(value, bool) or not isinstance(value, int) or value < 1
+            ):
+                raise ValueError(f"{field_name} must be a positive integer or None")
+        try:
+            re.compile(self.diff_pattern)
+        except re.error as error:
+            raise ValueError(f"Invalid regex pattern in diff_pattern: {error}") from error
+        if self.prompt.programs_as_changes_description and not self.diff_based_evolution:
             raise ValueError(
                 "prompt.programs_as_changes_description=true requires diff_based_evolution=true "
                 "(full rewrites cannot reliably update code and changes_description together)"
             )
-
-        return config
+        if self.enforce_evolve_blocks and not self.strict_diff_application:
+            raise ValueError(
+                "enforce_evolve_blocks=true requires strict_diff_application=true"
+            )
+        if self.enforce_evolve_blocks and not self.diff_based_evolution:
+            raise ValueError(
+                "enforce_evolve_blocks=true requires diff_based_evolution=true"
+            )
+        if self.max_diff_blocks < 1:
+            raise ValueError("max_diff_blocks must be at least 1")
+        scheduler = self.database.controller_scheduler
+        if not isinstance(scheduler.enabled, bool):
+            raise ValueError("database.controller_scheduler.enabled must be boolean")
+        for field_name in (
+            "exploitation_weight",
+            "underexplored_weight",
+            "validity_weight",
+            "diversity_weight",
+            "token_efficiency_weight",
+            "rejection_penalty",
+        ):
+            value = getattr(scheduler, field_name)
+            if (
+                isinstance(value, bool)
+                or not isinstance(value, (int, float))
+                or not math.isfinite(float(value))
+                or value < 0
+            ):
+                raise ValueError(
+                    f"database.controller_scheduler.{field_name} "
+                    "must be finite and nonnegative"
+                )
+        if (
+            isinstance(scheduler.minimum_calls, bool)
+            or not isinstance(scheduler.minimum_calls, int)
+            or scheduler.minimum_calls < 1
+        ):
+            raise ValueError(
+                "database.controller_scheduler.minimum_calls "
+                "must be a positive integer"
+            )
 
     def to_dict(self) -> Dict[str, Any]:
         return asdict(self)

openevolve/controller.py

@@ -48,6 +48,7 @@ def __init__(
     ):
         # Load configuration (loaded in main_async)
         self.config = config
+        self.config.validate()
 
         # Set up output directory
         self.output_dir = output_dir or os.path.join(
@@ -163,6 +164,12 @@ def __init__(
 
         # Initialize improved parallel processing components
         self.parallel_controller = None
+        # Preserve authoritative run completion metadata after the process
+        # controller is stopped and released.
+        self.completion_reason = "not_started"
+        self.last_completed_iteration: Optional[int] = None
+        self.completed_iteration_count = 0
+        self.llm_usage: Dict[str, Any] = {}
 
     def _setup_logging(self) -> None:
         """Set up logging"""
@@ -314,6 +321,7 @@ async def run(
                 self.database,
                 self.evolution_tracer,
                 file_suffix=self.config.file_suffix,
+                usage_output_path=os.path.join(self.output_dir, "llm_usage.jsonl"),
             )
 
             # Set up signal handlers for graceful shutdown
@@ -354,6 +362,15 @@ def force_exit_handler(signum, frame):
         finally:
             # Clean up parallel processing resources
             if self.parallel_controller:
+                self.completion_reason = self.parallel_controller.completion_reason
+                self.last_completed_iteration = (
+                    self.parallel_controller.last_completed_iteration
+                )
+                self.completed_iteration_count = (
+                    self.parallel_controller.completed_iteration_count
+                )
+                llm_usage = getattr(self.parallel_controller, "llm_usage", {})
+                self.llm_usage = dict(llm_usage) if isinstance(llm_usage, dict) else {}
                 self.parallel_controller.stop()
                 self.parallel_controller = None
 
@@ -501,15 +518,36 @@ async def _run_evolution_with_checkpoints(
         if self.parallel_controller.shutdown_event.is_set():
             logger.info("Evolution stopped due to shutdown request")
             return
-        elif self.parallel_controller.early_stopping_triggered:
+        elif getattr(self.parallel_controller, "early_stopping_triggered", False) is True:
             logger.info("Evolution stopped due to early stopping - saving final checkpoint")
             # Continue to save final checkpoint for early stopping
 
-        # Save final checkpoint if needed
-        # Note: start_iteration here is the evolution start (1 for fresh start, not 0)
-        # max_iterations is the number of evolution iterations to run
-        final_iteration = start_iteration + max_iterations - 1
-        if final_iteration > 0 and final_iteration % self.config.checkpoint_interval == 0:
+        # Bind the final checkpoint to work that actually completed. A target
+        # reached by one worker can leave higher-numbered work in flight, so
+        # the requested budget is not a valid completion cursor.
+        final_iteration = getattr(
+            self.parallel_controller, "last_completed_iteration", None
+        )
+        target_score_reached = (
+            getattr(self.parallel_controller, "target_score_reached", False) is True
+        )
+        early_stopping_triggered = (
+            getattr(self.parallel_controller, "early_stopping_triggered", False) is True
+        )
+        budget_triggered = bool(
+            getattr(self.parallel_controller, "budget_completion_reason", None)
+        )
+        should_save_final = (
+            isinstance(final_iteration, int)
+            and final_iteration > 0
+            and (
+                target_score_reached
+                or early_stopping_triggered
+                or budget_triggered
+                or final_iteration % self.config.checkpoint_interval == 0
+            )
+        )
+        if should_save_final:
             self._save_checkpoint(final_iteration)
 
     def _save_best_program(self, program: Optional[Program] = None) -> None:

openevolve/evaluation_result.py

@@ -7,6 +7,9 @@
 from typing import Dict, Union
 
 
+EVALUATION_FAILED_METRIC = "__evaluation_failed__"
+
+
 @dataclass
 class EvaluationResult:
     """

openevolve/evaluator.py

@@ -19,7 +19,10 @@
 
 from openevolve.config import EvaluatorConfig
 from openevolve.database import ProgramDatabase
-from openevolve.evaluation_result import EvaluationResult
+from openevolve.evaluation_result import (
+    EVALUATION_FAILED_METRIC,
+    EvaluationResult,
+)
 from openevolve.database import ProgramDatabase
 from openevolve.llm.ensemble import LLMEnsemble
 from openevolve.utils.async_utils import TaskPool, run_in_executor
@@ -262,7 +265,11 @@ async def evaluate_program(
                         "error_type": "timeout",
                     }
 
-                return {"error": 0.0, "timeout": True}
+                return {
+                    EVALUATION_FAILED_METRIC: 1.0,
+                    "error": 0.0,
+                    "timeout": True,
+                }
 
             except Exception as e:
                 last_exception = e
@@ -293,7 +300,7 @@ async def evaluate_program(
         logger.error(
             f"All evaluation attempts failed for program{program_id_str}. Last error: {str(last_exception)}"
         )
-        return {"error": 0.0}
+        return {EVALUATION_FAILED_METRIC: 1.0, "error": 0.0}
 
     def _process_evaluation_result(self, result: Any) -> EvaluationResult:
         """
@@ -314,7 +321,9 @@ def _process_evaluation_result(self, result: Any) -> EvaluationResult:
         else:
             # Error case - return error metrics
             logger.warning(f"Unexpected evaluation result type: {type(result)}")
-            return EvaluationResult(metrics={"error": 0.0})
+            return EvaluationResult(
+                metrics={EVALUATION_FAILED_METRIC: 1.0, "error": 0.0}
+            )
 
     def get_pending_artifacts(self, program_id: str) -> Optional[Dict[str, Union[str, bytes]]]:
         """

openevolve/llm/ensemble.py

@@ -38,6 +38,7 @@ class LLMEnsemble:
 
     def __init__(self, models_cfg: List[LLMModelConfig]):
         self.models_cfg = models_cfg
+        self.last_call_metadata: Dict[str, object] = {}
 
         # Initialize models from the configuration
         self.models = [_create_model(model_cfg) for model_cfg in models_cfg]
@@ -81,7 +82,9 @@ async def generate_with_context(
     ) -> str:
         """Generate text using a system message and conversational context"""
         model = self._sample_model()
-        return await model.generate_with_context(system_message, messages, **kwargs)
+        response = await model.generate_with_context(system_message, messages, **kwargs)
+        self.last_call_metadata = dict(getattr(model, "last_call_metadata", {}) or {})
+        return response
 
     def _sample_model(self) -> LLMInterface:
         """Sample a model from the ensemble based on weights"""

openevolve/llm/openai.py

@@ -21,6 +21,16 @@
 logger = logging.getLogger(__name__)
 
 
+def _uses_provider_managed_sampling(api_base: str | None, model: str | None) -> bool:
+    """Whether this provider/model pair requires sampling knobs to be omitted."""
+    base = str(api_base or "").rstrip("/")
+    name = str(model or "").lower()
+    return (
+        base.startswith("https://generativelanguage.googleapis.com/")
+        and name.startswith(("gemini-3.5-", "gemini-3.6-"))
+    )
+
+
 def _iso_now() -> str:
     return datetime.now(tz=timezone.utc).isoformat()
 
@@ -63,6 +73,7 @@ def __init__(
         self.api_key = model_cfg.api_key
         self.random_seed = getattr(model_cfg, "random_seed", None)
         self.reasoning_effort = getattr(model_cfg, "reasoning_effort", None)
+        self.last_call_metadata: Dict[str, Any] = {}
 
         # Manual mode: enabled via llm.manual_mode in config.yaml
         self.manual_mode = (getattr(model_cfg, "manual_mode", False) is True)
@@ -154,12 +165,15 @@ async def generate_with_context(
             params = {
                 "model": self.model,
                 "messages": formatted_messages,
-                "temperature": kwargs.get("temperature", self.temperature),
                 "max_tokens": kwargs.get("max_tokens", self.max_tokens),
             }
-            top_p = kwargs.get("top_p", self.top_p)
-            if top_p is not None:
-                params["top_p"] = top_p
+            if not _uses_provider_managed_sampling(self.api_base, self.model):
+                temperature = kwargs.get("temperature", self.temperature)
+                if temperature is not None:
+                    params["temperature"] = temperature
+                top_p = kwargs.get("top_p", self.top_p)
+                if top_p is not None:
+                    params["top_p"] = top_p
 
             # Handle reasoning_effort for open source reasoning models.
             reasoning_effort = kwargs.get("reasoning_effort", self.reasoning_effort)
@@ -221,6 +235,22 @@ async def _call_api(self, params: Dict[str, Any]) -> str:
         response = await loop.run_in_executor(
             None, lambda: self.client.chat.completions.create(**params)
         )
+        usage = getattr(response, "usage", None)
+        usage_details = (
+            usage.model_dump(mode="json")
+            if usage is not None and hasattr(usage, "model_dump")
+            else {}
+        )
+        self.last_call_metadata = {
+            "provider_response_id": getattr(response, "id", None),
+            "model": getattr(response, "model", None) or self.model,
+            "usage": {
+                "prompt_tokens": getattr(usage, "prompt_tokens", None),
+                "completion_tokens": getattr(usage, "completion_tokens", None),
+                "total_tokens": getattr(usage, "total_tokens", None),
+                "details": usage_details,
+            },
+        }
         # Logging of system prompt, user message and response content
         logger = logging.getLogger(__name__)
         logger.debug(f"API parameters: {params}")