Write a recipe

A recipe is one frozen dataclass. It says which traffic counts as training data, how accepted data becomes a batch, what the batch means numerically, and which backend objective consumes it. reef runs everything else: pairing records that arrive out of order, deciding what the store may delete, replaying after a crash, advancing the version chain, and serving the result.

There are two processes. The reef service owns records, retention, and signals. The training driver holds the GPUs, torch, and checkpoints. They must agree on the objective at boot (three places name it).

How a recipe runs#

The three filled boxes are yours. judge sees one report with its referenced inference records already resolved, and returns TRAIN, WAIT, or NEVER. make_batch shapes the accepted units into your batch type. The step preparer turns that batch into a StepSignal carrying the advantages, the metrics, and the loss family.

The dotted edge is why the engine exists. Records arrive in no guaranteed order, a grader retries its POST, and a crash replays the tail. WAIT parks a report until its last inference lands, and NEVER releases both records. The engine does all of that, so none of it belongs in recipe code.

What you define, what reef runs#

The slots the recipe class declares, and what a wrong one costs:

What to change#

In your own package#

All of these go in your own package, with nothing added inside reef:

1. A processor. Subclass PairingProcessor or DerivedProcessor, declare output_schema, and write judge and make_batch.
2. A step preparer. It is a plain function that takes a reserved batch plus the algorithm state and returns a StepSignal. Reuse a registered name from reef/train/algos/preparers.py when its signal already matches.
3. A recipe class. It is a frozen dataclass over WeightTrainingRecipe that binds those two plus a loss_family.
4. A deployment YAML. It names the recipe as a dotted kind, package.module:ClassName.

recipe_class_for imports a dotted kind on demand and refuses anything that is not a Recipe subclass, so no registration call is involved. Everything a bundled recipe can do is available here except appearing in recipe_kinds().

Bundled into reef#

The same files, plus the ones that make a kind discoverable, tested, and documented:

5. @register_kind("<kind>") on the class, with name set to the same string. Registering a kind twice raises at import.
6. An import in reef/recipes/__init__.py, and the __all__ entry. Registration happens at import time, so both are required for the kind to be discoverable.
7. Tests, copied from a sibling in the same cell of the grid below. A new kind needs its own suite and an entry in the shared recipe and schema parametrizations.
8. An example under examples/<kind>/ with a runnable config, and the doc rows: the top-level README table, reef/recipes/README.md, and the recipe table in reef/train/processors/README.md.

A bundled recipe starts as an RFC rather than a module. See the layout policy before adding one.

The kinds of recipe#

The shape of a recipe follows from what it updates and how work arrives. What it updates is the surface build_surface() returns. How work arrives is the traffic the scenario sees.

What each cell changes in the code you write:

• In weights × stream of tasks, one inference produces one report and one sample. The unit is paired and ungrouped, so judge returns Judgment.train(sample) with no group_key and make_batch takes unit.candidates[0].value. The worked example below sits in this cell.
• In weights × chat session, no report exists. The processor is derived, so ingest carries the recipe, written from the engine's catch_up / dispatch / track / retire verbs, and judge is async because it calls a model. report_schema is None on purpose.
• In weights × autoresearch, there is one scenario per stated problem, and a whole grid of rollouts has to finish before a batch is ready. This is the only cell that writes group_ready and sets ordered_groups = True.
• In harness artifact × stream of tasks, the report triggers the batch, while the accept/reject decision is computed in a LocalTrainingBackend from the operator's propose and evaluate. This is the only cell that subclasses bare Recipe rather than WeightTrainingRecipe. It declares no step_preparer, loss_family, or processor, and it overrides both build() and build_surface().
• With nothing as the surface, build() returns a bare DataProcessor that ingests for audit and never becomes ready. Start every integration here to prove the wiring before any objective is involved.

"Single graded rollout" is stream of tasks with batch_size = 1, a cadence setting on one class, so it gets no shape of its own. The empty cells are empty because nothing has needed them yet, and docs/rfcs/evolution-boundary.md §5 sketches the two harness ones.

Two modules under reef/recipes/ sit in no cell. ace.py registers with implemented=False, so its kind resolves but build() raises. The class is there to be discoverable while its proposer is built out. reasoningbank.py registers nothing at all.

Which engine#

The tier depends on whether judging needs to call a model. If it does not, use PairingProcessor, whose judge runs inside the engine's own ingest and decides on data already in hand. If it does, use DerivedProcessor, whose async judge runs on a private worker after the recipe's ingest dispatches, so a judgment taking minutes cannot stall serving. reef/train/processors/README.md carries the full comparison and the reason the two engines differ.

Build one#

The worked example is reward-weighted regression, where each sample trains with its score as a per-sample weight (RWR; Peters & Schaal, ICML 2007). It uses every piece you can extend and nothing more: a processor that decides acceptance, a preparer that computes the weights, a recipe class that binds them, and a YAML that deploys it.

Every code block below runs verbatim in reef's test suite: test_define_a_recipe_guide.py extracts them from this file and executes them.

Everything here assumes each report carries a numeric score. reef never produces that number. The section on where the score comes from returns to it once the machinery is built, including what to do when no honest number exists.

The step preparer#

The preparer comes first because it is pure computation. It says what the batch means numerically and needs to know nothing else. It is a plain function with no class and no registration, and it works with any backend, importing neither torch, ray, nor Slime.

# my_pkg/weighted_sft.py
from collections.abc import Mapping
from typing import Any

from reef.train.algos import StepSignal, next_steps, require_policy_batch
from reef.train.types import TrainingBatch


def prepare_step(batch: TrainingBatch, state: Mapping[str, Any]) -> StepSignal:
    """Reward-weighted regression: each sample's advantage is its reported score."""
    policy = require_policy_batch(batch, "weighted_sft")
    advantages = tuple(sample.reward for sample in policy.samples)
    steps = next_steps(state)
    return StepSignal(
        action="train",
        loss_family="pg",
        next_algorithm_state={"steps": steps},
        metrics={"steps": steps},
        advantages=advantages,
    )

action is "train" or "skip", where "skip" is a state-only transition with no backend step. loss_family names the backend objective. next_algorithm_state is committed only after the backend step succeeds, and here it holds a step counter that next_steps validates and advances. metrics passes through to telemetry untouched. advantages is one float per sample, in batch order.

The one substantive line is advantages, and it works because the processor put each report's score on sample.reward when it paired the report with its rollout. The reported score becomes the per-sample weight, and that is the whole recipe.

The family is pg rather than sft, because a per-sample weight is an advantage and Slime's sft_loss_function has no advantage slot. reef raises an error instead of quietly training without the weights.

The recipe class and its processor#

# my_pkg/weighted_sft.py (continued)
from dataclasses import dataclass
from typing import ClassVar

from reef.recipes import knob
from reef.recipes.base import WeightTrainingRecipe
from reef.train.processors.base import DataProcessor
from reef.train.processors.pairing import NEVER, Judgment, PairingProcessor, ReportView, SampleAssembly
from reef.train.types import PolicyBatch, ProcessorContext


class ThresholdProcessor(PairingProcessor):
    """Accept every trainable report at or above ``min_score``; one report, one sample."""

    output_schema = PolicyBatch

    def __init__(self, context: ProcessorContext) -> None:
        self._min_score = float(context.config.get("min_score", 0.0))
        self._assembly = SampleAssembly.from_config(context)
        super().__init__(context)

    def judge(self, view: ReportView) -> Judgment:
        if (gate := view.eligibility()) is not None:
            return gate  # NEVER, or WAIT until the last reference lands
        score = view.score
        assert score is not None  # eligibility guarantees a finite score
        sample = self._assembly.build(view, score)
        if sample is None or score < self._min_score:
            return NEVER
        return Judgment.train(sample)

    def make_batch(self, units, batch_number: int) -> PolicyBatch:
        return PolicyBatch(
            f"{self.scenario}:weighted_sft:{batch_number}",
            tuple(unit.candidates[0].value for unit in units),
        )


@dataclass(frozen=True, kw_only=True)
class WeightedSFTRecipe(WeightTrainingRecipe):
    batch_size: int = knob(4, env="REEF_WEIGHTED_SFT_BATCH_SIZE")
    min_score: float = knob(0.0, env="REEF_WEIGHTED_SFT_MIN_SCORE")
    name: str = "weighted_sft"
    step_preparer: ClassVar[str] = "my_pkg.weighted_sft:prepare_step"
    loss_family: ClassVar[str] = "pg"
    processor: ClassVar[type[DataProcessor]] = ThresholdProcessor

The processor is fifteen lines because the engine owns everything except the acceptance rule. view.eligibility() is the shared gate every score-based judge starts with. It returns NEVER for a report with no references, a non-finite score (NaN and inf never train), or an explicit ineligibility marker. It returns WAIT for an inference that has not arrived. It returns None to hand the judgment back to the recipe. Do not re-derive any of it. SampleAssembly is the shared shaping step, where one model call becomes one sample and an ordered multi-call report becomes one assembled sample.

frozen=True is mandatory, because Python refuses a non-frozen dataclass over the frozen base. kw_only=True makes the declared fields keyword-only while the inherited runtime argument stays positional. knob() is a normal dataclasses.field carrying reef metadata, so one declaration gives you the YAML key, the environment fallback, and the type-aware parser at once. A float knob arrives as 0.5 and is never truncated. A ClassVar is a class constant rather than a constructor field. A knob configures an instance, and a ClassVar binds the recipe.

build() is inherited. It constructs the declared processor with the recipe's data-side knob values as its config (processor_config() is overridable when the processor needs renamed or extra keys) and binds the preparer. from_environment() and service_config() derive their keys and parsing from the knob fields, with precedence config over environment over default. An override of __post_init__ must start with super().__post_init__() so inherited validation still runs.

Smoke-test it, in process#

Before any YAML and before any GPU, prove the whole data path with fake records: records in, pairing, batch reservation, preparer signal, and wire payload. reef.runtime.testing.StubTrainingRuntime is the stub the test needs, so you mock nothing by hand and the test runs in milliseconds.

# tests/test_weighted_sft.py
from my_pkg.weighted_sft import WeightedSFTRecipe
from reef.core.records_types import AgentRecord, RequestType
from reef.records import RecordStore
from reef.runtime.testing import StubTrainingRuntime
from reef.train.slime_runtime.reef_adapters.preparation import prepare_slime_step


def _inference(agent_record_id, tokens, loss_mask, rollout_log_probs):
    return AgentRecord.create(
        scenario="smoke",
        request_type=RequestType.INFERENCE,
        agent_record_id=agent_record_id,
        payload={"response": {"training": {
            "tokens": tokens,                       # prompt + response ids
            "loss_mask": loss_mask,                 # one entry per response token
            "rollout_log_probs": rollout_log_probs, # one per response token
        }}},
    )


def _report(agent_record_id, reference, score):
    return AgentRecord.create(
        scenario="smoke",
        request_type=RequestType.REPORT,
        agent_record_id=agent_record_id,
        payload={"score": score, "references": [reference]},
    )

The response.training block is the tensor contract a real harness posts back with each response, and Required training tensors is where that contract is documented. The report's references entry names the inference it grades.

# tests/test_weighted_sft.py (continued)
def test_weighted_sft_reserves_a_reward_weighted_pg_payload():
    records = RecordStore()
    records.append(_inference("i1", [10, 20, 30], [1, 1], [-0.1, -0.2]))
    records.append(_report("r1", "i1", 0.8))

    recipe = WeightedSFTRecipe(StubTrainingRuntime(), batch_size=1)
    trainer = recipe.build("smoke", records)

    batch = trainer.reserve_training_batch()
    assert batch is not None

    prepared = prepare_slime_step(batch, recipe.step_preparer, trainer.algorithm_state_dict())
    assert prepared.action == "train"
    assert prepared.payload["loss"] == "pg"
    assert prepared.payload["advantages"] == [0.8]

reserve_training_batch() returning a batch proves the processor's pairing and batch-size logic against your records. The prepared payload proves the preparer and the wire contract, because prepare_slime_step is exactly what the bridge runs.

Deploy it#

Three places name the objective#

The recipe class, the driver environment, and the Slime flags must agree on the objective. The driver validates the combination at startup and rejects a mismatch with an error, so a disagreement stops the run at boot instead of producing a wrong gradient.

When both variables are set, REEF_TRAINING_LOSS wins and REEF_TRAINING_RECIPE is informational. pg also requires --use-rollout-logprobs, so the recorded rollout log-probs serve as the old-policy probabilities. The bundled families are sft, pg, sao, opd, tttd, and openclawrl-topk, which is the list reef/train/slime_runtime/training_algos/registry.py registers. A family reef does not bundle registers a SlimeAlgorithm with register_loss_family, or it is named as a dotted package.module:SPEC reference. See Continual and train.

The service YAML#

The deployment config selects the recipe as a dotted kind and sets its knobs as flat reef.* keys, where the knob field names are the key names:

reef:
  recipe: "my_pkg.weighted_sft:WeightedSFTRecipe"
  model_path: ~/models/Qwen2.5-1.5B-Instruct
  batch_size: 4        # consumed by the batch_size knob; MUST equal
                       # training.global_batch_size
  min_score: 0.5       # float knob: arrives as 0.5, not 0
  checkpoint_every_n_versions: 1
  ray_address: 127.0.0.1:${training.ray_port}
  inference_url: http://${REEF_INFERENCE_HOST}:${training.sglang_port}
  # pg needs engine-native rollout tensors; see slime-training.md
  inference_backend_factory: >-
    reef.train.slime_runtime.reef_adapters.sglang_chat.SGLangChatTrainingInferenceBackend

Two vocabularies share the reef.* namespace. The keys the service consumes (recipe, model_path, ray_address, inference_url, inference_backend_factory) never reach the recipe. Every remaining key must match the selected recipe, meaning its knobs plus checkpoint_every_n_versions, which every training recipe accepts. A key that neither vocabulary consumes fails the boot with an error listing the keys the recipe does consume, so a knob typo cannot quietly run the deployment on defaults.

A stateful judge#

A fixed min_score goes stale. Once the policy clears the bar, every report passes and training rehearses what the policy already does. A relative filter avoids that by accepting only the reports that beat the rolling mean of recent scores.

# my_pkg/relative_sft.py
from collections import deque

from reef.train.processors.pairing import NEVER, Judgment, PairingProcessor, ReportView, SampleAssembly
from reef.train.types import PolicyBatch, ProcessorContext


class RelativeImprovementProcessor(PairingProcessor):
    """Train only on reports that beat the rolling mean of recent scores."""

    output_schema = PolicyBatch
    exclusive_sources = True

    def __init__(self, context: ProcessorContext) -> None:
        self._window = int(context.config.get("window", 8))
        self._assembly = SampleAssembly.from_config(context)
        self._recent: deque[float] = deque(maxlen=self._window)
        super().__init__(context)

    def judge(self, view: ReportView) -> Judgment:
        if (gate := view.eligibility()) is not None:
            return gate  # NEVER, or WAIT until the last reference lands
        score = view.score
        assert score is not None  # eligibility guarantees a finite score
        sample = self._assembly.build(view, score)
        baseline = sum(self._recent) / len(self._recent) if self._recent else None
        self._recent.append(score)  # every final judgment moves the baseline
        if sample is None:
            return NEVER  # rollout unusable for this recipe's samples
        if baseline is not None and score <= baseline:
            return NEVER  # did not beat the rolling mean
        return Judgment.train(sample)

    def make_batch(self, units, batch_number: int) -> PolicyBatch:
        return PolicyBatch(
            f"{self.scenario}:relative:{batch_number}",
            tuple(unit.candidates[0].value for unit in units),
        )

Closing over mutable state is safe here. The gate returns before the baseline update, and TRAIN and NEVER are final, so each report moves the baseline exactly once. The trainer serializes every processor call under one lock, so the deque needs no locking. Crash recovery replays records through ingest in their original order, so the state is rebuilt by the same code that built it. exclusive_sources = True states that each report owns its referenced rollout outright, so a terminal report releases its inference record with it.

The recipe swaps one ClassVar and trades min_score for window. The preparer is untouched, because the processor decides which units train and the preparer still decides what the training signal is:

# my_pkg/relative_sft.py (continued)
from dataclasses import dataclass
from typing import ClassVar

from reef.recipes import knob
from reef.recipes.base import WeightTrainingRecipe
from reef.train.processors.base import DataProcessor


@dataclass(frozen=True, kw_only=True)
class RelativeSFTRecipe(WeightTrainingRecipe):
    batch_size: int = knob(4, env="REEF_RELATIVE_SFT_BATCH_SIZE")
    window: int = knob(8, env="REEF_RELATIVE_SFT_WINDOW")
    name: str = "relative_sft"
    step_preparer: ClassVar[str] = "my_pkg.weighted_sft:prepare_step"
    loss_family: ClassVar[str] = "pg"
    processor: ClassVar[type[DataProcessor]] = RelativeImprovementProcessor

A custom processor is worth testing at its own surface, with the same fake records. A report waits for its inference, trains once the inference arrives, and a below-baseline report is terminal and released:

# tests/test_relative_sft.py
from my_pkg.relative_sft import RelativeImprovementProcessor
from reef.train.types import ProcessorContext

# The fake-record helpers from the first smoke test, reused unchanged.
from tests.test_weighted_sft import _inference, _report


def test_relative_filter_waits_trains_and_releases():
    processor = RelativeImprovementProcessor(
        ProcessorContext("smoke", {"batch_size": 1, "window": 4})
    )

    # The report arrives before its inference: WAIT, nothing to batch.
    processor.ingest(_report("r1", "i1", 0.5))
    assert not processor.ready()

    # The inference lands; r1 is judged again, and the first score has
    # no baseline to beat, so it trains.
    processor.ingest(_inference("i1", [10, 20, 30], [1, 1], [-0.1, -0.2]))
    assert processor.ready()
    batch = processor.build_batch()
    assert [sample.reward for sample in batch.samples] == [0.5]
    processor.acknowledge(batch.batch_id)

    # A score below the rolling mean (0.5) is terminal. The engine releases
    # the report with its rollout; the recipe holds no retention code.
    processor.ingest(_inference("i2", [11, 21, 31], [1, 1], [-0.3, -0.4]))
    processor.ingest(_report("r2", "i2", 0.2))
    assert not processor.ready()
    released = processor.retention_decision().releasable_agent_record_ids
    assert {"r2", "i2"} <= released

    # A score above the mean of (0.5, 0.2) trains again.
    processor.ingest(_inference("i3", [12, 22, 32], [1, 1], [-0.2, -0.1]))
    processor.ingest(_report("r3", "i3", 0.9))
    assert processor.ready()
    assert [sample.reward for sample in processor.build_batch().samples] == [0.9]

ingest → ready → build_batch → acknowledge is exactly what the trainer drives in deployment, so a processor proven at this surface is proven for the real loop.

Where the score comes from#

In verifiable domains a grade is nearly free: a test suite's pass rate, a checked math answer, a compile-and-run result. Outside them, deployments synthesize one, using an LLM judge scoring against a rubric, or a behavioral signal treated as a grade (the user accepted the answer, edited it, retried).

If your scenario has no defensible number, a score-based recipe is the wrong tool, and inventing a number to fit it is worse. Feedback in reef can be a score, plain text, or a structured object, and what a recipe reads says nothing about what it improves. Harness evolution reads a score like any weight recipe and improves the agent's tree instead of its weights. OpenClaw-RL is handed no report at all and derives its own signal from the traffic. Start from what your environment can honestly report, then pick the cell of the grid that consumes exactly that.

What not to write#