Method step

agentlightning/runner/agent.py:794–841

Execute a single task directly, bypassing the task queue. This method creates a new rollout for the given input and executes it immediately. Unlike [`iter()`][agentlightning.LitAgentRunner.iter], exceptions are propagated to the caller. Args: input: The

(
        self,
        input: T_task,
        *,
        resources: Optional[NamedResources] = None,
        mode: Optional[RolloutMode] = None,
        event: Optional[ExecutionEvent] = None,
    )

        logger.info(f"{self._log_prefix()} Finished async rollouts. Processed {num_tasks_processed} tasks.")

    async def step(
        self,
        input: T_task,
        *,
        resources: Optional[NamedResources] = None,
        mode: Optional[RolloutMode] = None,
        event: Optional[ExecutionEvent] = None,
    ) -> Rollout:
        """Execute a single task directly, bypassing the task queue.

        This method creates a new rollout for the given input and executes it
        immediately. Unlike [`iter()`][agentlightning.LitAgentRunner.iter],
        exceptions are propagated to the caller.

        Args:
            input: The task input to be processed by the agent.
            resources: Optional named resources to be used for this specific task.
                If provided, a new resources entry will be created in the store.
                If not provided, the latest resources from the store will be used.
            mode: Optional rollout mode ("train" or "validation"). If not provided,
                the agent's default mode will be used.
            event: Optional ExecutionEvent object to signal interruption (currently unused
                but included for interface consistency).

        Returns:
            The completed rollout.

        Raises:
            Exception: Any exception that occurs during rollout execution will be
                re-raised to the caller.
        """
        store = self.get_store()

        if resources is not None:
            resources_update = await store.add_resources(resources)
            resources_id = resources_update.resources_id
        else:
            resources_id = None

        attempted_rollout = await self.get_store().start_rollout(
            input=input, mode=mode, resources_id=resources_id, worker_id=self.get_worker_id()
        )
        rollout_id = await self._step_impl(attempted_rollout, raise_on_exception=True)

        completed_rollout = await store.get_rollout_by_id(rollout_id)
        if completed_rollout is None:
            raise RuntimeError(f"{self._log_prefix()} Failed to fetch completed rollout by id after step: {rollout_id}")
        return completed_rollout