概述

Batches 让你把一个目录里的每个文件都跑同一类产品任务。一个批次会引用：

1. 一个存放待处理文件的源目录。
2. 一个要在每个文件上执行的产品配置 ID。

用 parse_v2 搭配内置的 Parse 预设或已保存的 Parse 配置，即可解析每个文件。用 extract_v2 搭配已保存的 Extract 配置，即可从每个文件中抽取。

批次创建上限为 10,000 个源文件。

前置条件

• 一个 LlamaCloud 账户，套餐为 Pro 或 Enterprise

• 一个 API key（如何创建）

• 要运行的任务的配置 ID：

  • 对于 parse_v2，使用内置预设（如 cfg-PARSE_AGENTIC）或已保存的 Parse 配置 ID。
  • 对于 extract_v2，使用已保存的 Extract 配置 ID。

提示

最省事的建法是在 LlamaCloud UI 的 Parse 或 Extract playground 里创建配置。也可以通过 Product Configurations API 创建。

创建批次

对于一次性的上传，请创建一个 ephemeral 目录，这样源目录会自动进入可清理状态；然后把文件上传进去并创建批次。ephemeral 目录里的文件同样不计入存储计费，也不占用按项目的存储额度。

下面的示例会把 ./invoices 里的每个 PDF 都上传。先下载两份示例发票放进去：

终端窗口

mkdir -p invoices
curl -fL -o invoices/invoice_001.pdf https://media.githubusercontent.com/media/run-llama/llama-datasets/main/llama_agents/invoice-contracts/invoice-2-1.pdf
curl -fL -o invoices/invoice_002.pdf https://media.githubusercontent.com/media/run-llama/llama-datasets/main/llama_agents/invoice-contracts/invoice-3-1.pdf

• Python
• TypeScript
• Go
• Java
• CLI

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

后面的片段都基于这一段 —— 请按顺序把它们粘进同一个文件。

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

const client = new LlamaCloud(); // reads LLAMA_CLOUD_API_KEY from the environment

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

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

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

终端窗口

# llp reads LLAMA_CLOUD_API_KEY from the environment.
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"

若要改跑抽取，请使用一个已保存的 extract_v2 配置 ID，并把 type 设为 "extract_v2"。

轮询直到完成

批次处理是异步的。轮询批次状态直到它进入终态。

• Python
• TypeScript
• Go
• Java
• CLI

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)

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

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)

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

终端窗口

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

echo "$STATUS"

查看逐文件结果

在 get 端点上使用 expand=results，就能带上「源文件 → 产品任务」的映射。批次仍在运行时，结果可能是 null。有结果时，results 会为批次里每个源文件给出一条。逐文件的失败会通过 results[*].error_message 返回；成功的文件则带一个指向底层 Parse 或 Extract 任务的 job_reference。

一份精简后的响应可以在同一批次里同时给出成功与失败的文件条目：

{
  "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
• TypeScript
• Go
• Java
• CLI

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

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,
  );
}

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

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

终端窗口

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 里是指向底层 Parse 或 Extract 任务的引用。要通过各自的产品端点去取这些任务，以查看任务状态与输出。

• Python
• TypeScript
• Go
• Java
• CLI

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

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

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

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

终端窗口

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

列出批次

你可以列出现有 project 的批次，并按状态或源目录过滤。

• Python
• TypeScript
• Go
• Java
• CLI

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

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

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

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

终端窗口

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

失败语义

批次级的 FAILED 表示编排本身失败了，批次无法给出可靠的逐文件结果集。

当某个源文件无法处理、或无法映射到产品任务时，逐文件失败会体现在 results[*].error_message 里。若某条结果带有 job_reference，请通过被引用的 Parse 或 Extract 任务端点查看底层任务的状态与输出。

REST 端点

操作	端点
创建批次	POST /api/v2/batches
列出批次	GET /api/v2/batches
查询批次	GET /api/v2/batches/{batch_id}