Custom Evolver integration#
Implement an Evolver when you want to change the learning method while reusing the task pipeline, adapter specification, and Trainer. Start with the Evolver contract for inputs, outputs, and lifecycle behavior.
Implement the method#
Subclass Evolver and implement step. The following complete example reads the current batch summary, appends
an observation to persistent state, writes a report, and leaves the adapter unchanged:
"""An offline example of the Evolver contract; records evidence without optimizing."""
import json
from hermit.evolver.base import Evolver, StepContext, StepResult
class RecordOnlyEvolver(Evolver):
def step(self, ctx: StepContext) -> StepResult:
summary_path = ctx.workspace.learning_material / "current" / "summary.json"
summary = json.loads(summary_path.read_text())
state_path = ctx.workspace.state / "observations.json"
observations = json.loads(state_path.read_text()) if state_path.exists() else []
observations.append({"step": summary["step"], "mean_score": summary["mean_score"]})
state_path.write_text(json.dumps(observations, indent=2) + "\n")
(ctx.workspace.output / "report.md").write_text("Recorded feedback; adapter unchanged.\n")
return StepResult(status="done", summary="Recorded feedback; adapter unchanged")
The unchanged adapter is the one this step selects. A method that proposes changes writes them into
ctx.workspace.adapter; a method with a candidate pool copies its selected candidate there. StepResult reports
the call’s status and usage.
This example exercises the contract without model calls or optimization. It uses the default setup and
finalize methods. Put any state needed after a process restart in workspace.state, rather than only in instance
attributes.
Declare and select it#
Place evolver.yaml beside evolver.py:
class: evolver.py:RecordOnlyEvolver
needs: []
params: {}
The class value is either file.py:ClassName, relative to the manifest directory, or an importable
package.module:ClassName. The loader constructs the class without arguments and merges manifest params with
run-level evolver.params.
Select its directory in the run config:
evolver:
name: /path/to/record_only
params: {}
Use needs to declare required resources: validation_data, environment, or send_material_out. The Trainer
checks them against the run settings before starting. Set supports_resume = True if the method understands the
previous session handle; this is independent of storing files in state/.
Run the interface example#
From the repository root, with the package installed:
hermit check --config docs/_examples/interface_task/run.yaml
hermit train --config docs/_examples/interface_task/run.yaml
hermit eval --config docs/_examples/interface_task/run.yaml --adapter final
The task is a local echo stub. The run has two training steps and one test task; no model is called.
Inspect .demo/interface-example/interface-demo/ for the learning material, per-step
state/observations.json snapshots, final adapter card, and evaluation records.
This example uses trusted in-process Python and disables the learner-program sandbox. For a fresh repeat, add
trainer.run_name=interface-demo-02 to the commands. See Trainer for resume behavior.
Agent and command implementations#
You can reuse a launcher instead of implementing a new Python learning loop:
Class |
What you supply |
How it hands back an adapter |
|---|---|---|
|
Backend, prompt files, model selection, and host arguments. |
The learner agent edits the working directory. |
|
An external command and optional environment values. |
The program edits the working directory and may write a result file. |
AgentEvolver supports codex and claude. Its parameters include backend, prompt, resume_prompt, model,
args, and home. Prompt placeholders include the workspace paths, message, epoch, and step.
The bundled definitions are under hermit/algorithms/evolvers/.
CommandEvolver supplies HA_CONTEXT, a path to JSON describing the workspace, access, progress, session, and
parameters, plus HA_TOOL_URL for the tool endpoint. A program may write output/step_result.json using the
StepResult fields. Without that file, the process exit code determines success or failure.
Return a failure status when the program cannot produce a usable result.
The same protocol can wrap an external service. Declare required data access, preserve the adapter’s specification, and report known usage; the wrapper still owns translating the service’s result into the working directory.
Execution and cancellation#
Custom classes run in the Trainer process. Their Python code is trusted; the sandbox object wraps programs launched through it, not the class itself. Framework tools are callable in-process and can also be exposed to programs through HTTP or MCP.
During a step, long-running custom code should check ctx.cancel when present. A budget or timeout stops work at
the next boundary the framework controls; it cannot forcibly interrupt arbitrary Python code. See
budgets for accounting and cancellation limits.
Report calls made outside metered framework tools through StepResult.usage. Leave cost null when unknown.
Calls through the run’s metered model tool must not be counted a second time.
Python types#
These excerpts come directly from the implementation. The Evolver page explains their semantics and intended access boundaries.
Lifecycle#
class Evolver(ABC):
supports_resume: bool = False
def setup(self, ctx: Context) -> None:
"""Optional: before the first step (for example, initialise Training State)."""
def bootstrap(self, ctx: StepContext) -> StepResult | None:
"""Optional: change the starting adapter once, before the first rollouts of a fresh run (for example, a
learner session that reads the task's public information). The Trainer checks and commits the result as
step 0. Not overridden (or returning None): no bootstrap."""
return None
@abstractmethod
def step(self, ctx: StepContext) -> StepResult:
"""Read the learning material and change the adapter in ctx.workspace.adapter (or leave it)."""
def finalize(self, ctx: Context) -> FinalResult:
"""Optional: after training. Default: hand back the current adapter."""
return FinalResult(adapter=ctx.workspace.adapter, summary="current adapter")
Workspace and context#
@dataclass(frozen=True)
class Workspace:
adapter: Path # working copy of the current adapter (read/write)
learning_material: Path # step_NNNN/ directories, current, index.jsonl, README.md (read-only)
task_info: Path # task description (read-only)
history: Path # read-only view of the adapter's git history
state: Path # Training State (read/write, the Evolver's own format)
scratch: Path # temporary files (read/write, not kept)
output: Path # what the call hands back besides the adapter (write)
@dataclass(frozen=True)
class Access:
validation: Path | None = None # set when data.val is given
environment: tuple[Path, ...] = () # from evolver_access.environment
@dataclass(frozen=True)
class Context:
workspace: Workspace
access: Access
tools: dict[str, Tool] # always includes "check_spec"; every call is logged and metered
params: dict[str, Any] # evolver.params from the run config over the manifest defaults
message: str = "" # the task message of this call: accesses, tools, progress, remaining budget
sandbox: Any = None # the run's sandbox: wraps commands an Evolver starts (see hermit.sandbox)
cancel: Any = None # threading.Event set by the Trainer when the call must stop (budget, timeout)
tool_url: str | None = None # HTTP endpoint serving the same tools (for agents and external programs)
base: Path | None = None # the run config's directory: relative paths in params resolve against it
@dataclass(frozen=True)
class Progress:
epoch: int
step: int
is_epoch_end: bool
@dataclass(frozen=True)
class StepContext(Context):
progress: Progress | None = None
session: str | None = None # previous call's session handle, set only when resuming
Results and usage#
@dataclass
class StepResult:
status: Literal["done", "failed", "out_of_budget", "stop"]
summary: str # one line: what changed, or that nothing did
rationale: str = "" # optional: why it changed the adapter this way, or why not
session: str | None = None # handle for resuming next time, if any
usage: list[Usage] = field(default_factory=list) # model usage of this call
extra: dict[str, Any] = field(default_factory=dict) # recorded unchanged
@dataclass
class FinalResult:
adapter: Path # the adapter to hand back as the final result
summary: str
extra: dict[str, Any] = field(default_factory=dict)
@dataclass
class Usage:
model: str
input_tokens: int = 0
cached_input_tokens: int = 0
output_tokens: int = 0
calls: int = 0
cost_usd: float | None = None # None when no price applies (for example a subscription)