Getting Started with Batches

Overview

Batches let you run the same product job over every file in a directory. A batch references:

1. A source directory containing the files to process.
2. A product configuration ID for the work to run on each file.

Use parse_v2 with a built-in Parse preset or saved Parse configuration to parse every file. Use extract_v2 with a saved Extract configuration to extract from every file.

Batch creation is limited to 10,000 source files.

Prerequisites

• A LlamaCloud account with a Pro or Enterprise plan
• An API key (how to create one)
• A configuration ID for the job to run:
  • For parse_v2, use a built-in preset such as cfg-PARSE_AGENTIC, or a saved Parse configuration ID.
  • For extract_v2, use a saved Extract configuration ID.

Tip

The easiest way to create configurations is from the Parse or Extract playground in the LlamaCloud UI. You can also create them through the Product Configurations API.

Create a Batch

For one-off uploads, create an ephemeral directory so the source directory is automatically eligible for cleanup. Then upload files into that directory and create the batch. Files in ephemeral directories are also exempt from storage billing and the per-project storage limits.

Python

import asyncio
from pathlib import Path

from llama_cloud import AsyncLlamaCloud

configuration_id = "cfg-PARSE_AGENTIC"


async def create_batch() -> tuple[str, str]:
    # Reads LLAMA_CLOUD_API_KEY, or pass api_key="<your-api-key>" explicitly.
    async with AsyncLlamaCloud() as client:
        directory = await client.beta.directories.create(
            name="invoice-batch",
            type="ephemeral",
        )

        for path in Path("./invoices").glob("*.pdf"):
            await client.beta.directories.files.upload(
                directory.id,
                upload_file=path,
                display_name=path.name,
            )

        batch = await client.batches.create(
            source_directory_id=directory.id,
            config={
                "job": {
                    "type": "parse_v2",
                    "configuration_id": configuration_id,
                },
            },
        )

    print(batch.id, batch.status)
    return directory.id, batch.id


directory_id, batch_id = asyncio.run(create_batch())

The snippets below build on this one — paste them into one file in order.

TypeScript

import fs from "fs";
import path from "path";
import LlamaCloud from "@llamaindex/llama-cloud";

const client = new LlamaCloud({
  apiKey: "<your-api-key>",
});

const configurationId = "cfg-PARSE_AGENTIC";

const directory = await client.beta.directories.create({
  name: "invoice-batch",
  type: "ephemeral",
});

for (const fileName of fs.readdirSync("./invoices")) {
  if (!fileName.endsWith(".pdf")) continue;

  await client.beta.directories.files.upload(directory.id, {
    upload_file: fs.createReadStream(path.join("./invoices", fileName)),
    display_name: fileName,
  });
}

let batch = await client.batches.create({
  source_directory_id: directory.id,
  config: {
    job: {
      type: "parse_v2",
      configuration_id: configurationId,
    },
  },
});

console.log(batch.id, batch.status);

Go

package main

import (
  "context"
  "fmt"
  "log"
  "os"
  "path/filepath"

  llamacloud "github.com/run-llama/llama-parse-go"
)

func main() {
  ctx := context.Background()
  client := llamacloud.NewClient()

  configurationID := "cfg-PARSE_AGENTIC"

  directory, err := client.Beta.Directories.New(ctx, llamacloud.BetaDirectoryNewParams{
    Name: "invoice-batch",
    Type: llamacloud.BetaDirectoryNewParamsTypeEphemeral,
  })
  if err != nil {
    log.Fatal(err)
  }

  paths, err := filepath.Glob("./invoices/*.pdf")
  if err != nil {
    log.Fatal(err)
  }

  for _, path := range paths {
    file, err := os.Open(path)
    if err != nil {
      log.Fatal(err)
    }

    _, err = client.Beta.Directories.Files.Upload(ctx, directory.ID, llamacloud.BetaDirectoryFileUploadParams{
      UploadFile:  file,
      DisplayName: llamacloud.String(filepath.Base(path)),
    })
    file.Close()
    if err != nil {
      log.Fatal(err)
    }
  }

  batch, err := client.Batches.New(ctx, llamacloud.BatchNewParams{
    SourceDirectoryID: directory.ID,
    Config: llamacloud.BatchNewParamsConfig{
      Job: llamacloud.BatchNewParamsConfigJob{
        Type:            llamacloud.BatchNewParamsConfigJobTypeParseV2,
        ConfigurationID: configurationID,
      },
    },
  })
  if err != nil {
    log.Fatal(err)
  }

  fmt.Println(batch.ID, batch.Status)
}

Java

import ai.llamaindex.llamacloud.client.LlamaCloudClient;
import ai.llamaindex.llamacloud.client.okhttp.LlamaCloudOkHttpClient;
import ai.llamaindex.llamacloud.models.batches.BatchCreateParams;
import ai.llamaindex.llamacloud.models.batches.BatchCreateResponse;
import ai.llamaindex.llamacloud.models.beta.directories.DirectoryCreateParams;
import ai.llamaindex.llamacloud.models.beta.directories.DirectoryCreateResponse;
import ai.llamaindex.llamacloud.models.beta.directories.files.FileUploadParams;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;

LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();

String configurationId = "cfg-PARSE_AGENTIC";

DirectoryCreateResponse directory =
    client.beta()
        .directories()
        .create(
            DirectoryCreateParams.builder()
                .name("invoice-batch")
                .type(DirectoryCreateParams.Type.EPHEMERAL)
                .build());

try (DirectoryStream<Path> paths =
    Files.newDirectoryStream(Paths.get("./invoices"), "*.pdf")) {
    for (Path path : paths) {
        client.beta()
            .directories()
            .files()
            .upload(
                directory.id(),
                FileUploadParams.builder()
                    .uploadFile(path)
                    .displayName(path.getFileName().toString())
                    .build());
    }
}

BatchCreateResponse batch =
    client.batches()
        .create(
            BatchCreateParams.builder()
                .sourceDirectoryId(directory.id())
                .config(
                    BatchCreateParams.Config.builder()
                        .job(
                            BatchCreateParams.Config.Job.builder()
                                .type(BatchCreateParams.Config.Job.Type.PARSE_V2)
                                .configurationId(configurationId)
                                .build())
                        .build())
                .build());

System.out.println(batch.id() + " " + batch.status());

CLI

export LLAMA_CLOUD_API_KEY="<your-api-key>"

CONFIGURATION_ID="cfg-PARSE_AGENTIC"

DIRECTORY_ID=$(llp beta:directories create \
  --name invoice-batch \
  --type ephemeral | jq -r '.id')

for file in ./invoices/*.pdf; do
  llp beta:directories:files upload \
    --directory-id "$DIRECTORY_ID" \
    --upload-file "$file" \
    --display-name "$(basename "$file")"
done

BATCH_ID=$(llp batches create \
  --source-directory-id "$DIRECTORY_ID" \
  --config "{job: {type: parse_v2, configuration_id: $CONFIGURATION_ID}}" | jq -r '.id')

echo "$BATCH_ID"

To run extraction instead, use a saved extract_v2 configuration ID and set type to "extract_v2".

Poll for Completion

Batch processing is asynchronous. Poll the batch status until it reaches a terminal state.

Python

terminal_statuses = {"COMPLETED", "FAILED", "CANCELLED"}


async def wait_for_batch(batch_id: str):
    async with AsyncLlamaCloud() as client:
        batch = await client.batches.get(batch_id)
        while batch.status not in terminal_statuses:
            await asyncio.sleep(10)
            batch = await client.batches.get(batch_id)
        return batch


batch = asyncio.run(wait_for_batch(batch_id))

print(batch.status)

TypeScript

const terminalStatuses = new Set(["COMPLETED", "FAILED", "CANCELLED"]);

while (!terminalStatuses.has(batch.status)) {
  await new Promise((resolve) => setTimeout(resolve, 10_000));
  batch = await client.batches.get(batch.id);
}

console.log(batch.status);

Go

detail, err := client.Batches.Get(ctx, batch.ID, llamacloud.BatchGetParams{})
if err != nil {
  log.Fatal(err)
}

for detail.Status != llamacloud.BatchGetResponseStatusCompleted &&
  detail.Status != llamacloud.BatchGetResponseStatusFailed &&
  detail.Status != llamacloud.BatchGetResponseStatusCancelled {
  time.Sleep(2 * time.Second)
  detail, err = client.Batches.Get(ctx, batch.ID, llamacloud.BatchGetParams{})
  if err != nil {
    log.Fatal(err)
  }
}

fmt.Println(detail.Status)

Java

BatchGetResponse detail = client.batches().get(batch.id());

while (!detail.status().equals(BatchGetResponse.Status.COMPLETED)
    && !detail.status().equals(BatchGetResponse.Status.FAILED)
    && !detail.status().equals(BatchGetResponse.Status.CANCELLED)) {
    Thread.sleep(2000);
    detail = client.batches().get(batch.id());
}

System.out.println(detail.status());

CLI

while true; do
  STATUS=$(llp batches get --batch-id "$BATCH_ID" | jq -r '.status')
  case "$STATUS" in
    COMPLETED|FAILED|CANCELLED) break ;;
  esac
  sleep 2
done

echo "$STATUS"

Inspect Per-File Results

Use expand=results on the get endpoint to include the source-file to product-job mappings. Results may be null while the batch is still running. When available, results contains one entry per source file in the batch. Per-file failures are returned in results[*].error_message; successful files include a job_reference for the underlying Parse or Extract job.

A trimmed response can include successful and failed file-level entries in the same batch:

{
  "id": "bat-aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
  "status": "COMPLETED",
  "results": [
    {
      "source_directory_file_id": "dfl-aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
      "job_reference": {
        "type": "parse_v2",
        "id": "pjb-aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
      }
    },
    {
      "source_directory_file_id": "dfl-bbbbbbbb-cccc-dddd-eeee-ffffffffffff",
      "error_message": "Unable to process source file."
    }
  ]
}

Python

async def get_batch_with_results(batch_id: str):
    async with AsyncLlamaCloud() as client:
        return await client.batches.get(batch_id, expand=["results"])


batch = asyncio.run(get_batch_with_results(batch_id))

if batch.status == "FAILED":
    raise RuntimeError("Batch orchestration failed")

for result in batch.results or []:
    if result.error_message:
        print(result.source_directory_file_id, "failed:", result.error_message)
        continue

    if result.job_reference is None:
        print(result.source_directory_file_id, "has no job yet")
        continue

    print(
        result.source_directory_file_id,
        result.job_reference.type,
        result.job_reference.id,
    )

TypeScript

const detail = await client.batches.get(batch.id, {
  expand: ["results"],
});

if (detail.status === "FAILED") {
  throw new Error("Batch orchestration failed");
}

for (const result of detail.results ?? []) {
  if (result.error_message) {
    console.log(result.source_directory_file_id, "failed:", result.error_message);
    continue;
  }

  if (!result.job_reference) {
    console.log(result.source_directory_file_id, "has no job yet");
    continue;
  }

  console.log(
    result.source_directory_file_id,
    result.job_reference.type,
    result.job_reference.id,
  );
}

Go

detail, err := client.Batches.Get(ctx, batch.ID, llamacloud.BatchGetParams{
  Expand: []string{"results"},
})
if err != nil {
  log.Fatal(err)
}

if detail.Status == llamacloud.BatchGetResponseStatusFailed {
  log.Fatal("Batch orchestration failed")
}

for _, result := range detail.Results {
  if result.ErrorMessage != "" {
    fmt.Println(result.SourceDirectoryFileID, "failed:", result.ErrorMessage)
    continue
  }

  if result.JobReference.ID == "" {
    fmt.Println(result.SourceDirectoryFileID, "has no job yet")
    continue
  }

  fmt.Println(
    result.SourceDirectoryFileID,
    result.JobReference.Type,
    result.JobReference.ID,
  )
}

Java

BatchGetResponse detail =
    client.batches()
        .get(BatchGetParams.builder().batchId(batch.id()).addExpand("results").build());

if (detail.status().equals(BatchGetResponse.Status.FAILED)) {
    throw new IllegalStateException("Batch orchestration failed");
}

for (BatchGetResponse.Result result : detail.results().orElse(Collections.emptyList())) {
    if (result.errorMessage().isPresent()) {
        System.out.println(
            result.sourceDirectoryFileId() + " failed: " + result.errorMessage().get());
        continue;
    }

    if (!result.jobReference().isPresent()) {
        System.out.println(result.sourceDirectoryFileId() + " has no job yet");
        continue;
    }

    BatchGetResponse.Result.JobReference ref = result.jobReference().get();
    System.out.println(result.sourceDirectoryFileId() + " " + ref.type() + " " + ref.id());
}

CLI

llp batches get \
  --batch-id "$BATCH_ID" \
  --expand '[results]' \
  | jq -r '
      .results[]
      | if .error_message then "\(.source_directory_file_id) failed: \(.error_message)"
        elif .job_reference == null then "\(.source_directory_file_id) has no job yet"
        else "\(.source_directory_file_id) \(.job_reference.type) \(.job_reference.id)"
        end
    '

results contains references to the underlying Parse or Extract jobs. Fetch those jobs through their product endpoints to inspect job status and outputs.

Python

async def print_job_statuses(results) -> None:
    async with AsyncLlamaCloud() as client:
        for result in results or []:
            ref = result.job_reference
            if ref is None:
                continue

            if ref.type == "parse_v2":
                parse_job = await client.parsing.get(ref.id)
                print(ref.id, parse_job.job.status)
            else:
                extract_job = await client.extract.get(ref.id)
                print(ref.id, extract_job.status)


asyncio.run(print_job_statuses(batch.results))

TypeScript

for (const result of detail.results ?? []) {
  const ref = result.job_reference;
  if (!ref) continue;

  if (ref.type === "parse_v2") {
    const job = await client.parsing.get(ref.id);
    console.log(ref.id, job.job.status);
  } else {
    const job = await client.extract.get(ref.id);
    console.log(ref.id, job.status);
  }
}

Go

for _, result := range detail.Results {
  ref := result.JobReference
  if ref.ID == "" {
    continue
  }

  if ref.Type == llamacloud.BatchGetResponseResultJobReferenceTypeParseV2 {
    job, err := client.Parsing.Get(ctx, ref.ID, llamacloud.ParsingGetParams{})
    if err != nil {
      log.Fatal(err)
    }
    fmt.Println(ref.ID, job.Job.Status)
  } else {
    job, err := client.Extract.Get(ctx, ref.ID, llamacloud.ExtractGetParams{})
    if err != nil {
      log.Fatal(err)
    }
    fmt.Println(ref.ID, job.Status)
  }
}

Java

for (BatchGetResponse.Result result : detail.results().orElse(Collections.emptyList())) {
    if (!result.jobReference().isPresent()) {
        continue;
    }

    BatchGetResponse.Result.JobReference ref = result.jobReference().get();

    if (ref.type().equals(BatchGetResponse.Result.JobReference.Type.PARSE_V2)) {
        System.out.println(ref.id() + " " + client.parsing().get(ref.id()).job().status());
    } else {
        System.out.println(ref.id() + " " + client.extract().get(ref.id()).status());
    }
}

CLI

llp batches get --batch-id "$BATCH_ID" --expand '[results]' \
  | jq -r '.results[] | select(.job_reference != null) | "\(.job_reference.type) \(.job_reference.id)"' \
  | while read -r TYPE ID; do
      if [ "$TYPE" = "parse_v2" ]; then
        echo "$ID $(llp parsing get --job-id "$ID" | jq -r '.job.status')"
      else
        echo "$ID $(llp extract get --job-id "$ID" | jq -r '.status')"
      fi
    done

List Batches

You can list batches for the current project and filter by status or source directory.

Python

async def list_batches(directory_id: str) -> None:
    async with AsyncLlamaCloud() as client:
        async for item in client.batches.list(
            status="RUNNING",
            source_directory_id=directory_id,
        ):
            print(item.id, item.status)


asyncio.run(list_batches(directory_id))

TypeScript

for await (const item of client.batches.list({
  status: "RUNNING",
  source_directory_id: directory.id,
})) {
  console.log(item.id, item.status);
}

Go

iter := client.Batches.ListAutoPaging(ctx, llamacloud.BatchListParams{
  Status:            llamacloud.BatchListParamsStatusRunning,
  SourceDirectoryID: llamacloud.String(directory.ID),
})

for iter.Next() {
  item := iter.Current()
  fmt.Println(item.ID, item.Status)
}

if err := iter.Err(); err != nil {
  log.Fatal(err)
}

Java

BatchListPage page =
    client.batches()
        .list(
            BatchListParams.builder()
                .status(BatchListParams.Status.RUNNING)
                .sourceDirectoryId(directory.id())
                .build());

for (BatchListResponse item : page.autoPager()) {
    System.out.println(item.id() + " " + item.status());
}

CLI

llp batches list \
  --status RUNNING \
  --source-directory-id "$DIRECTORY_ID" \
  --max-items 100 \
  | jq -r '"\(.id) \(.status)"'

Failure Semantics

Batch-level FAILED means the orchestration failed and the batch cannot provide a reliable per-file result set.

Per-file failures are represented in results[*].error_message when a source file could not be processed or mapped to a product job. If a result has a job_reference, use the referenced Parse or Extract job endpoint for the underlying job status and output.

REST Endpoints

Operation	Endpoint
Create batch	POST /api/v2/batches
List batches	GET /api/v2/batches
Get batch	GET /api/v2/batches/{batch_id}