To create a new task, inherit from the Task class and implement its required interface.
Required Methods:
__init__(self, name: str, concurrency: int = 1, *args, **kwargs): Initializes the task. name is typically specified in the config, and concurrency defines the maximum concurrency supported within a single worker.get_indices(self) -> List[SampleIndex]: Returns a list of all test sample indices. SampleIndex can be an int or str.async start_sample(self, index: SampleIndex, session: Session) -> TaskSampleExecutionResult: Contains the logic for a single test sample. Use the session object to interact with the Agent.calculate_overall(self, results: List[TaskOutput]) -> Dict[str, Any]: Calculates the final score/metrics after all samples are tested. The returned dictionary is saved to overall.json.release(self): Performs cleanup after the entire worker process finishes (not after each sample).
Interacting with the Agent via Session:
session.inject(item: Union[ChatHistoryItem, List[ChatHistoryItem]]): Inserts one or more history records into the session.await session.action(*injection) -> AgentOutput: Waits for the Agent's response. You can optionally pass history items to be injected simultaneously.
Handling AgentOutput:
When calling session.action, always check the AgentOutputStatus. If the status is CANCELLED, you must terminate the sample execution quickly to avoid impacting subsequent tests.
class VirtualTask(Task):
def __init__(self, *args, **kwargs) -> None:
super().__init__(name="virtual-task", *args, **kwargs)
def get_indices(self) -> List[Any]:
return list(range(10))
async def start_sample(self, index, session: Session):
print("task start sample")
for loop_times in range(3):
await asyncio.sleep(1)
res = await session.action(
{"role": "user", "content": "Loop: %d" % loop_times}
)
print("TASK", res.content)
return TaskSampleExecutionResult(
status=SampleStatus.COMPLETED,
result={"result": "ok"},
)
def calculate_overall(self, results: List[TaskOutput]) -> Dict[str, Any]:
return {"score": 0.4}
