Batch operations

Defines the batch operations for an LM invoker.

Authors

Henry Wicaksono (henry.wicaksono@gdplabs.id)

References

NONE

BatchOperations(invoker)

Handles batch operations for an LM invoker.

This class provides a wrapper around the batch operations of an LM invoker. It provides a simple interface to perform batch invocation:

results = await lm_invoker.batch.invoke(...)

Additionally, it also supports the following standalone batch operations:

1. Create a batch job:

batch_id = await lm_invoker.batch.create(...)

1. Get the status of a batch job:

status = await lm_invoker.batch.status(batch_id)

1. Retrieve the results of a batch job:

results = await lm_invoker.batch.retrieve(batch_id)

1. List the batch jobs:

batch_jobs = await lm_invoker.batch.list()

1. Cancel a batch job:

await lm_invoker.batch.cancel(batch_id)

Initializes the batch operations.

Parameters:

Name	Type	Description	Default
invoker	BaseLMInvoker	The LM invoker to use for the batch operations.	required

cancel(batch_id) async

Cancels a batch job.

Parameters:

Name	Type	Description	Default
batch_id	str	The ID of the batch job to cancel.	required

create(requests, hyperparameters=None) async

Creates a new batch job.

Parameters:

Name	Type	Description	Default
requests	dict[str, LMInput]	The dictionary of requests that maps request ID to the request. Each request must be a valid input for the language model. 1. If the request is a list of Message objects, it is used as is. 2. If the request is a list of MessageContent or a string, it is converted into a user message.	required
hyperparameters	dict[str, Any] | None	A dictionary of hyperparameters for the language model. Defaults to None, in which case the default hyperparameters are used.	None

Returns:

Name	Type	Description
str	str	The ID of the batch job.

invoke(requests, hyperparameters=None, status_check_interval=DEFAULT_STATUS_CHECK_INTERVAL, max_iterations=None) async

Invokes the language model in batch mode.

This method orchestrates the entire batch invocation process, including; 1. Creating a batch job. 2. Iteratively checking the status of the batch job until it is finished. 3. Retrieving the results of the batch job. The method includes retry logic with exponential backoff for transient failures.

Parameters:

Name	Type	Description	Default
requests	dict[str, LMInput]	The dictionary of requests that maps request ID to the request. Each request must be a valid input for the language model. 1. If the request is a list of Message objects, it is used as is. 2. If the request is a list of MessageContent or a string, it is converted into a user message.	required
hyperparameters	dict[str, Any] | None	A dictionary of hyperparameters for the language model. Defaults to None, in which case the default hyperparameters are used.	None
status_check_interval	float	The interval in seconds to check the status of the batch job. Defaults to DEFAULT_STATUS_CHECK_INTERVAL.	DEFAULT_STATUS_CHECK_INTERVAL
max_iterations	int | None	The maximum number of iterations to check the status of the batch job. Defaults to None, in which case the number of iterations is infinite.	None

Returns:

Type	Description
dict[str, LMOutput]	dict[str, LMOutput]: The results of the batch job.

Raises:

Type	Description
CancelledError	If the invocation is cancelled.
ModelNotFoundError	If the model is not found.
ProviderAuthError	If the model authentication fails.
ProviderInternalError	If the model internal error occurs.
ProviderInvalidArgsError	If the model parameters are invalid.
ProviderOverloadedError	If the model is overloaded.
ProviderRateLimitError	If the model rate limit is exceeded.
TimeoutError	If the invocation times out.
ValueError	If the messages are not in the correct format.

list() async

Lists the batch jobs.

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: The list of batch jobs.

retrieve(batch_id, **kwargs) async

Retrieves the results of a batch job.

Parameters:

Name	Type	Description	Default
batch_id	str	The ID of the batch job to get the results of.	required
**kwargs	Any	Additional keyword arguments.	{}

Returns:

Type	Description
dict[str, LMOutput]	dict[str, LMOutput]: The results of the batch job.

status(batch_id) async

Gets the status of a batch job.

Parameters:

Name	Type	Description	Default
batch_id	str	The ID of the batch job to get the status of.	required

Returns:

Name	Type	Description
BatchStatus	BatchStatus	The status of the batch job.