构建评分器 - Osmosis

Grader 类定义了如何评估和评分您的 Agent 输出。它产生驱动强化学习的 reward 信号 —— 更好的输出获得更高的 reward，较差的输出获得更低的 reward。

Grader 基类

from osmosis_ai.rollout import Grader, GraderContext

class MyGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        for sample_id, sample in ctx.samples.items():
            # 评估样本并分配 reward
            ctx.set_sample_reward(sample_id, 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

与 AgentWorkflow 类似，Grader 有一个抽象方法 —— grade() —— 接收包含 Agent 输出和当前数据集行参考答案的 GraderContext。

GraderContext

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

字段	类型	描述
`ctx.label`	`str \| None`	当前数据集行的参考答案（通常对应 `ground_truth` 列）
`ctx.metadata`	`dict[str, Any] \| None`	来自数据集可选 `metadata` 列的每行 metadata。该行无 metadata 时为 `None`。
`ctx.samples`	`dict[str, RolloutSample]`	以 sample ID 为键的 Agent 输出
`ctx.project_path`	`str \| None`	由执行 harness 提供的可选项目路径
`ctx.artifacts`	`dict[str, Any] \| None`	grader 附加的可选输出 JSON；初始为 `None`
`ctx.set_sample_reward(sample_id, reward)`	方法	为 sample 分配一个浮点数 reward
`ctx.set_artifacts(artifacts)`	方法	附加可选的输出 JSON payload（参见 Artifacts）

只要数据集行包含 label 或 metadata，Grader 就会运行，因此您可以仅依靠 metadata 驱动 reward signal（例如预期的工具调用或按行设定的评分规则）。

ctx.samples 以单次 workflow 执行内的 sample source ID 为键。使用内置 integrations 时，sample ID 通常来自 Strands agent name 或 OpenAI Agents session name。Evaluation run 和 training run 仍然可以对同一个 prompt 多次执行 workflow（evaluation config 中的 [evaluation].n，training config 中的 n_samples_per_prompt）；每次执行都会收到自己的 GraderContext。

`set_sample_reward`

调用 ctx.set_sample_reward(sample_id, reward) 为每个 sample 分配 reward。reward 应为浮点数 —— 通常在 0.0 到 1.0 之间，但接受任意浮点值。

ctx.set_sample_reward(sample_id, 0.85)

如果 sample_id 在 ctx.samples 中不存在，set_sample_reward 会抛出 ValueError。始终通过遍历 ctx.samples.items() 来确保使用有效的 sample ID。

Artifacts

ctx.set_artifacts(artifacts) 是一个可选的输出通道，用于在 reward 之外回传结构化 JSON —— 例如 judge 的解释、ID，或指向更大 trace 的引用，供前端展示。当 reward 本身无法解释你为什么这样打分时使用它。不调用该方法时，wire 上不会有任何变化，已有 callback 字节级别保持一致。

class JudgeGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        for sample_id, sample in ctx.samples.items():
            ctx.set_sample_reward(sample_id, 0.7)
        ctx.set_artifacts({
            "judge": {"explanation": "未满足最后一个约束。"},
            "trace_ref": {
                "path": "rollout_traces/run_123/sample_456.jsonl",
                "content_type": "application/jsonl",
                "size_bytes": 38291,
            },
        })

需要注意的规则：

传入一个可 JSON 序列化的 dict。不可序列化的值、NaN 或 Infinity 会被拒绝。
经过紧凑 UTF‑8 JSON 编码后，payload 上限为 64 KiB。
超大或无效 payload 会降级为一个小的 {"_error": {...}} 标记，因此 reward 始终能送达 —— 净化流程永远不会阻塞 reward 投递。
不要直接嵌入日志、trace 或二进制内容。请通过 {path|url, content_type, size_bytes} 引用，并把实际数据放在对象存储中。

ctx.metadata（输入侧，只读）与 ctx.artifacts（输出侧，由你设置）是两个独立的通道。把 metadata 看作数据集行的输入，把 artifacts 看作你希望平台在 reward 旁展示的内容。

RolloutSample

ctx.samples 中的每个条目都是一个 RolloutSample 对象，包含 AgentWorkflow 的输出：

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

from pydantic import BaseModel, Field


class RolloutSample(BaseModel):
    id: str
    messages: Sequence[Mapping[str, Any]] = Field(default_factory=list)
    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:
        for sample_id, sample in ctx.samples.items():
            answer = _last_text(sample).strip()
            reward = 1.0 if ctx.label and answer == ctx.label.strip() else 0.0
            ctx.set_sample_reward(sample_id, reward)

LLM-as-Judge 评分

使用另一个 LLM 来评估 Agent 输出的质量 —— 适用于正确性是主观的或难以通过程序化方式检查的场景。和 workflow 不同，grader 并不在训练路径上，因此可以直接调用任意 LLM：

import litellm
from osmosis_ai.rollout import Grader, GraderContext


class LLMJudgeGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        for sample_id, sample in ctx.samples.items():
            agent_output = _last_text(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_sample_reward(sample_id, 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:
        for sample_id, sample in ctx.samples.items():
            used_tool = False
            for m in sample.messages:
                if m.get("role") != "assistant":
                    continue
                content = m.get("content") or []
                if isinstance(content, list) and any(
                    isinstance(b, dict) and "toolUse" in b for b in content
                ):
                    used_tool = True
                    break
            ctx.set_sample_reward(sample_id, 1.0 if used_tool else 0.0)

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

GraderConfig

自定义评分器配置遵循与 AgentWorkflowConfig 相同的模式 —— 扩展 GraderConfig，并在 rollout entrypoint 中定义 module-level config instance：

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` 不会强制执行

自动发现

与 AgentWorkflow 类似，osmosis train submit preflight 可以从 entrypoint module 中发现您的 Grader 子类。无需 registration decorator，但 rollout entrypoint 仍需要把 grader class 和可选 config 传给它构造的 backend。

osmosis train submit 要求 rollout entrypoint 中有一个具体的 Grader。如果 SDK 没有发现 Grader，preflight validation 会失败，而不是分配默认 reward。

下一步

评估

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

​Grader 基类

​GraderContext

​set_sample_reward

​Artifacts

​RolloutSample

​实现模式

​精确匹配评分

​LLM-as-Judge 评分

​基于工具调用的评分

​GraderConfig

​自动发现

​下一步

评估

Grader 基类

GraderContext

`set_sample_reward`

Artifacts

RolloutSample

实现模式

精确匹配评分

LLM-as-Judge 评分

基于工具调用的评分

GraderConfig

自动发现

下一步