跳转到主要内容
Strands Agents 是 AWS 的 agent 框架，用于构建工具型 agents。如果您需要 Strands tools、Strands message handling，或想从已有 strands.Agent 直接迁移，请使用 Osmosis Strands integration。 osmosis-ai 包包含 strands-agents[litellm]，因此 starter rollouts 可以直接使用 Strands，无需添加额外依赖。

​
集成对象

ObjectPurpose
OsmosisStrandsAgentStrands Agent 的直接替代品，会把 samples 注册到活跃的 rollout context
OsmosisRolloutModel占位模型，会在运行时解析为当前 Osmosis policy
OsmosisStrandsAgent 保留常规 Strands constructor arguments，例如 tools、system_prompt、messages 和 callback handlers。

​
快速示例

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


class StrandsWorkflow(AgentWorkflow):
    async def run(self, ctx: AgentWorkflowContext) -> None:
        agent = OsmosisStrandsAgent(
            name="assistant",
            model=OsmosisRolloutModel(params={"temperature": 1.0}),
            messages=ctx.prompt,
            callback_handler=None,
        )
        await agent.invoke_async()
ctx.prompt 已经是当前 sample 可直接使用的输入。如果您的数据集行包含 system_prompt 和 user_prompt，SDK 会在 workflow 运行前组装这些字段。

​
Tools

按常规方式用 @tool 定义 Strands tools，然后传给 OsmosisStrandsAgent： 
from strands import tool
from osmosis_ai.rollout import AgentWorkflow, AgentWorkflowContext
from osmosis_ai.rollout.integrations.agents.strands import (
    OsmosisRolloutModel,
    OsmosisStrandsAgent,
)


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


class SearchWorkflow(AgentWorkflow):
    async def run(self, ctx: AgentWorkflowContext) -> None:
        agent = OsmosisStrandsAgent(
            name="search-agent",
            model=OsmosisRolloutModel(params={"temperature": 1.0}),
            tools=[search],
            system_prompt="You are a helpful assistant.",
            messages=ctx.prompt,
            callback_handler=None,
        )
        await agent.invoke_async()
对于大多数使用工具的 agents，一次 invoke_async() 调用就足够，因为 Strands 会在内部处理 model-tool loop。只有在需要额外停止条件，或需要跨多次调用设置硬上限时，才添加外层循环。

​
OsmosisRolloutModel

OsmosisRolloutModel 不接受 model_id。SDK 会在运行时使用占位符 model id openai/osmosis-rollout，并由 Osmosis 将其路由到当前策略。 请用 params dict 传入采样选项： 
model = OsmosisRolloutModel(
    params={
        "temperature": 1.0,
        "max_tokens": 1024,
    }
)
请在 AgentWorkflow.run() 内，或其他执行后端已经安装活跃 RolloutContext 的路径里构造 OsmosisStrandsAgent。在 module import time 构造会失败，因为此时还没有 rollout context。

​
Sample 收集方式

当使用 OsmosisRolloutModel 构造时，OsmosisStrandsAgent 会执行以下步骤：
1

读取 rollout context

它会从当前执行作用域读取活跃的 RolloutContext；如果没有可用 context，则抛出 RuntimeError。
2

选择 sample ID

它会从 agent 的 name、agent_id 或生成的 UUID 派生 sample ID。
3

解析模型

它会调用 model.for_sample(sample_id, rollout_ctx)，创建连接到 Osmosis chat completions endpoint 的真实 LiteLLM model。
4

注册 agent

它会把自身注册到 rollout context，让 backend 可以将 Strands message history 收集为 RolloutSample。
5

初始化 Strands Agent

它会用解析后的 model 委托给常规 Strands Agent constructor。Tools、prompts、messages 和 callbacks 都会原样传递。

​
完整示例

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


@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))


class MathWorkflow(AgentWorkflow):
    async def run(self, ctx: AgentWorkflowContext) -> None:
        agent = OsmosisStrandsAgent(
            name="math-agent",
            model=OsmosisRolloutModel(params={"temperature": 1.0}),
            tools=[calculator],
            system_prompt="You are a math assistant. Use the calculator tool.",
            messages=ctx.prompt,
            callback_handler=None,
        )
        await agent.invoke_async()


def _last_text(sample) -> str:
    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 MathGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        for sample_id, sample in ctx.samples.items():
            answer = _last_text(sample)
            reward = 1.0 if ctx.label and ctx.label.strip() in answer else 0.0
            ctx.set_sample_reward(sample_id, reward)

​
从 Strands Agent 迁移

如果您已有 Strands agent，可通过四步迁移：
1

替换 agent import

将 from strands import Agent 替换为：
from osmosis_ai.rollout.integrations.agents.strands import OsmosisStrandsAgent
2

替换 model

将固定 LiteLLM model 替换为 Osmosis 占位符：
from osmosis_ai.rollout.integrations.agents.strands import OsmosisRolloutModel

model = OsmosisRolloutModel(params={"temperature": 1.0})
删除 model_id；训练集群会决定服务哪个 policy。
3

替换 agent class

将 Agent(...) 替换为 OsmosisStrandsAgent(...)。保持 tools、system prompt、messages 和 callbacks 不变。
4

包装到 AgentWorkflow

将 agent construction 和 await agent.invoke_async() 调用移入 AgentWorkflow.run()。

​
Strands vs OpenAI Agents

Choose Strands whenChoose OpenAI Agents when
您的 tools 已经是 Strands @tool functions您的 workflow 已经使用 Runner.run 和 OpenAI Agents SDK sessions
您希望 sample.messages 中是 Strands message traces您希望 sample.messages 中是 persisted Responses API-style session items
您正在从 strands.Agent 迁移您正在从 OpenAI Agents SDK Agent 迁移
OpenAI Agents SDK 路径请参见 OpenAI Agents 集成。

​
下一步

构建 AgentWorkflow

查看共享 workflow contract。

构建 Graders

为 Strands agent 生成的 samples 打分。

评估

为您的 Strands rollout 提交 evaluation run。

OpenAI Agents 集成

对比 OpenAI Agents SDK integration。