Grader

Grader assigns the reward for the single sample produced by a rollout execution.

Grader Base Class

from osmosis_ai.rollout import Grader, GraderContext

class MyGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        if ctx.sample is None:
            raise ValueError("workflow produced no sample")
        # Evaluate ctx.sample and assign its reward
        ctx.set_reward(1.0)

The base class signature from the SDK:

class Grader(ABC):
    def __init__(self, config: GraderConfig | None = None):
        self.config = config

    @abstractmethod
    async def grade(self, ctx: GraderContext) -> Any:
        raise NotImplementedError

grade() receives the sample, reference label, metadata, and optional artifacts directory through GraderContext.

GraderContext

The ctx parameter passed to grade() provides:

Field	Type	Description
`ctx.label`	`str \| None`	Reference answer for the current dataset row (typically your `ground_truth` column)
`ctx.metadata`	`dict[str, Any] \| None`	Per-row metadata from the dataset’s optional `metadata` column. `None` when the row has no metadata.
`ctx.sample`	`RolloutSample \| None`	The single agent output, or `None` when the workflow registered no sample source
`ctx.project_path`	`str \| None`	Optional project path supplied by the execution harness
`ctx.artifacts_dir`	`pathlib.Path \| None`	Per-rollout directory where the grader can write log, trace, and other output files. `None` when the execution environment can’t provision a writable directory, so check for `None` before writing files.
`ctx.set_reward(reward)`	method	Assign a float reward to `ctx.sample`

With LocalBackend, a configured Grader runs whenever a dataset row has a label or metadata, so metadata alone can drive reward. With HarborBackend, an existing task tests/test.sh remains authoritative; the Osmosis Grader is installed as the verifier only when that file is absent. See Harbor reward precedence.

One workflow execution produces at most one sample. Evaluation and training can still execute the workflow multiple times for the same prompt ([evaluation].n in evaluation configs, n_samples_per_prompt in training configs); each independent execution receives its own GraderContext.

`set_reward`

Call ctx.set_reward(reward) to assign a reward to the rollout’s sample. The reward should be a float, typically between 0.0 and 1.0. Any finite float value is accepted, and NumPy-like numeric scalars are normalized to float.

ctx.set_reward(0.85)

set_reward raises a ValueError when ctx.sample is None. Check for a sample before scoring it; a missing sample usually means the workflow did not construct its supported agent or session inside run().

NaN, infinity, and non-numeric values raise pydantic.ValidationError because they violate the reward’s JSON wire contract. Return the intended numeric reward, or leave the reward unset (do not call set_reward) when the sample is “not graded”.

Writing Artifacts

Use ctx.artifacts_dir to persist rubric traces, diffs, or any other files your grader produces. The directory is per-rollout and shared with the workflow that produced the sample, so your grader can also read files the workflow wrote. It’s None when the environment can’t provision one, so guard with if ctx.artifacts_dir: before writing to it — an unguarded write raises and fails the grader.

import json

from osmosis_ai.rollout import Grader, GraderContext


class RubricGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        if ctx.artifacts_dir:
            (ctx.artifacts_dir / "grade_trace.json").write_text(
                json.dumps({"reason": "matched rubric"})
            )
        if ctx.sample is not None:
            ctx.set_reward(1.0)

After the rollout finishes, collected files appear alongside its sample in the run’s Artifacts panel on the Osmosis Platform, mirroring the layout you write under ctx.artifacts_dir. Artifact collection never affects rewards or rollout status.

RolloutSample

ctx.sample is a RolloutSample object containing the AgentWorkflow’s output:

from collections.abc import Mapping, Sequence
from typing import Any

from pydantic import BaseModel, Field


class RolloutSample(BaseModel):
    messages: Sequence[Mapping[str, Any]] = Field(default_factory=list)
    trajectory_messages: Sequence[Mapping[str, Any]] | None = None
    label: str | None = None
    reward: float | None = None
    remove_sample: bool = False
    metrics: dict[str, Any] = Field(default_factory=dict)
    extra_fields: dict[str, Any] = Field(default_factory=dict)

The messages list is the conversation your workflow produced for that sample. In many graders, you only need to extract the final answer text from the last assistant message.

For real-world references, see rollouts/multiply-local-strands/main.py and rollouts/multiply-local-openai/main.py in the workspace-template repository. Those files are the source of truth for platform-created workspace repositories.

Implementation Patterns

Exact Match Grading

The simplest grading strategy is to compare the agent’s final text against ctx.label. The helper below extracts text from the last message:

from osmosis_ai.rollout import Grader, GraderContext


def _last_text(sample) -> str:
    """Extract the final text block from a sample's last message."""
    if not sample.messages:
        return ""
    content = sample.messages[-1].get("content", "")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return next((b["text"] for b in content if isinstance(b, dict) and "text" in b), "")
    return ""


class ExactMatchGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        if ctx.sample is None:
            raise ValueError("workflow produced no sample")
        answer = _last_text(ctx.sample).strip()
        reward = 1.0 if ctx.label and answer == ctx.label.strip() else 0.0
        ctx.set_reward(reward)

LLM-as-Judge Grading

Use a separate LLM to evaluate agent outputs when correctness is subjective or hard to check programmatically. Judge calls do not need the rollout model integration used for policy calls, so you can call another LLM directly. Grading still runs synchronously after the workflow; its latency and failures affect the rollout.

import litellm
from osmosis_ai.rollout import Grader, GraderContext


class LLMJudgeGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        if ctx.sample is None:
            raise ValueError("workflow produced no sample")
        agent_output = _last_text(ctx.sample)
        judge_response = await litellm.acompletion(
            model="openai/gpt-5.2",
            messages=[{
                "role": "user",
                "content": f"Rate this response from 0.0 to 1.0.\n\n"
                           f"Expected: {ctx.label}\n"
                           f"Actual: {agent_output}\n\n"
                           f"Score (just the number):"
            }],
        )
        score = float(judge_response.choices[0].message.content.strip())
        ctx.set_reward(max(0.0, min(1.0, score)))

Tool-Call Based Grading

Evaluate whether the agent made any tool calls, rather than just checking the final text output. Strands records tool invocations as toolUse content blocks on assistant messages:

from osmosis_ai.rollout import Grader, GraderContext


class ToolCallGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        if ctx.sample is None:
            raise ValueError("workflow produced no sample")
        used_tool = False
        for message in ctx.sample.messages:
            if message.get("role") != "assistant":
                continue
            content = message.get("content") or []
            if isinstance(content, list) and any(
                isinstance(block, dict) and "toolUse" in block for block in content
            ):
                used_tool = True
                break
        ctx.set_reward(1.0 if used_tool else 0.0)

You can combine multiple grading strategies — for example, check that the agent used the right tools and produced a correct final answer, then weight the scores together.

GraderConfig

Custom grader configs follow the same pattern as AgentWorkflowConfig: extend GraderConfig, create an instance, and pass it explicitly to the backend:

from osmosis_ai.rollout import Grader, GraderConfig, GraderContext

class MyGraderConfig(GraderConfig):
    name: str = "my-grader"
    partial_credit: bool = True
    similarity_threshold: float = 0.8

class MyGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        threshold = self.config.similarity_threshold if self.config else 0.8
        # ... use config values in grading logic ...

my_grader_config = MyGraderConfig()

Pass the config instance to LocalBackend(grader_config=my_grader_config). Evaluation and training TOML files do not currently set grader config fields directly. GraderConfig extends BaseConfig and includes the same concurrency field as AgentWorkflowConfig, but current backends do not use it to limit grader concurrency. Use evaluation [evaluation].batch_size, workflow/backend concurrency, or an explicit limiter inside the grader when your grader calls external services.

Field	Type	Default	Description
`name`	`str`	(required)	Identifier for the grader
`description`	`str \| None`	`None`	Optional description
`concurrency`	`ConcurrencyConfig`	unlimited	Present on the config model; not currently enforced by `LocalBackend`

Entry Point Wiring

Grader classes and config objects require explicit wiring. Select them in the backend constructor:

from osmosis_ai.rollout import LocalBackend
from osmosis_ai.rollout.server import create_rollout_server

backend = LocalBackend(
    workflow=MyWorkflow,
    grader=MyGrader,
    grader_config=my_grader_config,
)
app = create_rollout_server(backend=backend)

Multiple concrete Grader subclasses can coexist in the entrypoint; only the class passed as grader runs. Submit preflight imports the entrypoint once to surface constructor and dependency errors and does not inspect its module namespace; see Files in a Rollout for when that import is skipped and what it executes locally.

Next Steps

Evaluation

Submit an evaluation run to test your AgentWorkflow and Grader before a training run.

Get started

Core API

Integrations

Execution

Agent tools

Grader Base Class

GraderContext

`set_reward`

Writing Artifacts

RolloutSample

Implementation Patterns

Exact Match Grading

LLM-as-Judge Grading

Tool-Call Based Grading

GraderConfig

Entry Point Wiring

Next Steps

Evaluation

​Grader Base Class

​GraderContext

​set_reward

​Writing Artifacts

​RolloutSample

​Implementation Patterns

​Exact Match Grading

​LLM-as-Judge Grading

​Tool-Call Based Grading

​GraderConfig

​Entry Point Wiring

​Next Steps

Evaluation

Grader Base Class

GraderContext

`set_reward`

Writing Artifacts

RolloutSample

Implementation Patterns

Exact Match Grading

LLM-as-Judge Grading

Tool-Call Based Grading

GraderConfig

Entry Point Wiring

Next Steps