Skip to content

Batch

The Batch API lets you submit multiple requests as a single job for asynchronous processing, typically at lower cost than real-time requests.

  1. Prepare a JSONL file where each line is a batch request object.
  2. Call create_batch() with the file path and target endpoint.
  3. any-llm uploads the file to the provider and creates the batch job.
  4. Poll with retrieve_batch() to check status.
  5. When complete, download results from the provider.

The input file must be a JSONL file where each line follows this structure:

{"custom_id": "request-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4.1-mini", "messages": [{"role": "user", "content": "Hello"}]}}
{"custom_id": "request-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4.1-mini", "messages": [{"role": "user", "content": "World"}]}}

Create a batch job by uploading a local JSONL file.

def create_batch(
provider: str | LLMProvider,
input_file_path: str,
endpoint: str,
*,
completion_window: str = "24h",
metadata: dict[str, str] | None = None,
api_key: str | None = None,
api_base: str | None = None,
client_args: dict[str, Any] | None = None,
**kwargs: Any,
) -> Batch

Async variant with the same parameters.

ParameterTypeDefaultDescription
providerstr | LLMProviderrequiredProvider name to use for the request (e.g., ‘openai’, ‘mistral’)
input_file_pathstrrequiredPath to a local file containing batch requests in JSONL format.
endpointstrrequiredThe endpoint to be used for all requests (e.g., ‘/v1/chat/completions’)
completion_windowstr”24h”The time frame within which the batch should be processed (default: ‘24h’)
metadatadict[str, str] | NoneNoneOptional custom metadata for the batch
api_keystr | NoneNoneAPI key for the provider
api_basestr | NoneNoneBase URL for the provider API
client_argsdict[str, Any] | NoneNoneAdditional provider-specific arguments for client instantiation
**kwargsAnyrequiredAdditional provider-specific arguments

Returns: A Batch object.

Retrieve the current status and details of a batch job.

def retrieve_batch(
provider: str | LLMProvider,
batch_id: str,
*,
api_key: str | None = None,
api_base: str | None = None,
client_args: dict[str, Any] | None = None,
**kwargs: Any,
) -> Batch

Async variant with the same parameters.

ParameterTypeDefaultDescription
providerstr | LLMProviderrequiredProvider name to use for the request (e.g., ‘openai’, ‘mistral’)
batch_idstrrequiredThe ID of the batch to retrieve
api_keystr | NoneNoneAPI key for the provider
api_basestr | NoneNoneBase URL for the provider API
client_argsdict[str, Any] | NoneNoneAdditional provider-specific arguments for client instantiation
**kwargsAnyrequiredAdditional provider-specific arguments

Returns: A Batch object.

Cancel an in-progress batch job.

def cancel_batch(
provider: str | LLMProvider,
batch_id: str,
*,
api_key: str | None = None,
api_base: str | None = None,
client_args: dict[str, Any] | None = None,
**kwargs: Any,
) -> Batch

Async variant with the same parameters.

Returns: The cancelled Batch object.

List batch jobs for a provider.

def list_batches(
provider: str | LLMProvider,
*,
after: str | None = None,
limit: int | None = None,
api_key: str | None = None,
api_base: str | None = None,
client_args: dict[str, Any] | None = None,
**kwargs: Any,
) -> Sequence[Batch]

Async variant with the same parameters.

ParameterTypeDefaultDescription
providerstr | LLMProviderrequiredProvider name to use for the request (e.g., ‘openai’, ‘mistral’)
afterstr | NoneNoneA cursor for pagination. Returns batches after this batch ID.
limitint | NoneNoneMaximum number of batches to return (default: 20)
api_keystr | NoneNoneAPI key for the provider
api_basestr | NoneNoneBase URL for the provider API
client_argsdict[str, Any] | NoneNoneAdditional provider-specific arguments for client instantiation
**kwargsAnyrequiredAdditional provider-specific arguments

Returns: A Sequence of Batch objects.

from any_llm import create_batch, retrieve_batch, list_batches
# Create a batch job
batch = create_batch(
provider="openai",
input_file_path="requests.jsonl",
endpoint="/v1/chat/completions",
)
print(f"Batch created: {batch.id}, status: {batch.status}")
# Check status
batch = retrieve_batch("openai", batch.id)
print(f"Status: {batch.status}")
# List all batches
batches = list_batches("openai")
for b in batches:
print(f"{b.id}: {b.status}")