> ## Documentation Index > Fetch the complete documentation index at: https://docs.osmosis.ai/llms.txt > Use this file to discover all available pages before exploring further. # 构建评分器 > 实现 Grader 类以定义训练的 reward 信号 `Grader` 类定义了如何评估和评分您的 Agent 输出。它产生驱动强化学习的 reward 信号 —— 更好的输出获得更高的 reward，较差的输出获得更低的 reward。 ## Grader 基类 ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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 中的基类签名： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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](#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 之间，但接受任意浮点值。 ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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 字节级别保持一致。 ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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 的输出： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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` 直接比较。下面的辅助函数演示了如何从最后一条消息中提取文本： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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。