Skip to main content

​
Goals

  • Learn the different ways to access tasks from an environment
  • Understand when to use listing vs index-based access
  • Learn how to implement task methods in your environment server

​
Prerequisites

​
Introduction

Tasks are JSON objects that describe a specific problem or challenge for an agent to solve. Each environment organises tasks into splits (e.g. "train", "test"). The SDK provides several ways to access tasks depending on your needs — from listing all tasks in a split, to fetching individual tasks by index, to retrieving a range of tasks for partitioned workloads.

​
Listing Tasks

You can list all tasks for a split: 
tasks = await environment.list_tasks(split="train")
This returns the full list of task objects for the given split. It’s the simplest approach and works well when your task set is small enough to load in one call.

​
Index-Based Access

For environments with large task sets, you can query the number of tasks and fetch individual tasks by index without loading the entire list: 
# Get the number of tasks in a split
num = await environment.num_tasks(split="train")
print(f"{num} tasks available")

# Fetch a single task by index
task = await environment.get_task(split="train", index=0)

​
Getting a Range of Tasks

You can also fetch a contiguous range of tasks by index. This is useful for partitioning tasks across multiple workers: 
# Get tasks 0 through 9
tasks = await environment.get_task_range(split="train", start=0, stop=10)
Task ordering is consistent across calls, so ranges are safe to use for partitioning workloads.

​
Implementing Tasks in Your Environment

When building an environment server, implement list_tasks, list_splits, and optionally override get_task and num_tasks: 
from openreward.environments import Environment, Split, JSONObject

class MyEnvironment(Environment):
    @classmethod
    def list_splits(cls):
        return [Split("train", type="train"), Split("test", type="test")]

    @classmethod
    def list_tasks(cls, split: str) -> list[JSONObject]:
        if split == "train":
            return [{"task_id": str(i)} for i in range(1000)]
        return [{"task_id": "0"}]
The base class provides default implementations of num_tasks and get_task that call list_tasks internally. If your task set is large or expensive to load, you can override these for better performance: 
    @classmethod
    async def num_tasks(cls, split: str) -> int:
        # Return count without loading all tasks
        return db.count_tasks(split)

    @classmethod
    async def get_task(cls, split: str, index: int) -> JSONObject:
        # Fetch a single task without loading the full list
        return db.get_task(split, index)

​
Next Steps

Using Task-Specific Tools

Define tools that vary per task for environments with multiple tool interfaces