LLM Batch Service

The LLM Batch Service lets you process large volumes of LLM requests asynchronously. Instead of individual synchronous calls, you submit a JSONL input file and the service processes it in the background.

note

Batch consumption only supports native LLM calls. Orchestration requests are not supported. 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

Complete the setup steps 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 (Amazon S3, Azure Blob Storage, GCS, Alibaba OSS, or SAP HANA Cloud Data Lake).
3. Prepare your input .jsonl file and upload it to your object store.

Once the input file is uploaded, note down its ai:// URI of the input file and the output directory.

Initialize

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

batch_service = BatchService()

# Or with an explicit proxy client:
# from gen_ai_hub.proxy import get_proxy_client
# batch_service = BatchService(proxy_client=get_proxy_client('gen-ai-hub'))

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.

INPUT_URI = "ai://<object_store_secret_name>/<path>/input.jsonl"
OUTPUT_URI = "ai://<object_store_secret_name>/<output_folder>/"
MODEL = "gpt-4.1"
PROVIDER = "azure-openai"

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

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"Final status: {current}")
if status_response.message:
    print(f"Message: {status_response.message}")

List and Get Batch Jobs

Retrieve a summary of all batch jobs for the current resource group, or the complete details of a specific batch job (input/output URIs, provider, model, and current status).

# List all batch jobs
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}  |  {job.status}")

# Get full details
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_response = batch_service.cancel(BATCH_ID)
print(f"ID:      {cancel_response.id}")
print(f"Message: {cancel_response.message}")

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. Only jobs in COMPLETED, FAILED, or CANCELLED state can be deleted.

status_response = batch_service.get_status(BATCH_ID)
if status_response.current_status not in {"COMPLETED", "FAILED", "CANCELLED"}:
    print(f"Cannot delete: job is in state '{status_response.current_status}'.")
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
from gen_ai_hub.batch_service import BatchService

async def run_batch_async():
    service = BatchService()

    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})")

    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)

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

    await service.aclose_http_connection()

await run_batch_async()