AgentWorkflow

AgentWorkflow 是 rollout 行为的 SDK contract。您需要继承它、实现一个 async run() 方法，并通过返回 message history，或通过 Osmosis 支持的 agent integration 调用当前策略并注册 sample source，来创建一个 sample。 Workflow 应该回答一个问题：给定这个数据集 prompt，agent 在 grader 打分之前应该做什么？

基类

from osmosis_ai.rollout import AgentWorkflow, AgentWorkflowContext


class MyWorkflow(AgentWorkflow):
    async def run(self, ctx: AgentWorkflowContext) -> None:
        # Build and run your agent here.
        pass

SDK 形状如下：

class AgentWorkflow[TConfig: AgentWorkflowConfig](ABC):
    def __init__(self, config: TConfig | None = None):
        self.config = config

    @abstractmethod
    async def run(
        self, ctx: AgentWorkflowContext[TConfig]
    ) -> AgentWorkflowOutput | list[dict[str, Any]] | None:
        raise NotImplementedError

run() 会为每次 workflow execution 调用一次。在方法内部构造任何 per-execution agent 或 session objects 并运行 agent。然后直接返回 sample，或让 integration 把生成的 conversation 注册到活跃的 RolloutContext。

Workflow 返回值

run() 接受三种形状，都会被规范化为单个 sample：

返回值	效果
`AgentWorkflowOutput`	`messages` 字段成为 sample；可选的 `metrics` 会附加上去
`list[dict[str, Any]]`	被包装成 `AgentWorkflowOutput(messages=...)`
`None`	SDK 从活跃 `RolloutContext` 上注册的 source 收集 sample

受支持的 integrations 会为您注册 sample source，因此大多数 workflow 返回 None：OsmosisStrandsAgent 注册其 Strands history，OsmosisMemorySession 注册 OpenAI Agents SDK session。当您自己构建 message history 时返回 AgentWorkflowOutput：

from osmosis_ai.rollout import (
    AgentWorkflow,
    AgentWorkflowContext,
    AgentWorkflowOutput,
)


class ExplicitWorkflow(AgentWorkflow):
    async def run(self, ctx: AgentWorkflowContext) -> AgentWorkflowOutput:
        messages = [*ctx.prompt, {"role": "assistant", "content": "done"}]
        return AgentWorkflowOutput(
            messages=messages,
            metrics={"steps": 1.0},
        )

AgentWorkflowOutput.info 是保留字段，目前不会传给 grader。有限数值的 sample measurements 请放在 metrics 中，文件请写入 ctx.artifacts_dir。

AgentWorkflowOutput 拒绝未知的顶层字段和非有限的 metric 值（NaN、inf、-inf）。相同的校验会在 Local 与 Harbor/container 执行中运行。

AgentWorkflowContext

ctx object 为 workflow 提供输入和 config：

字段	类型	说明
`ctx.prompt`	`list[dict[str, Any]]`	当前数据集行的输入 messages
`ctx.config`	`TConfig \| None`	如果提供了，则为自定义 workflow config object
`ctx.metadata`	`dict[str, Any] \| None`	来自数据集可选 `metadata` 列的每行 metadata。该行无 metadata 时为 `None`。
`ctx.artifacts_dir`	`pathlib.Path \| None`	每次 rollout 的目录，用于写入日志、trace 和其他输出文件。当执行环境无法提供可写目录时为 `None`，写入文件前请先检查它是否为 `None`。

如果您的数据集行包含 system_prompt、user_prompt 和 ground_truth，prompt 字段会被组装进 ctx.prompt。参考答案不会传给 workflow；它会作为 ctx.label 暴露给 grader。AgentWorkflowContext 和 GraderContext 上的 metadata 对象相同，因此 workflow 和 grader 可以读取同一份按行的上下文。

不要把 task answers 放进 AgentWorkflow.run()。Workflow 应该产出行为；Grader 应该决定该行为是否值得 reward。

写入 artifacts

使用 ctx.artifacts_dir 写入不适合放进 sample payload 的文件 —— 日志、trace、截图或其他大文件/二进制输出。每次 rollout 都有独立目录，但当执行环境无法提供时为 None，因此写入前请用 if ctx.artifacts_dir: 进行判断 —— 未加判断的写入会抛出异常并导致 workflow 失败。

from osmosis_ai.rollout import AgentWorkflow, AgentWorkflowContext


class TracingWorkflow(AgentWorkflow):
    async def run(self, ctx: AgentWorkflowContext) -> None:
        if ctx.artifacts_dir:
            (ctx.artifacts_dir / "run.log").write_text("started\n")
        ...

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

保存的 Trajectory

使用 create_rollout_server() 创建的 rollout server 会把已完成的 samples 作为 ATIF（Agent Trajectory Interchange Format）文档保存在 run artifacts 旁。Persistence 采用 best-effort 策略，永远不会改变 reward 或 rollout 状态。在自定义 harness 中直接调用 LocalBackend.run_workflow() 或其他 backend 方法不会安装这个 server lifecycle；该 harness 必须自行保存所需 trajectory。文件位于平台管理的主机路径下：

~/.osmosis/<rollout_id>/
├── trajectory.json          # rollout sample 的 ATIF 文档
└── artifacts/...            # 你在 ctx.artifacts_dir 下写入的文件

RolloutSample.messages 会保留 grader 读取的 framework-native history。对于 ATIF persistence，内置 integrations 会在 trajectory_messages 中创建一份独立的 best-effort normalized copy；normalization 失败时 native sample 保持不变，并跳过 trajectory persistence。显式 workflow output 在可复制时会将返回的 messages 用于两个 views。

如果你构建的自定义 sample source 的原生历史不是 OpenAI chat-completions 形式，请在返回的 sample 上设置 RolloutSample.trajectory_messages 来控制持久化内容（显式设为 None 会跳过该 sample 的 trajectory 保存）。

只有 source messages 或 server report 提供 usage、model 和 timestamp 数据时，ATIF 才会包含这些字段。缺失的 metadata 会保持缺失；SDK 不会虚构它们。

模型路由要求

run() 内的 LLM 调用必须通过执行后端安装的 RolloutContext 路由。训练集群使用该 context 中的 rollout-scoped chat-completions URL 服务当前策略、收集 traces，并把 reward 连接到 sample。

请使用以下受支持 integrations 之一：

框架	适用场景	Integration 对象
Strands Agents	Strands tools、Strands message history、从 `strands.Agent` 迁移	`OsmosisStrandsAgent`、`OsmosisRolloutModel`
OpenAI Agents SDK	`Runner.run`、sessions、handoffs、OpenAI-style tools	`OsmosisAgent`、`OsmosisRolloutModel`、`OsmosisMemorySession`

不要在 run() 中用硬编码 policy model 直接调用 litellm、OpenAI SDK 或其他 provider SDK。直接调用会绕过 rollout context，无法用于训练。

Strands 模式

对于 Strands，直接把 ctx.prompt 作为 messages 传入，并调用 invoke_async()：

from osmosis_ai.rollout import AgentWorkflow, AgentWorkflowContext
from osmosis_ai.rollout.integrations.agents.strands import (
    OsmosisRolloutModel,
    OsmosisStrandsAgent,
)


class SimpleStrandsWorkflow(AgentWorkflow):
    async def run(self, ctx: AgentWorkflowContext) -> None:
        agent = OsmosisStrandsAgent(
            name="simple-strands-agent",
            model=OsmosisRolloutModel(params={"temperature": 1.0}),
            messages=ctx.prompt,
            callback_handler=None,
        )
        await agent.invoke_async()

在 run() 内构造 OsmosisStrandsAgent 会将其绑定到活跃的 rollout context，并把 agent 注册为 sample source。工具 examples、迁移步骤和 OsmosisRolloutModel 细节请参见 Strands 集成。

OpenAI Agents 模式

对于 OpenAI Agents，请构造 OsmosisAgent，附加 OpenAI Agents ModelSettings，创建一个 OsmosisMemorySession，并把该 session 传给 Runner.run()：

from agents import ModelSettings, Runner
from osmosis_ai.rollout import AgentWorkflow, AgentWorkflowContext
from osmosis_ai.rollout.integrations.agents.openai_agents import (
    OsmosisAgent,
    OsmosisMemorySession,
    OsmosisRolloutModel,
)


class SimpleOpenAIWorkflow(AgentWorkflow):
    async def run(self, ctx: AgentWorkflowContext) -> None:
        agent = OsmosisAgent(
            name="simple-openai-agent",
            instructions="Answer the user's request clearly.",
            model=OsmosisRolloutModel(),
            model_settings=ModelSettings(temperature=1.0, max_tokens=4096),
        )
        session = OsmosisMemorySession()
        await Runner.run(
            agent,
            ctx.prompt,
            session=session,
        )

Session 会记录 OpenAI Agents SDK conversation 以供 grader 使用。请在 run() 内创建它，使其注册到当前 RolloutContext。 Session 行为、tracing 说明和迁移步骤请参见 OpenAI Agents 集成。

自定义配置

自定义 configs 继承 AgentWorkflowConfig。请在 rollout entrypoint 中定义 module-level config instance，并把它传给 backend：

from osmosis_ai.rollout import (
    AgentWorkflow,
    AgentWorkflowConfig,
    AgentWorkflowContext,
    ConcurrencyConfig,
)


class SearchWorkflowConfig(AgentWorkflowConfig):
    name: str = "search-workflow"
    max_iterations: int = 8
    temperature: float = 1.0
    concurrency: ConcurrencyConfig = ConcurrencyConfig(max_concurrent=4)


class SearchWorkflow(AgentWorkflow[SearchWorkflowConfig]):
    async def run(self, ctx: AgentWorkflowContext[SearchWorkflowConfig]) -> None:
        config = ctx.config or SearchWorkflowConfig()
        max_iterations = config.max_iterations
        temperature = config.temperature
        # Use these values when constructing your agent.


search_workflow_config = SearchWorkflowConfig()

请把 config instance 显式传给 backend constructor，例如 LocalBackend(workflow=SearchWorkflow, workflow_config=search_workflow_config)。Eval 和 training TOML 文件目前不会直接设置 workflow config fields。 BaseConfig 允许额外字段，因此简单 rollout configs 通常不需要额外的 Pydantic 样板代码。

字段	类型	默认值	说明
`name`	`str`	required	workflow 标识符
`description`	`str \| None`	`None`	可选描述
`concurrency`	`ConcurrencyConfig`	unlimited	最大并发 workflow executions

使用工具的 Workflows

工具使用属于 agent framework，而不是 backend。请按您的 framework 期望的方式定义 tools，把它们传给 Osmosis-wrapped agent，并让 integration 记录生成的 messages。例如，Strands workflow 可以把 tool list 放在 config 中：

from typing import Any

from strands import tool
from osmosis_ai.rollout import (
    AgentWorkflow,
    AgentWorkflowConfig,
    AgentWorkflowContext,
)
from osmosis_ai.rollout.integrations.agents.strands import (
    OsmosisRolloutModel,
    OsmosisStrandsAgent,
)


@tool(name="search")
def search_tool(query: str) -> str:
    """Search for information."""
    return f"results for {query}"


class ToolWorkflowConfig(AgentWorkflowConfig):
    name: str = "tool-workflow"
    model: Any = OsmosisRolloutModel(params={"temperature": 1.0})
    tools: Any = [search_tool]
    max_iterations: int = 8


tool_workflow_config = ToolWorkflowConfig()


class ToolWorkflow(AgentWorkflow[ToolWorkflowConfig]):
    async def run(self, ctx: AgentWorkflowContext[ToolWorkflowConfig]) -> None:
        config = ctx.config or ToolWorkflowConfig()
        agent = OsmosisStrandsAgent(
            name="search-agent",
            model=config.model,
            tools=config.tools,
            messages=ctx.prompt,
            callback_handler=None,
        )

        for _ in range(config.max_iterations):
            result = await agent.invoke_async()
            content = result.message.get("content", [])
            if not any("toolUse" in block for block in content):
                break

Entry Point 连接

Workflow classes 和 config objects 需要显式 wiring。请在 backend constructor 中选择它们，并通过 rollout server 暴露该 backend：

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

backend = LocalBackend(
    workflow=SearchWorkflow,
    workflow_config=search_workflow_config,
)
app = create_rollout_server(backend=backend)

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

下一步

Strands 集成

使用 tools 和 OsmosisStrandsAgent 构建基于 Strands 的 rollout。

OpenAI Agents 集成

使用 OsmosisAgent 和 OsmosisMemorySession 构建 OpenAI Agents SDK rollout。

Grader

为 workflow 生成的 sample 定义 reward 逻辑。

评估

在训练前提交 evaluation run 测试 workflow 和 grader。

开始使用

核心 API

集成

执行

Agent 工具

基类

Workflow 返回值

AgentWorkflowContext

写入 artifacts

保存的 Trajectory

模型路由要求

Strands 模式

OpenAI Agents 模式

自定义配置

使用工具的 Workflows

Entry Point 连接

下一步

Strands 集成

OpenAI Agents 集成

Grader

评估

​基类

​Workflow 返回值

​AgentWorkflowContext

​写入 artifacts

​保存的 Trajectory

​模型路由要求

​Strands 模式

​OpenAI Agents 模式

​自定义配置

​使用工具的 Workflows

​Entry Point 连接

​下一步

Strands 集成

OpenAI Agents 集成

Grader

评估

基类

Workflow 返回值

AgentWorkflowContext

写入 artifacts

保存的 Trajectory

模型路由要求

Strands 模式

OpenAI Agents 模式

自定义配置

使用工具的 Workflows

Entry Point 连接

下一步