跳转到主要内容

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 类定义了如何评估和评分您的 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.labelstr | None当前数据集行的参考答案（通常对应 ground_truth 列）
ctx.samplesdict[str, RolloutSample]以 sample ID 为键的 Agent 输出
ctx.workspace_pathstr | None工作区根目录路径
ctx.set_sample_reward(sample_id, reward)方法为 sample 分配一个浮点数 reward
ctx.samples 是一个字典，因为训练集群可能会对每个 prompt 运行多次 rollout（由 num_samples 训练参数控制）。每个 sample 代表对同一 prompt 的一次独立 AgentWorkflow 执行。

​
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。

​
RolloutSample

ctx.samples 中的每个条目都是一个 RolloutSample 对象，包含 AgentWorkflow 的输出： 
class RolloutSample(BaseModel):
    id: str                              # 唯一的 sample 标识符
    messages: list[MessageDict]          # Agent 产生的对话消息
    reward: float | None = None          # Reward（由 Grader 设置）
messages 列表就是您的 workflow 为该 sample 产出的对话记录。在很多 grader 里，您只需要从最后一条 assistant 消息中提取最终答案文本即可。
真实参考可以看 osmosis-sdk-python 仓库里的 examples/rollout/multiply_rollout/grader.py —— 它展示了从 sample 最后一条消息里提取文本的规范写法。

​
实现模式

​
精确匹配评分

最简单的评分策略就是把 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 以添加自定义字段： 
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):
    def __init__(self):
        super().__init__(config=MyGraderConfig())

    async def grade(self, ctx: GraderContext) -> None:
        threshold = self.config.similarity_threshold if self.config else 0.8
        # ... 在评分逻辑中使用配置值 ...
GraderConfig 扩展自 BaseConfig，并包含与 AgentWorkflowConfig 相同的 concurrency 字段：
字段类型默认值描述
namestr（必填）评分器的标识符
descriptionstr | NoneNone可选描述
concurrencyConcurrencyConfig无限制控制最大并发评分操作数

​
自动发现

与 AgentWorkflow 类似，SDK 会自动从入口模块中发现您的 Grader 子类。无需注册。
Grader 是可选的 —— 您的入口文件可以定义零个或一个 Grader 子类。如果未定义 Grader，所有 sample 将获得默认 reward。这在早期开发阶段非常有用，当您希望先专注于 Agent 行为再添加评估逻辑时。

​
下一步

本地评估

使用 eval 模式在本地测试您的 AgentWorkflow 和 Grader。

执行后端

在本地或远程基础设施上运行 rollout。