LLM Batch Service

The LLM Batch Service lets you process large volumes of LLM requests asynchronously. Instead of making individual synchronous inference calls, you submit a collection of requests as a single JSONL input file. The service processes them in the background and writes the results to your object store.

Note: Batch consumption only supports native LLM calls. Orchestration requests are not supported.

Batch consumption is available in EU and US regions, except prod-euonly and sovereign cloud deployments.

Key capabilities:

• Process hundreds or thousands of LLM requests in a single submission

• Reduced cost compared to synchronous inference calls

• Automatic retry handling for transient provider errors

• No rate-limit management required on the client side

Prerequisites

Before using the SDK, complete the setup steps described in the SAP Help Portal — Batch Consumption:

1. Ensure you have a valid SAP AI Core service instance with access to the generative AI hub.

2. Register an object store secret for one of the supported providers (Amazon S3, Azure Blob Storage, Google Cloud Storage, Alibaba Cloud OSS, or SAP HANA Cloud, Data Lake).

3. Prepare your input file as a JSON Lines (.jsonl) file and upload it to your object store, following the input file specification and upload instructions on the Help Portal.

Once the input file is uploaded, note down its ai:// URI and the ai:// URI of the output directory — you will need both below.

The BatchService initialises a GenAIHubProxyClient automatically, which reads credentials from configuration files or environment variables. You can also pass a custom proxy_client instance.

from gen_ai_hub.batch_service import BatchService

# Initialise using credentials from the environment / config file
batch_service = BatchService()

# Or provide a proxy client explicitly:
# from gen_ai_hub.proxy import get_proxy_client
# proxy_client = get_proxy_client('gen-ai-hub')
# batch_service = BatchService(proxy_client=proxy_client)

# Configure these values before running the workflow cells
INPUT_URI = "ai://<object_store_secret_name>/<path>/input.jsonl"
OUTPUT_URI = "ai://<object_store_secret_name>/<output_folder>/"
MODEL = "gpt-4.1"   # Must match the model used in the input file
PROVIDER = "azure-openai"

Create a Batch Job

Submit the batch job by calling batch_service.create(). Provide:

Parameter	Description
type	Batch processing type. Only "llm-native" is supported.
input_uri	ai:// URI of the input .jsonl file.
output_uri	ai:// URI of the output directory. Must end with /.
provider	LLM provider (e.g. "azure-openai").
model	Model name — must match the value used in the input file.

The service schedules the job and returns a unique batch ID.

create_response = batch_service.create(
    type="llm-native",
    input_uri=INPUT_URI,
    output_uri=OUTPUT_URI,
    provider=PROVIDER,
    model=MODEL,
)

BATCH_ID = create_response.id
print(f"Batch ID: {BATCH_ID}")
print(f"Status:   {create_response.status}")
print(f"Message:  {create_response.message}")

Check Batch Status

Batches are processed asynchronously. Poll the status endpoint until the job reaches a terminal state (COMPLETED, FAILED, or CANCELLED).

import time

TERMINAL_STATUSES = {"COMPLETED", "FAILED", "CANCELLED"}
POLLING_INTERVAL = 30  # seconds

while True:
    status_response = batch_service.get_status(BATCH_ID)
    current = status_response.current_status
    print(f"[{time.strftime('%H:%M:%S')}] Status: {current}  →  target: {status_response.target_status}")

    if current in TERMINAL_STATUSES:
        break

    time.sleep(POLLING_INTERVAL)

print(f"\nFinal status: {current}")
if status_response.message:
    print(f"Message: {status_response.message}")

List Batch Jobs and Get Batch Details

List all batch jobs

Retrieve a summary of all batch jobs for the current resource group.

list_response = batch_service.list()

print(f"Total batch jobs: {list_response.count}")
for job in list_response.resources or []:
    print(f"  {job.id}  |  {job.provider:<15}  |  {job.created_at}  |  {job.status}")

Get full details of a batch job

Retrieve the complete details of a specific batch job: input/output URIs, provider, model, and current status.

detail = batch_service.get(BATCH_ID)

print(f"ID:           {detail.id}")
print(f"Type:         {detail.type}")
print(f"Provider:     {detail.provider}")
print(f"Created at:   {detail.created_at}")
print(f"Input URI:    {detail.input.uri}")
print(f"Output URI:   {detail.output.uri}")
print(f"Model:        {detail.spec.get('model') if detail.spec else None}")
print(f"Status:       {detail.status.current_status}")
print(f"Updated at:   {detail.status.updated_at}")
if detail.status.message:
    print(f"Message:      {detail.status.message}")

Cancel a Batch Job

You can cancel a batch job that is in a non-terminal state (e.g. PENDING or IN_PROGRESS). Cancellation is asynchronous — the job transitions to CANCELLING and then CANCELLED after any in-flight provider requests have been terminated.

Use get_status() to track the cancellation progress.

# Cancel the batch job — only meaningful if the job is still in a non-terminal state.
# Skip this cell if the job has already completed.

cancel_response = batch_service.cancel(BATCH_ID)
print(f"ID:      {cancel_response.id}")
print(f"Message: {cancel_response.message}")

# Poll until the cancellation is confirmed
while True:
    status_response = batch_service.get_status(BATCH_ID)
    print(f"Status: {status_response.current_status}")
    if status_response.current_status in TERMINAL_STATUSES:
        break
    time.sleep(10)

Delete a Batch Job

Deleting a job removes its metadata from the service. Output files in the object store are not affected.

Restriction: Only batch jobs in COMPLETED, FAILED, or CANCELLED state can be deleted.

# Confirm the job is in a deletable state before deleting
status_response = batch_service.get_status(BATCH_ID)
deletable = {"COMPLETED", "FAILED", "CANCELLED"}

if status_response.current_status not in deletable:
    print(f"Cannot delete: job is in state '{status_response.current_status}'. Cancel it first.")
else:
    delete_response = batch_service.delete(BATCH_ID)
    print(f"Deleted batch job: {delete_response.id}")
    print(f"Message: {delete_response.message}")

Retrieving Results

Once the job status is COMPLETED, the results are available in your object store under the output URI specified at creation, in a subdirectory named after the batch ID. Successful responses are written to output.jsonl and any failed individual requests to error.jsonl, each line matched to its input by custom_id.

Async Support

All BatchService methods have async counterparts: acreate, alist, aget, aget_status, acancel, adelete. The example below runs the full workflow asynchronously.

import asyncio

async def run_batch_async():
    service = BatchService()

    # Create
    resp = await service.acreate(
        type="llm-native",
        input_uri=INPUT_URI,
        output_uri=OUTPUT_URI,
        provider=PROVIDER,
        model=MODEL,
    )
    batch_id = resp.id
    print(f"Created batch: {batch_id}  ({resp.status})")

    # Poll until terminal
    while True:
        status = await service.aget_status(batch_id)
        print(f"Status: {status.current_status}")
        if status.current_status in TERMINAL_STATUSES:
            break
        await asyncio.sleep(30)

    # Get full details
    detail = await service.aget(batch_id)
    print(f"Completed. Output at: {detail.output.uri}{batch_id}/output.jsonl")

    await service.aclose_http_connection()

await run_batch_async()