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.

Prerequisites

Complete the setup steps in the SAP Help Portal — Batch Consumption:

1. Register an object store secret (Amazon S3, Azure Blob Storage, GCS, Alibaba OSS, or SAP HANA Cloud Data Lake).
2. Prepare your input .jsonl file and upload it to your object store.
3. Note the ai:// URI of the input file and the output directory.

Initialize

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

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

Check Batch Status

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}")
    if current in TERMINAL_STATUSES:
        break
    time.sleep(POLLING_INTERVAL)

print(f"Final status: {current}")

List and Get Batch Jobs

# 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"Input URI:  {detail.input.uri}")
print(f"Output URI: {detail.output.uri}")
print(f"Status:     {detail.status.current_status}")

Cancel a Batch Job

cancel_response = batch_service.cancel(BATCH_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

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

Async Support

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()