> ## 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. > ## Agent Instructions > Treat this site as the source of truth for public Osmosis behavior. > Distinguish the web Platform, the open source Python SDK, and the CLI. > Use documented commands, configuration fields, and public APIs exactly as written; do not infer internal endpoints or services. # AgentWorkflow > 实现 AgentWorkflow 类以定义训练中的 Agent 行为 `AgentWorkflow` 是 rollout 行为的 SDK contract。您需要继承它、实现一个 async `run()` 方法，并通过返回 message history，或通过 Osmosis 支持的 agent integration 调用当前策略并注册 sample source，来创建一个 sample。 Workflow 应该回答一个问题：**给定这个数据集 prompt，agent 在 grader 打分之前应该做什么？** ## 基类 ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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 形状如下： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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`： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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 失败。 ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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](https://www.harborframework.com/docs/agents/trajectory-format)（Agent Trajectory Interchange Format）文档保存在 run artifacts 旁。Persistence 采用 best-effort 策略，永远不会改变 reward 或 rollout 状态。在自定义 harness 中直接调用 `LocalBackend.run_workflow()` 或其他 backend 方法不会安装这个 server lifecycle；该 harness 必须自行保存所需 trajectory。 文件位于平台管理的主机路径下： ``` ~/.osmosis// ├── 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()`： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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 集成](/zh/sdk/integrations/strands)。 ## OpenAI Agents 模式 对于 OpenAI Agents，请构造 `OsmosisAgent`，附加 OpenAI Agents `ModelSettings`，创建一个 `OsmosisMemorySession`，并把该 session 传给 `Runner.run()`： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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 集成](/zh/sdk/integrations/openai-agents)。 ## 自定义配置 自定义 configs 继承 `AgentWorkflowConfig`。请在 rollout entrypoint 中定义 module-level config instance，并把它传给 backend： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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 中： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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： ```python theme={"theme":{"light":"github-light","dark":"github-dark"},"languages":{"custom":["/languages/cli.json"]}} 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 中的文件](/zh/sdk/overview#files-in-a-rollout)。 ## 下一步 使用 tools 和 `OsmosisStrandsAgent` 构建基于 Strands 的 rollout。 使用 `OsmosisAgent` 和 `OsmosisMemorySession` 构建 OpenAI Agents SDK rollout。 为 workflow 生成的 sample 定义 reward 逻辑。 在训练前提交 evaluation run 测试 workflow 和 grader。