# Batch API Usage and Optimization

**Last updated:** 2026-02
**Status:** GA
**Category:** Performance & Scalability
**Source:** https://learn.microsoft.com/azure/foundry/openai/how-to/batch-blob-storage

---

## Introduksjon

Azure OpenAI Batch API er designet for storskala, asynkron prosessering av AI-arbeidslaster. Med 50% lavere kostnad enn Global Standard-prising og separat kvote som ikke pavirker online-trafikken, er Batch API ideelt for norsk offentlig sektor som trenger a prosessere store volumer av dokumenter, klassifiseringer eller analyser.

For offentlige virksomheter som Direktoratet for digital tjenesteutvikling, Nav, eller Skatteetaten kan Batch API brukes til masseprosessering av henvendelser, dokumentanalyse, oversettelser og datautvinning uten a forstyrre sanntidstjenestene. Tjenesten er spesielt verdifull for periodiske oppgaver som kvartalsvis rapportering, arsavslutning, eller migrering av historiske data.

Denne referansen dekker hele arbeidsflyten for Batch API, fra filsammensetning og opplasting til kostnadsberegning og feilhhandtering, med fokus pa optimalisering for store volumer.

## Oversikt over Batch API

### Nokkelegenskaper

| Egenskap | Verdi |
|----------|-------|
| Kostnadsreduksjon | 50% lavere enn Global Standard |
| Malsatt leveringstid | 24 timer |
| Maksimal filstorrelse | 200 MB (direkte), 1 GB (via Blob Storage) |
| Maksimalt antall requests per fil | 100 000 |
| Maks batch-filer per ressurs | 500 (uten utlop), 10 000 (med utlop) |
| Kvotetype | Separat enqueued token-kvote |
| Stottede modeller | GPT-4o, GPT-4o mini, GPT-4.1, o3-mini m.fl. |
| Deployment-type | GlobalBatch eller DataZoneBatch |

### Batch API vs. Standard API

| Aspekt | Standard API | Batch API |
|--------|-------------|-----------|
| Prosessering | Synkron, umiddelbar | Asynkron, 24-timers mal |
| Kostnad | Full pris | 50% rabatt |
| Kvote | Delt TPM-kvote | Separat enqueued token-kvote |
| Bruksomrade | Sanntid, interaktivt | Masseprosessering, analyse |
| Pavirkning pa online | Ja, deler kapasitet | Nei, separat kapasitet |
| Filformat | JSON per request | JSONL (samlet fil) |

## Batch Job-sammensetning

### JSONL-filformat

Batch API bruker JSON Lines-format (`.jsonl`), der hver linje er en selvstendig JSON-objekt:

```jsonl
{"custom_id": "req-001", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-batch", "messages": [{"role": "system", "content": "Klassifiser henvendelsen som KLAGE, SPORSMAL, eller TILBAKEMELDING."}, {"role": "user", "content": "Jeg er svart misfornoyd med ventetiden pa fornyelse av forerkort."}], "max_tokens": 50}}
{"custom_id": "req-002", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-batch", "messages": [{"role": "system", "content": "Klassifiser henvendelsen som KLAGE, SPORSMAL, eller TILBAKEMELDING."}, {"role": "user", "content": "Hvordan soker jeg om nytt forerkort?"}], "max_tokens": 50}}
{"custom_id": "req-003", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-batch", "messages": [{"role": "system", "content": "Klassifiser henvendelsen som KLAGE, SPORSMAL, eller TILBAKEMELDING."}, {"role": "user", "content": "Fint arbeid med den nye tunnelen i Rogaland!"}], "max_tokens": 50}}
```

**Viktige regler:**
- `custom_id` er obligatorisk og lar deg koble respons til input
- `model` ma vaere identisk pa alle linjer og matche deployment-navnet
- Responser returneres IKKE i samme rekkefolge som input
- For best ytelse: send store filer fremfor mange sma filer

### Responses API-format (nyere)

```jsonl
{"custom_id": "req-001", "method": "POST", "url": "/v1/responses", "body": {"model": "gpt-4o-batch", "input": [{"role": "user", "content": "Oppsummer dette dokumentet: ..."}], "max_output_tokens": 500}}
```

### Programmatisk filgenerering (Python)

```python
import json
from pathlib import Path
from typing import Iterator

def generate_batch_file(
    items: list[dict],
    system_prompt: str,
    model: str,
    output_path: str,
    max_tokens: int = 200
) -> Path:
    """Generer JSONL batch-fil fra en liste med items."""
    path = Path(output_path)

    with open(path, "w", encoding="utf-8") as f:
        for i, item in enumerate(items):
            request = {
                "custom_id": f"req-{i:06d}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": model,
                    "messages": [
                        {"role": "system", "content": system_prompt},
                        {"role": "user", "content": item["content"]}
                    ],
                    "max_tokens": max_tokens,
                    "temperature": 0.1  # Lav temperatur for konsistens
                }
            }
            f.write(json.dumps(request, ensure_ascii=False) + "\n")

    file_size = path.stat().st_size
    print(f"Generert batch-fil: {path}")
    print(f"Antall requests: {len(items)}")
    print(f"Filstorrelse: {file_size / (1024*1024):.1f} MB")

    return path


def chunk_batch_file(
    input_path: str,
    max_requests: int = 100_000,
    max_size_mb: int = 190
) -> list[str]:
    """Del opp en stor batch-fil i mindre filer innenfor grensene."""
    chunks = []
    current_chunk = []
    current_size = 0
    chunk_index = 0

    with open(input_path, "r", encoding="utf-8") as f:
        for line in f:
            line_size = len(line.encode("utf-8"))

            if (len(current_chunk) >= max_requests or
                (current_size + line_size) > max_size_mb * 1024 * 1024):
                # Skriv gjeldende chunk og start ny
                chunk_path = f"{input_path}.chunk_{chunk_index:03d}.jsonl"
                with open(chunk_path, "w", encoding="utf-8") as cf:
                    cf.writelines(current_chunk)
                chunks.append(chunk_path)
                current_chunk = []
                current_size = 0
                chunk_index += 1

            current_chunk.append(line)
            current_size += line_size

    # Skriv siste chunk
    if current_chunk:
        chunk_path = f"{input_path}.chunk_{chunk_index:03d}.jsonl"
        with open(chunk_path, "w", encoding="utf-8") as cf:
            cf.writelines(current_chunk)
        chunks.append(chunk_path)

    return chunks
```

## Filopplasting og -handtering

### Opplasting via Python SDK

```python
from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint="https://your-resource.openai.azure.com/",
    api_key="your-api-key",
    api_version="2025-03-01-preview"
)

# Last opp batch-fil med utlopsdato (14 dager)
file_response = client.files.create(
    file=open("batch_requests.jsonl", "rb"),
    purpose="batch",
    extra_body={
        "expires_after": {
            "seconds": 1209600,  # 14 dager
            "anchor": "created_at"
        }
    }
)

print(f"Fil-ID: {file_response.id}")
print(f"Status: {file_response.status}")
print(f"Storrelse: {file_response.bytes} bytes")
```

### Opplasting via REST API

```bash
curl https://YOUR_RESOURCE.openai.azure.com/openai/files?api-version=2025-03-01-preview \
  -H "api-key: $AZURE_OPENAI_API_KEY" \
  -F "purpose=batch" \
  -F "file=@batch_requests.jsonl" \
  -F "expires_after.seconds=1209600" \
  -F "expires_after.anchor=created_at"
```

### Stor fil via Azure Blob Storage (BYOS)

For filer over 200 MB (opptil 1 GB):

```python
from azure.storage.blob import BlobServiceClient

# 1. Last opp til Azure Blob Storage
blob_service = BlobServiceClient.from_connection_string(conn_str)
container = blob_service.get_container_client("batch-files")

with open("large_batch.jsonl", "rb") as data:
    container.upload_blob(
        name="large_batch.jsonl",
        data=data,
        overwrite=True
    )

# 2. Konfigurer Azure OpenAI til a bruke Blob Storage
# Se: https://learn.microsoft.com/azure/foundry-classic/openai/how-to/batch-blob-storage
```

### Filgrenser

| Grense | Verdi | Med utlopsdato |
|--------|-------|----------------|
| Maks input-filstorrelse | 200 MB | 200 MB |
| Maks input-filstorrelse (BYOS) | 1 GB | 1 GB |
| Maks requests per fil | 100 000 | 100 000 |
| Maks input-filer per ressurs | 500 | 10 000 |
| Utlopstid | Ingen utlop | 14-30 dager |

## Batch Job-oppretting og -overvaking

### Opprett batch job

```python
# Opprett batch job
batch_response = client.batches.create(
    input_file_id=file_response.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
    extra_body={
        "output_expires_after": {
            "seconds": 1209600,
            "anchor": "created_at"
        }
    }
)

print(f"Batch-ID: {batch_response.id}")
print(f"Status: {batch_response.status}")
```

### Overvak batch-status

```python
import time

def monitor_batch(client, batch_id: str, poll_interval: int = 60):
    """Overvak batch job til den er ferdig."""
    while True:
        batch = client.batches.retrieve(batch_id)

        print(f"Status: {batch.status}")
        print(f"  Requests total: {batch.request_counts.total}")
        print(f"  Completed: {batch.request_counts.completed}")
        print(f"  Failed: {batch.request_counts.failed}")

        if batch.status in ("completed", "failed", "cancelled", "expired"):
            return batch

        time.sleep(poll_interval)

# Overvak med 60 sekunders intervall
final_batch = monitor_batch(client, batch_response.id)
```

### Batch-statusflyten

```
validating -> in_progress -> completed
                          -> failed
                          -> cancelled (manuelt)
                          -> expired (sjelden)
```

### Hent resultater

```python
if final_batch.status == "completed":
    # Hent resultatfil
    output_file_id = final_batch.output_file_id
    result_content = client.files.content(output_file_id)

    # Parse resultater
    results = {}
    for line in result_content.text.strip().split("\n"):
        result = json.loads(line)
        custom_id = result["custom_id"]
        response_body = result["response"]["body"]

        if result["response"]["status_code"] == 200:
            content = response_body["choices"][0]["message"]["content"]
            results[custom_id] = {
                "status": "success",
                "content": content,
                "tokens": response_body["usage"]
            }
        else:
            results[custom_id] = {
                "status": "error",
                "error": result.get("error", {})
            }

    # Sjekk feilfil
    if final_batch.error_file_id:
        error_content = client.files.content(final_batch.error_file_id)
        errors = [json.loads(line) for line in error_content.text.strip().split("\n")]
        print(f"Antall feil: {len(errors)}")
```

## Kostnadsberegning og besparelser

### Prissammenligning

| Modell | Standard input (per 1M tokens) | Batch input (per 1M tokens) | Besparelse |
|--------|-------------------------------|---------------------------|-----------|
| GPT-4o | $2.50 | $1.25 | 50% |
| GPT-4o mini | $0.15 | $0.075 | 50% |
| GPT-4.1 | $2.00 | $1.00 | 50% |

*Priser er illustrative og kan variere. Sjekk azure.microsoft.com/pricing for oppdaterte priser.*

### Kostnadsestimering

```python
def estimate_batch_cost(
    num_requests: int,
    avg_input_tokens: int,
    avg_output_tokens: int,
    model: str = "gpt-4o"
) -> dict:
    """Estimer kostnad for batch vs. standard prosessering."""

    # Priser per 1M tokens (NOK, ca. kurs)
    prices = {
        "gpt-4o": {
            "standard_input": 27.50,    # NOK per 1M tokens
            "standard_output": 110.00,
            "batch_input": 13.75,       # 50% rabatt
            "batch_output": 55.00       # 50% rabatt
        },
        "gpt-4o-mini": {
            "standard_input": 1.65,
            "standard_output": 6.60,
            "batch_input": 0.825,
            "batch_output": 3.30
        }
    }

    p = prices.get(model, prices["gpt-4o"])

    total_input_tokens = num_requests * avg_input_tokens
    total_output_tokens = num_requests * avg_output_tokens

    standard_cost = (
        (total_input_tokens / 1_000_000) * p["standard_input"] +
        (total_output_tokens / 1_000_000) * p["standard_output"]
    )

    batch_cost = (
        (total_input_tokens / 1_000_000) * p["batch_input"] +
        (total_output_tokens / 1_000_000) * p["batch_output"]
    )

    return {
        "num_requests": num_requests,
        "total_tokens": total_input_tokens + total_output_tokens,
        "standard_cost_nok": round(standard_cost, 2),
        "batch_cost_nok": round(batch_cost, 2),
        "savings_nok": round(standard_cost - batch_cost, 2),
        "savings_percent": 50
    }

# Eksempel: 50 000 henvendelser, 500 input tokens, 100 output tokens
estimate = estimate_batch_cost(50_000, 500, 100, "gpt-4o")
print(f"Standard: {estimate['standard_cost_nok']} NOK")
print(f"Batch:    {estimate['batch_cost_nok']} NOK")
print(f"Besparelse: {estimate['savings_nok']} NOK ({estimate['savings_percent']}%)")
```

### Kostnadsoptimalisering for batch

| Strategi | Beskrivelse | Effekt |
|----------|-------------|--------|
| Bruk GPT-4o mini | For enklere oppgaver (klassifisering, utvinning) | 90%+ billigere enn GPT-4o |
| Lav max_tokens | Tilpass til forventet output | Unngaer overfakturering |
| Lav temperatur | Mer konsistent, potensielt kortere output | 5-15% |
| Strukturert output | JSON schema for forutsigbar lengde | 10-20% |
| Store batch-filer | Samle mange requests i en fil | Bedre throughput |

## Retry og feilhhandtering

### Koe-handtering ved kvotegrense

Nar batch jobs er for store for tilgjengelig kvote, bruk fail-fast med eksponentiell backoff:

```python
import time
from openai import AzureOpenAI, BadRequestError

def submit_batch_with_retry(
    client: AzureOpenAI,
    file_id: str,
    max_retries: int = 10,
    initial_wait: int = 300  # 5 minutter
):
    """Submit batch job med automatisk retry ved kvotegrense."""
    for attempt in range(max_retries):
        try:
            batch = client.batches.create(
                input_file_id=file_id,
                endpoint="/v1/chat/completions",
                completion_window="24h"
            )
            print(f"Batch opprettet: {batch.id}")
            return batch

        except BadRequestError as e:
            if "enqueued token limit" in str(e).lower():
                wait_time = initial_wait * (2 ** attempt)
                print(f"Kvotegrense nadd. Venter {wait_time}s for forsok {attempt + 1}/{max_retries}")
                time.sleep(wait_time)
            else:
                raise

    raise Exception(f"Kunne ikke opprette batch etter {max_retries} forsok")
```

### Handtering av delvise feil

```python
def process_batch_results(client, batch_id: str) -> dict:
    """Prosesser batch-resultater og segreger suksess/feil."""
    batch = client.batches.retrieve(batch_id)

    results = {"success": [], "errors": [], "stats": {}}

    # Hent suksessfulle resultater
    if batch.output_file_id:
        output = client.files.content(batch.output_file_id)
        for line in output.text.strip().split("\n"):
            result = json.loads(line)
            if result["response"]["status_code"] == 200:
                results["success"].append(result)
            else:
                results["errors"].append(result)

    # Hent dedikerte feil
    if batch.error_file_id:
        errors = client.files.content(batch.error_file_id)
        for line in errors.text.strip().split("\n"):
            results["errors"].append(json.loads(line))

    # Statistikk
    results["stats"] = {
        "total": batch.request_counts.total,
        "completed": batch.request_counts.completed,
        "failed": batch.request_counts.failed,
        "success_rate": (
            batch.request_counts.completed / batch.request_counts.total * 100
            if batch.request_counts.total > 0 else 0
        )
    }

    return results


def retry_failed_requests(
    client: AzureOpenAI,
    failed_results: list,
    original_file_path: str,
    model: str
) -> str:
    """Generer ny batch-fil fra feilede requests for retry."""
    # Les originale requests for a finne matchende custom_ids
    original_requests = {}
    with open(original_file_path, "r") as f:
        for line in f:
            req = json.loads(line)
            original_requests[req["custom_id"]] = req

    # Generer retry-fil
    retry_path = original_file_path.replace(".jsonl", "_retry.jsonl")
    failed_ids = {r["custom_id"] for r in failed_results}

    with open(retry_path, "w") as f:
        for custom_id in failed_ids:
            if custom_id in original_requests:
                f.write(json.dumps(original_requests[custom_id]) + "\n")

    print(f"Retry-fil generert: {retry_path} ({len(failed_ids)} requests)")
    return retry_path
```

## Bruksomrader for norsk offentlig sektor

### Masseklassifisering av innbyggerhenvendelser

```python
# Eksempel: Klassifiser 100 000 henvendelser fra innbyggerportal
system_prompt = """Klassifiser henvendelsen. Svar med JSON:
{"kategori": "KLAGE|SPORSMAL|TILBAKEMELDING|SOKNAD",
 "prioritet": "HOY|NORMAL|LAV",
 "etat": "FORERKORT|KJORETOYREG|VEIPROSJEKT|ANNET"}"""

batch_file = generate_batch_file(
    items=henvendelser,
    system_prompt=system_prompt,
    model="gpt-4o-mini-batch",
    output_path="henvendelser_batch.jsonl",
    max_tokens=100
)
```

### Dokumentanalyse og oppsummering

```python
# Eksempel: Oppsummer 10 000 hoeringsuttalelser
system_prompt = """Oppsummer hoeringsuttalelsen i 2-3 setninger.
Identifiser hovedstandpunkt og eventuelle konkrete forslag."""

batch_file = generate_batch_file(
    items=hoeringsuttalelser,
    system_prompt=system_prompt,
    model="gpt-4o-batch",
    output_path="hoering_batch.jsonl",
    max_tokens=300
)
```

### Sprakvasking og oversettelse

```python
# Eksempel: Oversett 50 000 dokumentfragmenter til nynorsk
system_prompt = "Oversett teksten fra bokmaal til nynorsk. Bevar fagterminologi."
```

## Sjekkliste for batch-optimalisering

| Nr | Tiltak | Prioritet |
|----|--------|-----------|
| 1 | Bruk store filer (ikke mange sma) | Hoy |
| 2 | Sett utlop pa filer (expires_after) for a unnga 500-filgrensen | Hoy |
| 3 | Velg GPT-4o mini for enklere oppgaver | Hoy |
| 4 | Implementer retry med eksponentiell backoff | Hoy |
| 5 | Tilpass max_tokens til faktisk behov | Medium |
| 6 | Bruk strukturert output (JSON schema) | Medium |
| 7 | Overvak med polling (60s intervall) | Medium |
| 8 | Implementer retry for feilede requests | Medium |
| 9 | Bruk Blob Storage for filer over 200 MB | Ved behov |
| 10 | Sett opp alerting for batch completion | Anbefalt |

## For Cosmo

- **50% kostnadsreduksjon** gjor Batch API til forstevalgdet for all ikke-sanntids AI-prosessering i offentlig sektor. Masseklassifisering, dokumentanalyse og oversettelse bor alltid bruke batch.
- **Datasuverenitet:** Batch API prosesserer data i enhver Azure OpenAI-region for Global Batch. Bruk DataZoneBatch for a begrense til EU-regioner, eller Regional Batch for strengeste krav.
- **Completion window pa 24 timer** er et mal, ikke en garanti. Jobs som tar lengre tid utloper ikke, men du kan kansellere og fa resultater for fullfort arbeid.
- **Enqueued token-kvote** er separat fra online-kvote, sa batch-prosessering pavirker ikke sanntidstjenester. Ideelt for nattlige batch-kjoeringer.
- **Retry-pattern er kritisk:** Store batch jobs kan feile pa kvotegrenser. Implementer alltid fail-fast med eksponentiell backoff og retry av feilede requests.
