Evaluator

Evaluator

Evaluator(
    env: list[VectorEnv] | list[VecEnv] | list[Env] | Env,
    log_dir: Path | str | PathLike | None = "logs",
    log_subdir: str = "eval",
    log_level: int = INFO,
    *,
    colorize_logs: bool = False,
)

Evaluates a policy on registered Property instances.

The evaluator repeatedly rolls out a policy on one or more (vectorized) environments, feeds each resulting trajectory to every registered property, and updates the property's statistical confidence interval. Evaluation continues until all properties have converged (when stop_on_convergence=True) or a resource limit (episode_limit / time_limit) is reached.

Typical usage:

evaluator = Evaluator(env=envs)
evaluator.register_property(my_property)
results = evaluator.eval(agent=agent, stop_on_convergence=True)

Attributes:

Name	Type	Description
envs	list[Any]	The (list of) vectorized environments used for evaluation.
time_passed	float	Wall-clock seconds spent in the most recent evaluation.
log_dir	Path | None	Directory of the current run, or None when logging is disabled.
properties	list[Property]	List of registered Property instances.

Create an evaluator over one or more environments.

The environment(s) may be a single Gymnasium environment, a single vectorized environment, or a list of vectorized environments (one per worker thread). Both Gymnasium (gym.vector.VectorEnv) and Stable-Baselines3 (VecEnv) vectorized environments are supported. A non-vectorized environment is automatically wrapped in an AsyncVectorEnv. Use create_eval_envs to build environments in the expected format.

Parameters:

Name	Type	Description	Default
env	list[VectorEnv] | list[VecEnv] | list[Env] | Env	The evaluation environment(s). A single (vectorized) environment, or a list of vectorized environments — one per thread used during eval.	required
log_dir	Path | str | PathLike | None	Base directory for run logs. Each eval call creates a numbered subdirectory under it. Pass None to disable all file logging (nothing is written to disk).	'logs'
log_subdir	str	Prefix for the per-run subdirectory (e.g. eval_0, eval_1, ...).	'eval'
log_level	int	Logging verbosity, as a logging level (e.g. logging.INFO).	INFO
colorize_logs	bool	Whether to colorize console log output.	False

Raises:

Type	Description
ValueError	If the environment is not a vectorized Gymnasium or Stable-Baselines3 environment, or uses an unsupported autoreset mode (only NextStep is supported).

Source code in pydsmc/evaluator.py

def __init__(
    self,
    env: list[gym.vector.VectorEnv] | list[SB3VecEnv] | list[GymEnv] | GymEnv,
    log_dir: Path | str | os.PathLike | None = "logs",
    log_subdir: str = "eval",
    log_level: int = logging.INFO,
    *,
    colorize_logs: bool = False,
) -> None:
    """Create an evaluator over one or more environments.

    The environment(s) may be a single Gymnasium environment, a single vectorized environment,
    or a list of vectorized environments (one per worker thread). Both Gymnasium
    (`gym.vector.VectorEnv`) and Stable-Baselines3 (`VecEnv`) vectorized environments are
    supported. A non-vectorized environment is automatically wrapped in an `AsyncVectorEnv`.
    Use [`create_eval_envs`][pydsmc.create_eval_envs] to build environments in the expected
    format.

    Args:
        env: The evaluation environment(s). A single (vectorized) environment, or a list of
            vectorized environments — one per thread used during [`eval`][pydsmc.Evaluator.eval].
        log_dir: Base directory for run logs. Each `eval` call creates a numbered subdirectory
            under it. Pass `None` to disable all file logging (nothing is written to disk).
        log_subdir: Prefix for the per-run subdirectory (e.g. `eval_0`, `eval_1`, ...).
        log_level: Logging verbosity, as a `logging` level (e.g. `logging.INFO`).
        colorize_logs: Whether to colorize console log output.

    Raises:
        ValueError: If the environment is not a vectorized Gymnasium or Stable-Baselines3
            environment, or uses an unsupported autoreset mode (only `NextStep` is supported).
    """
    self.log_base = Path(log_dir) if log_dir is not None else None
    self.log_dir = None  # Will be set once eval starts
    if self.log_base is not None:
        self.log_base.mkdir(exist_ok=True, parents=True)
    self.log_subdir = log_subdir
    self.properties: list[Property] = []
    self.eval_params: dict[str, Any] = {}
    self.total_episodes = 0
    self.time_passed = 0
    self.next_log_time = 0 if log_dir is not None else np.inf

    # For parallel episode execution
    self.lock = threading.Lock()
    self.thread_local = threading.local()
    self.next_free_env = 0

    self.logger = DSMCLogger.get_logger()
    DSMCLogger.set_colorize(colorize=colorize_logs)
    self.logger.setLevel(log_level)

    if not isinstance(env, list):
        # Single (vectorized) environment
        if hasattr(env, "num_envs") or hasattr(env, "n_envs"):
            self.envs = [env]  # Already vectorized, just not a list
        else:
            self.envs = [gym.vector.AsyncVectorEnv([lambda: env])]  # Not vectorized, nor list

    elif not (hasattr(env[0], "num_envs") or hasattr(env[0], "n_envs")):
        self.envs = [
            gym.vector.AsyncVectorEnv([lambda e=e: e]) for e in env
        ]  # List but not vectorized
    else:
        self.envs = env

    # earlier versions used n_envs. So we'd enforce a more recent version here otherwise
    # Support _some_ backwards compatibility at least
    if hasattr(self.envs[0], "n_envs"):
        for e in self.envs:
            e.num_envs = e.n_envs

    if not hasattr(self.envs[0], "num_envs"):
        raise ValueError(
            "Environment must be a vectorized gymnasium or stable_baselines3 environment.",
        )

    self.gym_vecenv = isinstance(self.envs[0], gym.vector.VectorEnv)
    self.gym_version_lt_1 = Version(gym.__version__) < Version("1.0.0")

    if self.gym_vecenv and hasattr(self.envs[0], "autoreset_mode"):
        from gymnasium.vector.vector_env import AutoresetMode

        if any(env.autoreset_mode != AutoresetMode.NEXT_STEP for env in self.envs):
            raise ValueError("Only the default `NextStep` autoreset mode is supported.")

num_envs

num_envs() -> int

Number of parallel environments per (vectorized) environment / thread.

Source code in pydsmc/evaluator.py

@property
def num_envs(self) -> int:
    """Number of parallel environments per (vectorized) environment / thread."""
    return self.envs[0].num_envs

register_property

register_property(property_: Property) -> None

Register a property to be evaluated on the next eval call.

Parameters:

Name	Type	Description	Default
property_	Property	The property to evaluate. It is assigned a fresh property ID on registration.	required

Source code in pydsmc/evaluator.py

def register_property(self, property_: Property) -> None:
    """Register a property to be evaluated on the next [`eval`][pydsmc.Evaluator.eval] call.

    Args:
        property_: The property to evaluate. It is assigned a fresh property ID on registration.
    """
    property_.set_property_id()
    self.properties.append(property_)

register_properties

register_properties(properties: Iterable[Property]) -> None

Register several properties at once.

Parameters:

Name	Type	Description	Default
properties	Iterable[Property]	An iterable of properties to register.	required

Source code in pydsmc/evaluator.py

def register_properties(self, properties: Iterable[Property]) -> None:
    """Register several properties at once.

    Args:
        properties: An iterable of properties to register.
    """
    for prop in properties:
        self.register_property(prop)

eval

eval(
    agent: Any = None,
    predict_fn: Callable | None = None,
    episode_limit: int | None = None,
    time_limit: float | None = None,
    num_initial_episodes: int = 100,
    num_episodes_per_policy_run: int = 50,
    save_every_n_episodes: int | None = None,
    *,
    stop_on_convergence: bool = True,
    save_full_results: bool = False,
    save_full_trajectory: bool = False,
    num_threads: int | None = None,
    extra_log_info: Any = None,
    reset_hidden_states: tuple[ndarray, ...] | None = None,
    **predict_kwargs: Any,
) -> list[Property]

Evaluate the policy on all registered properties until convergence or a resource limit.

Provide the policy either as an agent exposing a predict(obs, ...) method (e.g. any Stable-Baselines3 model) or directly as a predict_fn callable. Extra keyword arguments are forwarded to the policy (e.g. deterministic=True).

At least one stopping criterion must be active: episode_limit, time_limit, or stop_on_convergence=True together with at least one property that has an epsilon.

Parameters:

Name	Type	Description	Default
agent	Any	An object with a predict method. Used only when predict_fn is None.	None
predict_fn	Callable | None	A callable predict_fn(obs, hidden_states, episode_starts, **kwargs) returning (actions, hidden_states) for a batch of observations. Takes precedence over agent.	None
episode_limit	int | None	Maximum number of episodes to sample before stopping. None for no limit.	None
time_limit	float | None	Maximum wall-clock time in minutes before stopping. None for no limit.	None
num_initial_episodes	int	Number of episodes sampled in the first policy run.	100
num_episodes_per_policy_run	int	Number of episodes sampled per subsequent policy run (i.e. between convergence checks).	50
save_every_n_episodes	int | None	Persist intermediate results every this many episodes. None uses num_episodes_per_policy_run.	None
stop_on_convergence	bool	Whether to stop once all properties with an epsilon converged.	True
save_full_results	bool	Whether to store every individual sample.	False
save_full_trajectory	bool	Whether to store full trajectories to disk. Slow and disk-heavy; not recommended.	False
num_threads	int | None	Number of worker threads. Defaults to the number of environments. Must not exceed the number of environments.	None
extra_log_info	Any	Arbitrary extra metadata stored alongside the run settings.	None
**predict_kwargs	Any	Extra keyword arguments forwarded to agent.predict / predict_fn.	{}

Returns:

Type	Description
list[Property]	The list of registered Property instances, each holding its final estimate and confidence interval.

Raises:

Type	Description
ValueError	If arguments are invalid (e.g. no stopping criterion, no registered properties, num_threads < 1, or a non-positive time_limit).
TypeError	If neither a callable predict_fn nor an agent with a predict method is given.

Source code in pydsmc/evaluator.py

def eval(
    self,
    agent: Any = None,  # Allow None, since agent is _ONLY_ necessary if predict_fn is None
    predict_fn: Callable | None = None,
    episode_limit: int | None = None,
    time_limit: float | None = None,
    num_initial_episodes: int = 100,
    num_episodes_per_policy_run: int = 50,
    save_every_n_episodes: int | None = None,
    *,
    stop_on_convergence: bool = True,
    save_full_results: bool = False,
    save_full_trajectory: bool = False,
    num_threads: int | None = None,
    extra_log_info: Any = None,
    reset_hidden_states: tuple[np.ndarray, ...] | None = None,
    **predict_kwargs: Any,
) -> list[Property]:
    """Evaluate the policy on all registered properties until convergence or a resource limit.

    Provide the policy either as an `agent` exposing a `predict(obs, ...)` method (e.g. any
    Stable-Baselines3 model) or directly as a `predict_fn` callable. Extra keyword arguments are
    forwarded to the policy (e.g. `deterministic=True`).

    At least one stopping criterion must be active: `episode_limit`, `time_limit`, or
    `stop_on_convergence=True` together with at least one property that has an `epsilon`.

    Args:
        agent: An object with a `predict` method. Used only when `predict_fn` is `None`.
        predict_fn: A callable `predict_fn(obs, hidden_states, episode_starts, **kwargs)`
            returning `(actions, hidden_states)` for a batch of observations. Takes precedence
            over `agent`.
        episode_limit: Maximum number of episodes to sample before stopping. `None` for no limit.
        time_limit: Maximum wall-clock time **in minutes** before stopping. `None` for no limit.
        num_initial_episodes: Number of episodes sampled in the first policy run.
        num_episodes_per_policy_run: Number of episodes sampled per subsequent policy run
            (i.e. between convergence checks).
        save_every_n_episodes: Persist intermediate results every this many episodes. `None`
            uses `num_episodes_per_policy_run`.
        stop_on_convergence: Whether to stop once all properties with an `epsilon` converged.
        save_full_results: Whether to store every individual sample.
        save_full_trajectory: Whether to store full trajectories to disk. Slow and disk-heavy;
            not recommended.
        num_threads: Number of worker threads. Defaults to the number of environments. Must not
            exceed the number of environments.
        extra_log_info: Arbitrary extra metadata stored alongside the run settings.
        **predict_kwargs: Extra keyword arguments forwarded to `agent.predict` / `predict_fn`.

    Returns:
        The list of registered [`Property`][pydsmc.Property] instances, each holding its final estimate and confidence interval.

    Raises:
        ValueError: If arguments are invalid (e.g. no stopping criterion, no registered
            properties, `num_threads < 1`, or a non-positive `time_limit`).
        TypeError: If neither a callable `predict_fn` nor an `agent` with a `predict` method is
            given.
    """
    if num_initial_episodes < 1 or num_episodes_per_policy_run < 1:
        raise ValueError("Number of initial episodes, and per policy run, must be at least 1")
    if num_threads is None:
        num_threads = len(self.envs)
    if num_threads < 1:
        raise ValueError("Number of threads must be at least 1")
    if len(self.properties) == 0:
        raise ValueError("No properties registered. Use `register_property`.")

    has_resource_limit = episode_limit is not None or time_limit is not None
    epsilon_props = [prop for prop in self.properties if prop.epsilon is not None]
    if not has_resource_limit and not (stop_on_convergence and len(epsilon_props) > 0):
        raise ValueError(
            "At least one stopping criterion must be set: episode_limit, time_limit, or "
            "stop_on_convergence with at least one property having an epsilon value set.",
        )

    if save_full_trajectory:
        self.logger.warning(
            "SAVING FULL TRAJECTORIES ENABLED. "
            "This is usually not recommended as it will slow down evaluation as well as consume a lot of disk space.",
        )

    eval_params = predict_kwargs | {
        "num_initial_episodes": num_initial_episodes,
        "episode_limit": episode_limit,
        "time_limit": time_limit,
    }

    predict_fn, hidden_states = self.__setup_eval(
        agent=agent,
        predict_fn=predict_fn,
        num_episodes_per_policy_run=num_episodes_per_policy_run,
        save_every_n_episodes=save_every_n_episodes,
        save_full_results=save_full_results,
        eval_params=eval_params,
        num_threads=num_threads,
        extra_log_info=extra_log_info,
    )

    # Fixed-runs (no-epsilon) properties run in their fixed-run setting and are merely reported.
    # if resource limit is set, len(epsilon_props) might be 0
    if stop_on_convergence and len(epsilon_props) == 0:
        self.logger.warning(
            "stop_on_convergence is set to True, but no epsilon property is registered.",
        )
        stop_on_convergence = False

    if time_limit is not None:
        if time_limit <= 0:
            raise ValueError("Time limit must be positive.")
        # convert time limit from minutes to seconds, time limit is a float, 2.5 hours are 2 hours and 30 minutes
        time_limit_seconds = time_limit * 60

    if save_full_results:
        stop_event = threading.Event()
        threading.Thread(
            target=Evaluator.__save_full_results_daemon,
            args=(stop_event, self.properties),
            daemon=True,
        ).start()

    ### run the policy until all properties have converged
    # So sadly, a ProcessPoolExecutor does not work here because
    # (1) VecEnvs are not picklable which could be circumvented
    # (2) The predict_fn is not picklable, which _is not_ circumventable
    # Therefore, still with ThreadPoolExecutor, and suggest the user to use AsyncVectorEnv or SubprocVecEnv
    executor_cm = (
        ThreadPoolExecutor(max_workers=num_threads) if num_threads > 1 else nullcontext()
    )
    with executor_cm as executor:
        start_time = time.perf_counter()

        eval_string = "Starting evaluation"
        if episode_limit is not None and time_limit is not None:
            eval_string += f" with episode limit of {episode_limit} episodes and time limit of {time_limit} minutes"
        elif episode_limit is not None:
            eval_string += f" with episode limit of {episode_limit} episodes"
        elif time_limit is not None:
            eval_string += f" with time limit of {time_limit} minutes"
        self.logger.info(eval_string)

        self.logger.info("The agent will be evaluated according to the following properties:")
        for property_ in self.properties:
            property_string = (
                f"\t{property_.name} using {property_.st_method.__class__.__name__}"
            )
            if property_.epsilon is None:
                property_string += f" in the fixed run setting with min_samples={property_.st_method.min_samples}"
            else:
                property_string += (
                    f" in the sequential setting with epsilon={property_.epsilon}"
                )
            self.logger.info(property_string)

        while True:
            self.__run_policy(
                predict_fn=predict_fn,
                hidden_states=hidden_states,
                executor=executor,
                num_episodes=(
                    num_initial_episodes
                    if self.total_episodes == 0
                    else num_episodes_per_policy_run
                ),
                num_threads=num_threads,
                save_full_trajectory=save_full_trajectory,
                reset_hidden_states=reset_hidden_states,
                **predict_kwargs,
            )

            self.time_passed = time.perf_counter() - start_time
            if stop_on_convergence and all(prop.converged() for prop in epsilon_props):
                self.logger.info("All epsilon properties converged!")
                break

            if (time_limit is not None) and (self.time_passed >= time_limit_seconds):
                self.logger.info("Time limit reached!")
                self.__check_fallback()
                break

            if (episode_limit is not None) and (self.total_episodes >= episode_limit):
                self.logger.info("Episode limit reached!")
                self.__check_fallback()
                break

            if save_every_n_episodes and self.total_episodes >= self.next_log_time:
                overwrite = self.next_log_time == save_every_n_episodes
                for property_ in self.properties:
                    property_.save_results(overwrite=overwrite, logging_fn=self.logger.debug)

                if self.log_dir is not None:
                    save_path = self.log_dir / "resources.jsonl"
                    with save_path.open("w" if overwrite else "a") as f:
                        f.write(
                            json.dumps(
                                {
                                    "total_episodes": self.total_episodes,
                                    "time_passed": self.time_passed,
                                },
                            )
                            + "\n",
                        )

                self.next_log_time = self.total_episodes + save_every_n_episodes

    # Save resources at the end again
    if self.log_dir is not None:
        save_path = self.log_dir / "resources.jsonl"
        with save_path.open("a") as f:
            f.write(
                json.dumps(
                    {"total_episodes": self.total_episodes, "time_passed": self.time_passed},
                )
                + "\n",
            )

    hours, rem = divmod(time.perf_counter() - start_time, 3600)
    minutes, seconds = divmod(rem, 60)
    self.logger.info(
        f"Evaluation finished after {self.total_episodes} episodes, which took "  # noqa: G004
        f"{f'{hours:.0f} hours, ' if hours > 0 else ''}"
        f"{f'{minutes:.0f} minutes, ' if minutes > 0 else ''}"
        f"{seconds:.0f} seconds.",
    )
    for property_ in self.properties:
        converged, intv = property_.get_interval()
        property_string = f"\t{property_.name}: {property_.mean:.3f} ± {property_.std:.3f}"
        if intv is not None:
            property_string += (
                f" ([{float(intv[0]):.3f}, {float(intv[1]):.3f}] / "
                f"{'converged' if converged else 'not converged'})"
            )
        self.logger.info(property_string)

    if save_full_results:
        stop_event.set()

    self.__end_eval(save_full_results)
    return self.properties

clear_properties

clear_properties() -> None

Remove all registered properties.

Source code in pydsmc/evaluator.py

def clear_properties(self) -> None:
    """Remove all registered properties."""
    self.properties = []

set_log_dir

set_log_dir(
    log_dir: Path | str | PathLike | None = "logs",
) -> None

Set the base directory for run logs.

Parameters:

Name	Type	Description	Default
log_dir	Path | str | PathLike | None	New base log directory, or None to disable file logging.	'logs'

Source code in pydsmc/evaluator.py

def set_log_dir(self, log_dir: Path | str | os.PathLike | None = "logs") -> None:
    """Set the base directory for run logs.

    Args:
        log_dir: New base log directory, or `None` to disable file logging.
    """
    self.log_base = Path(log_dir) if log_dir is not None else None
    if self.log_base is not None:
        self.log_base.mkdir(exist_ok=True, parents=True)