RolloutClient 与可等待的 Rollout Handle

RolloutClient 向 rollout server 提交请求，并轮询结果直到 rollout 结束。osmosis eval run 在内部使用它。当您在自托管 harness 或自定义评估循环中驱动 rollout server 时，可以自行创建客户端。

从 0.3.3 起，RolloutClient 随基础 osmosis-ai package 提供，从 osmosis_ai.rollout.client 导入。服务端使用 create_rollout_server()，需要 server extra。请同时升级调用方和服务端；参见升级到 0.3.3。

运行 Rollout 并等待完成

客户端接收服务端连接设置和可选的 admission deadline。Chat completions URL、llm_api_key、是否 grading、metadata 和各阶段 timeout 则属于每次 rollout 请求。此示例假定 rollout server 已配置 grader，且 chat endpoint 可达；endpoint 要求认证时，请设置 ROLLOUT_LLM_API_KEY：

import asyncio
import os
from uuid import uuid4

import httpx

from osmosis_ai.rollout.client import RolloutClient


async def main():
    async with httpx.AsyncClient() as http_client:
        client = RolloutClient(
            url="http://127.0.0.1:8000",
            http_client=http_client,
            admission_timeout_sec=60,
        )
        result = await client.run_rollout(
            initial_messages=[{"role": "user", "content": "What is 6 * 7?"}],
            chat_completions_url="http://127.0.0.1:9000/v1",
            rollout_id=str(uuid4()),
            llm_api_key=os.environ.get("ROLLOUT_LLM_API_KEY"),
            label="42",
            grade=True,
        )
        print(result.status, result.sample.reward if result.sample else None)


asyncio.run(main())

run_rollout() 异步等待 rollout 结束，并返回终态 RolloutResultResponse。检查 status、sample、err_message 和 err_category，以区分成功与执行失败。传入 grade=False 可跳过该请求的 grading；需要 grading 的 LocalBackend 请求必须同时满足已配置 grader 且包含 label 或 metadata。每次新尝试都使用新的 rollout_id。llm_api_key 用于 chat endpoint 认证，不用于 rollout server 认证。要向 rollout server 认证，请参见向 Rollout Server 认证。所有 handle 完成之前，请保持 HTTP client 打开。如果让 RolloutClient 自行创建 HTTP client，请在使用结束后调用 await client.aclose()。

使用 Rollout Handle 跟踪进度

await client.run_rollout_async() 在 admission 完成后返回可等待的 RolloutHandle，并启动后台轮询。调用以下辅助函数时传入尚未关闭的客户端，并在关闭客户端之前等待函数完成：

from uuid import uuid4

from osmosis_ai.rollout.client import RolloutClient
from osmosis_ai.rollout.types import RolloutResultResponse


async def track_rollout(
    client: RolloutClient,
    chat_completions_url: str,
    llm_api_key: str | None = None,
) -> RolloutResultResponse:
    rollout = await client.run_rollout_async(
        initial_messages=[{"role": "user", "content": "What is 6 * 7?"}],
        chat_completions_url=chat_completions_url,
        rollout_id=str(uuid4()),
        llm_api_key=llm_api_key,
        label="42",
        grade=True,
    )
    phase = await rollout.wait_for_grading()
    print(phase)
    return await rollout

等待 handle 会返回与 run_rollout() 相同的终态 RolloutResultResponse。Handle 还提供以下生命周期接口：

成员	行为
`await handle`	返回终态 `RolloutResultResponse`
`wait_for_running()` / `wait_for_grading()`	当到达或经过对应阶段，或 rollout 结束时，返回最新状态
`wait_for_completion()`	等同于等待 handle
`status`	最近观察到的 `RolloutStatus`
`latest_result`	首次轮询后的最新结果响应；此前为 `None`
`done()`	轮询任务是否已结束，包括异常或取消
`cancel()`	请求取消客户端轮询任务，不联系服务端

生命周期状态为 queued、running 和 grading；终态为 success、failure 和 cancelled。Rollout 可能跳过阶段，轮询也可能未观察到中间状态。特别是，跳过 grading 或执行失败时，wait_for_grading() 可以返回终态。结果响应不包含仅用于持久化的 trajectory_messages 字段。

Admission、Lease 与 Timeout

服务端在 admission 时创建 polling lease，并决定长轮询等待时间和 lease timeout。客户端在每次结果请求中通过 X-Osmosis-Rollout-Lease 发送返回的 token；每个有效请求都会续期。您无需提供 lease token 或等待时长。

Admission 遇到 HTTP 429 时会自动重试。与活跃或保留结果的 rollout 重复的 rollout_id 会收到 HTTP 409。
在客户端设置 admission_timeout_sec 时，该值必须有限，并限制整个 admission 过程，包括 HTTP 请求和重试等待；它不限制 admission 之后的执行时间。
HTTP 请求期间触发 admission deadline，并不能证明服务端已拒绝 rollout。RolloutAdmissionTimeoutError 会说明 admission 可能已经成功。客户端不会自动按 ID 取消，因为如果丢失的是重复 ID 拒绝响应，取消可能误伤另一个活跃 rollout。
如果轮询停止并导致 lease 过期，服务端会发布错误类别为 lease_expired 的 failure，并请求取消。此 lease failure 可能在执行清理完成之前就已可见。

在 0.3.7 中，每次结果 GET 有两套独立的重试计划：

失败类型	重试等待时间
`httpx.RemoteProtocolError`、`httpx.NetworkError` 或 `httpx.ReadTimeout`	最多重试两次，分别等待 0.1 秒和 0.5 秒
HTTP 502、503 或 504	最多重试六次，分别等待 0.25、0.5、1、2、4 和 8 秒

重试复用 rollout ID 和 polling lease。只有当退避时间加上一次完整请求能在最后确认的 lease 截止时间之前完成时才会开始重试；失败的请求不会延长该期限。持续的网关错误会抛出带有最后状态码的 RolloutProtocolError。连接、写入和连接池超时、其他 HTTP 错误（包括 500 和所有 4xx）以及无效结果响应会立即失败。取消仍会向上传播，结果重试不会重放 admission 或 cancellation 请求。请移除自定义结果 GET 重试包装器，避免嵌套重试。升级调用方环境后即可使用此恢复行为。

取消 Rollout

await client.cancel_rollout(rollout_id) 请求服务端取消 rollout，整个请求受五秒墙钟时间限制。对于内置 backend（包括 LocalBackend），取消是幂等操作。请求响应只确认取消；请继续轮询以观察清理后的终态结果：

from osmosis_ai.rollout.client import RolloutClient, RolloutHandle
from osmosis_ai.rollout.types import RolloutResultResponse


async def cancel_and_wait(
    client: RolloutClient, rollout: RolloutHandle
) -> RolloutResultResponse:
    await client.cancel_rollout(rollout.rollout_id)
    return await rollout

handle.cancel() 在本地取消活跃的轮询任务，不会向服务端发送取消请求；等待已取消的 handle 会抛出 asyncio.CancelledError。如果需要等待服务端清理，请使用 client.cancel_rollout() 并保持 handle 轮询。

向 Rollout Server 认证

向 create_rollout_server() 传入 api_key，即可要求每个路由都携带 bearer token，包括 /health、admission、结果轮询和取消。缺少匹配的 Authorization: Bearer <api_key> header 的请求会收到 HTTP 401。向 RolloutClient(api_key=...) 传入相同的值，客户端会在每个请求中发送该 header：

import os

from osmosis_ai.rollout.client import RolloutClient
from osmosis_ai.rollout.server import create_rollout_server

app = create_rollout_server(
    backend=backend,
    api_key=os.environ["ROLLOUT_SERVER_API_KEY"],
)

client = RolloutClient(
    url="http://127.0.0.1:8000",
    api_key=os.environ["ROLLOUT_SERVER_API_KEY"],
)

认证是可选的。未传入 api_key 创建的服务端仍像以前一样接受未认证请求。服务端和客户端在 api_key 为空或只包含空白字符时都会抛出 ValueError。客户端不会跟随重定向，因此不会把 key 转发到其他 URL。

检查健康状态并等待空闲

在重启、重新部署或缩减 rollout server 之前，可以使用以下生命周期方法观察服务端状态。wait_idle() 观察工作是否完成，但不会阻止新的 admission，也不确认 artifacts 已上传到外部存储。

方法	行为
`await client.health()`	返回 `GET /health` 的 payload：backend 健康字段，以及 `instance_id`、`process_id` 和 `lifecycle`
`await client.wait_idle(timeout_sec=120, poll_interval_sec=1, process_id=None)`	轮询 `/health`，直到 `lifecycle.active_rollouts` 为 `0`，并返回该健康 payload。它不会停止新的 admission

lifecycle 对象报告 accepting_rollouts 和 active_rollouts。在服务端完成执行、artifact 归档、trajectory 持久化和结果定稿之前，rollout 都算作活跃。process_id 对每个服务端进程唯一，因此该值变化表示服务端已重启。 wait_idle() 会在 timeout_sec 到期前重试传输错误以及 HTTP 408、429、500、502、503 和 504 响应，到期后抛出 TimeoutError。timeout_sec 和 poll_interval_sec 都必须是有限的正数。当服务端未报告生命周期元数据（例如旧版服务端）时，它会抛出 RolloutProtocolError。轮询期间 process_id 发生变化，或与您传入的 process_id 不同时，它也会抛出该错误。要确认某个服务端进程已完成活跃工作，请将等待绑定到健康检查返回的 process_id：

from osmosis_ai.rollout.client import RolloutClient


async def wait_until_idle(client: RolloutClient) -> None:
    health = await client.health()
    await client.wait_idle(timeout_sec=120, process_id=health["process_id"])

要停止 Uvicorn rollout server，请向其发送 SIGTERM。Uvicorn 会先停止接受连接，并等待进行中的 HTTP 请求（包括结果长轮询）结束。随后应用关闭 admission，等待活跃工作最多 10 秒，再请求取消并允许最多 120 秒完成清理，随后取消剩余任务并再等待 10 秒，最后调用 backend.shutdown()。Uvicorn 开始关闭后，调用方无法再发起新的结果轮询；如果需要拿到每个结果，请先停止提交 rollout，并在发送 SIGTERM 前调用 wait_idle()。Host 的终止宽限期应涵盖这些阶段及 backend 的额外清理时间。无需先发送 HTTP drain 请求。

Execution Backends

选择在 rollout server 后运行 workflow 的 backend。

LocalBackend

配置进程内执行、grading、concurrency、artifacts 和错误处理。

开始使用

核心 API

集成

执行

Agent 工具

RolloutClient 与可等待的 Rollout Handle

运行 Rollout 并等待完成

使用 Rollout Handle 跟踪进度

Admission、Lease 与 Timeout

取消 Rollout

向 Rollout Server 认证

检查健康状态并等待空闲

相关页面

Execution Backends

LocalBackend

​运行 Rollout 并等待完成

​使用 Rollout Handle 跟踪进度

​Admission、Lease 与 Timeout

​取消 Rollout

​向 Rollout Server 认证

​检查健康状态并等待空闲

​相关页面

Execution Backends

LocalBackend

运行 Rollout 并等待完成

使用 Rollout Handle 跟踪进度

Admission、Lease 与 Timeout

取消 Rollout

向 Rollout Server 认证

检查健康状态并等待空闲

相关页面