Grader

Grader 为一次 rollout execution 产生的单个 sample 分配 reward。

Grader 基类

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")
        # 评估 ctx.sample 并分配 reward
        ctx.set_reward(1.0)

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() 通过 GraderContext 接收 sample、reference label、metadata 和可选的 artifacts directory。

GraderContext

传递给 grade() 的 ctx 参数提供：

字段	类型	描述
`ctx.label`	`str \| None`	当前数据集行的参考答案（通常对应 `ground_truth` 列）
`ctx.metadata`	`dict[str, Any] \| None`	来自数据集可选 `metadata` 列的每行 metadata。该行无 metadata 时为 `None`。
`ctx.sample`	`RolloutSample \| None`	单个 Agent 输出；workflow 未注册 sample source 时为 `None`
`ctx.project_path`	`str \| None`	由执行 harness 提供的可选项目路径
`ctx.artifacts_dir`	`pathlib.Path \| None`	每次 rollout 的目录，grader 可在其中写入日志、trace 和其他输出文件。当执行环境无法提供可写目录时为 `None`，写入文件前请先检查它是否为 `None`。
`ctx.set_reward(reward)`	方法	为 `ctx.sample` 分配一个浮点数 reward

使用 LocalBackend 时，只要 dataset row 包含 label 或 metadata，配置好的 Grader 就会运行，因此可以仅依靠 metadata 驱动 reward。使用 HarborBackend 时，已有 task tests/test.sh 仍是权威来源；只有该文件不存在时，Osmosis Grader 才会作为 verifier 安装。请参见 Harbor reward 优先级。

一次 workflow 执行最多生成一个 sample。Evaluation 和 training 仍然可以对同一个 prompt 多次执行 workflow（evaluation config 中的 [evaluation].n，training config 中的 n_samples_per_prompt）；每次独立执行都会收到自己的 GraderContext。

`set_reward`

调用 ctx.set_reward(reward) 为 rollout 的 sample 分配 reward。reward 应为浮点数，通常在 0.0 到 1.0 之间。任何有限浮点值均可接受，NumPy 风格的数值标量会被自动规范化为 float。

ctx.set_reward(0.85)

当 ctx.sample 为 None 时，set_reward 会抛出 ValueError。评分前请检查 sample；缺少 sample 通常意味着 workflow 没有在 run() 内构造受支持的 agent 或 session。

NaN、无穷大以及非数值值会触发 pydantic.ValidationError，因为它们违反 reward 的 JSON wire contract。请返回预期的数值 reward，或直接不调用 set_reward 表示该 sample 未评分。

写入 artifacts

Grader 可以通过 ctx.artifacts_dir 持久化评分 trace、diff 或任何其他文件。该目录按 rollout 独立，并与产生 sample 的 workflow 共享，因此 grader 也可以读取 workflow 写入的文件。当执行环境无法提供时为 None，因此写入前请用 if ctx.artifacts_dir: 进行判断 —— 未加判断的写入会抛出异常并导致 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)

Rollout 结束后，收集到的文件会与对应 sample 一起显示在 Osmosis 平台上 run 的 Artifacts 面板中，并保留你在 ctx.artifacts_dir 下写入的目录结构。Artifact 收集永远不会影响 rewards 或 rollout 状态。

RolloutSample

ctx.sample 是一个包含 AgentWorkflow 输出的 RolloutSample 对象：

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)

messages 列表就是您的 workflow 为该 sample 产出的对话记录。在很多 grader 里，您只需要从最后一条 assistant 消息中提取最终答案文本即可。

真实参考请看 workspace-template 仓库中的 rollouts/multiply-local-strands/main.py 和 rollouts/multiply-local-openai/main.py。这些文件是平台创建 workspace repositories 时使用的 source of truth。

实现模式

精确匹配评分

最简单的评分策略就是把 Agent 的最终文本和 ctx.label 直接比较。下面的辅助函数演示了如何从最后一条消息中提取文本：

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 评分

当正确性具有主观性或难以通过程序化方式检查时，可以使用另一个 LLM 评估 Agent 输出。Judge call 不需要 policy call 使用的 rollout model integration，因此可以直接调用其他 LLM。Grading 仍会在 workflow 结束后同步执行；其延迟和失败会影响 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)))

基于工具调用的评分

评估 Agent 是否进行了工具调用，而不仅仅检查最终文本输出。Strands 会把工具调用记录为 assistant 消息上的 toolUse content block：

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)

您可以组合多种评分策略 —— 例如，检查 Agent 是否使用了正确的工具并且生成了正确的最终答案，然后对分数进行加权。

GraderConfig

自定义 grader configs 遵循与 AgentWorkflowConfig 相同的模式：继承 GraderConfig、创建 instance，并将它显式传给 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
        # ... 在评分逻辑中使用配置值 ...

my_grader_config = MyGraderConfig()

把 config instance 传给 LocalBackend(grader_config=my_grader_config)。Eval 和 training TOML 文件目前不会直接设置 grader config fields。 GraderConfig 扩展自 BaseConfig，并包含与 AgentWorkflowConfig 相同的 concurrency 字段，但当前 backends 不会用它限制 grader concurrency。如果 grader 会调用外部服务，请使用 eval [evaluation].batch_size、workflow/backend concurrency，或在 grader 内部显式加 limiter。

字段	类型	默认值	描述
`name`	`str`	（必填）	评分器的标识符
`description`	`str \| None`	`None`	可选描述
`concurrency`	`ConcurrencyConfig`	无限制	存在于 config model 上；当前 `LocalBackend` 不会强制执行

Entry Point 连接

Grader classes 和 config objects 需要显式 wiring。请在 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)

多个具体 Grader subclasses 可以在 entrypoint 中共存；只有通过 grader 传入的 class 会运行。Submit preflight 会导入 entrypoint 一次以暴露 constructor 和 dependency errors，不会检查它的 module namespace；关于该导入在何种情况下被跳过、以及它在本地执行了什么，参见 Rollout 中的文件。

下一步

评估

在训练前提交 evaluation run 测试您的 AgentWorkflow 和 Grader。

开始使用

核心 API

集成

执行

Agent 工具

Grader 基类

GraderContext

`set_reward`

写入 artifacts

RolloutSample

实现模式

精确匹配评分

LLM-as-Judge 评分

基于工具调用的评分

GraderConfig

Entry Point 连接

下一步

评估

​Grader 基类

​GraderContext

​set_reward

​写入 artifacts

​RolloutSample

​实现模式

​精确匹配评分

​LLM-as-Judge 评分

​基于工具调用的评分

​GraderConfig

​Entry Point 连接

​下一步

评估

Grader 基类

GraderContext

`set_reward`

写入 artifacts

RolloutSample

实现模式

精确匹配评分

LLM-as-Judge 评分

基于工具调用的评分

GraderConfig

Entry Point 连接

下一步