Parse All PDFs in a Folder (Async)

This example demonstrates how to process multiple PDFs from a folder using Parse with controlled concurrency via asyncio and semaphores. You can follow along with this tutorial alongside an example script that handles async parsing, given a directory name in our llama_cloud_services repository: batch_parse_folder.py

Setup

Environment Variables

Set your LLAMA_CLOUD_API_KEY environment variable:

export LLAMA_CLOUD_API_KEY='llx-...'

Or create a .env file:

LLAMA_CLOUD_API_KEY=llx-...

Install Dependencies

pip install llama-cloud python-dotenv requests

Quick Start

Download Example PDFs

Download sample PDFs to test with:

import os
import requests
from pathlib import Path

# Create sample_files directory
sample_dir = Path("sample_files")
sample_dir.mkdir(exist_ok=True)

# Sample documents to download
sample_docs = {
    "attention.pdf": "https://arxiv.org/pdf/1706.03762.pdf",
    "bert.pdf": "https://arxiv.org/pdf/1810.04805.pdf",
}

# Download sample documents with error handling
for filename, url in sample_docs.items():
    filepath = sample_dir / filename
    if not filepath.exists():
        print(f"📥 Downloading {filename}...")
        try:
            response = requests.get(url, timeout=30)
            response.raise_for_status()

            # Basic content validation
            if response.headers.get('content-type', '').startswith('application/pdf'):
                with open(filepath, "wb") as f:
                    f.write(response.content)
                print(f"   ✅ Downloaded {filename}")
            else:
                print(f"   ⚠️ Warning: {filename} may not be a valid PDF")
        except requests.RequestException as e:
            print(f"   ❌ Failed to download {filename}: {e}")
    else:
        print(f"📁 {filename} already exists")

print("\n✅ Sample files ready!")

Use Asyncio and Semaphore with Parse

Batch-parsing a folder means keeping several parse jobs in flight without exceeding your API rate limits. Each tab below parses every PDF in the sample_files directory, capping the number of concurrent jobs at 2 — Python with an asyncio.Semaphore, TypeScript with a worker pool, Go with a buffered-channel semaphore, Java with a fixed thread pool, and the CLI with backgrounded jobs behind a throttle gate:

Python

import asyncio
import os

from llama_cloud import AsyncLlamaCloud

pdf_files = list(sample_dir.glob("*.pdf"))

# Initialize parser
llama_cloud_client = AsyncLlamaCloud(
    api_key=os.getenv("LLAMA_CLOUD_API_KEY"),
)

# Create semaphore to limit concurrent requests
semaphore = asyncio.Semaphore(2)

# A helper function to parse a single file with semaphore
async def parse_single_file(
    file_path,
    semaphore,
):
    async with semaphore:
        try:
            print(f"Starting parse: {file_path.name}")

            file_obj = await llama_cloud_client.files.create(
                file=str(file_path),
                purpose="parse",
                external_file_id=str(file_path),
            )

            result = await llama_cloud_client.parsing.parse(
                tier="agentic",
                version="latest",
                file_id=file_obj.id,
                expand=["markdown", "text", "items"],
            )

            print(f"✓ Completed: {file_path.name} ({len(result.items.pages)} pages)")

            return {
                "file": file_path.name,
                "status": "success",
                "result": result,
                "pages": len(result.items.pages) if result.items.pages else 0,
            }
        except Exception as e:
            print(f"✗ Error parsing {file_path.name}: {str(e)}")
            return {
                "file": file_path.name,
                "status": "error",
                "error": str(e),
            }

# Create tasks for all files
tasks = [
    parse_single_file(pdf_file, semaphore)
    for pdf_file in pdf_files
]

results = await asyncio.gather(*tasks)

TypeScript

import fs from 'fs';
import path from 'path';

import LlamaCloud from '@llamaindex/llama-cloud';

const sampleDir = 'sample_files';
const pdfFiles = fs
  .readdirSync(sampleDir)
  .filter((name) => name.endsWith('.pdf'))
  .map((name) => path.join(sampleDir, name));

// Initialize parser
const client = new LlamaCloud({ apiKey: process.env.LLAMA_CLOUD_API_KEY });

// Cap on parse jobs in flight at once
const MAX_CONCURRENT = 2;

type ParseOutcome = {
  file: string;
  status: 'success' | 'error';
  pages?: number;
  error?: string;
};

// A helper function to parse a single file
async function parseSingleFile(filePath: string): Promise<ParseOutcome> {
  const name = path.basename(filePath);
  try {
    console.log(`Starting parse: ${name}`);

    const fileObj = await client.files.create({
      file: fs.createReadStream(filePath),
      purpose: 'parse',
      external_file_id: filePath,
    });

    const result = await client.parsing.parse({
      tier: 'agentic',
      version: 'latest',
      file_id: fileObj.id,
      expand: ['markdown', 'text', 'items'],
    });

    const pages = result.items?.pages.length ?? 0;
    console.log(`✓ Completed: ${name} (${pages} pages)`);

    return { file: name, status: 'success', pages };
  } catch (error) {
    console.log(`✗ Error parsing ${name}: ${error}`);
    return { file: name, status: 'error', error: String(error) };
  }
}

// Worker pool: MAX_CONCURRENT workers pull from a shared cursor, so no more
// than MAX_CONCURRENT parse jobs are ever in flight
const results: ParseOutcome[] = [];
let cursor = 0;

await Promise.all(
  Array.from({ length: Math.min(MAX_CONCURRENT, pdfFiles.length) }, async () => {
    while (cursor < pdfFiles.length) {
      const index = cursor++;
      results[index] = await parseSingleFile(pdfFiles[index]!);
    }
  }),
);

Go

package main

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

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

// Cap on parse jobs in flight at once
const maxConcurrent = 2

type parseOutcome struct {
  File   string
  Status string
  Pages  int
  Err    error
}

// A helper function to parse a single file
func parseSingleFile(ctx context.Context, client *llamacloud.Client, filePath string) parseOutcome {
  name := filepath.Base(filePath)
  fmt.Printf("Starting parse: %s\n", name)

  f, err := os.Open(filePath)
  if err != nil {
    return parseOutcome{File: name, Status: "error", Err: err}
  }
  defer f.Close()

  fileObj, err := client.Files.New(ctx, llamacloud.FileNewParams{
    File:           f,
    Purpose:        "parse",
    ExternalFileID: llamacloud.String(filePath),
  })
  if err != nil {
    return parseOutcome{File: name, Status: "error", Err: err}
  }

  job, err := client.Parsing.New(ctx, llamacloud.ParsingNewParams{
    FileID:  llamacloud.String(fileObj.ID),
    Tier:    llamacloud.ParsingNewParamsTierAgentic,
    Version: llamacloud.ParsingNewParamsVersionLatest,
  })
  if err != nil {
    return parseOutcome{File: name, Status: "error", Err: err}
  }

  // Poll until the job reaches a terminal status
  getParams := llamacloud.ParsingGetParams{Expand: []string{"markdown", "text", "items"}}
  result, err := client.Parsing.Get(ctx, job.ID, getParams)
  if err != nil {
    return parseOutcome{File: name, Status: "error", Err: err}
  }
  for result.Job.Status != "COMPLETED" && result.Job.Status != "FAILED" && result.Job.Status != "CANCELLED" {
    time.Sleep(2 * time.Second)
    result, err = client.Parsing.Get(ctx, job.ID, getParams)
    if err != nil {
      return parseOutcome{File: name, Status: "error", Err: err}
    }
  }
  if result.Job.Status != "COMPLETED" {
    return parseOutcome{File: name, Status: "error", Err: fmt.Errorf("parse ended as %s", result.Job.Status)}
  }

  pages := len(result.Items.Pages)
  fmt.Printf("✓ Completed: %s (%d pages)\n", name, pages)

  return parseOutcome{File: name, Status: "success", Pages: pages}
}

func main() {
  ctx := context.Background()

  pdfFiles, err := filepath.Glob(filepath.Join("sample_files", "*.pdf"))
  if err != nil {
    log.Fatal(err)
  }

  // Initialize parser
  client := llamacloud.NewClient()

  // A buffered channel acts as the semaphore: a goroutine can only run while
  // it holds one of the maxConcurrent slots
  sem := make(chan struct{}, maxConcurrent)
  results := make([]parseOutcome, len(pdfFiles))

  var wg sync.WaitGroup
  for i, filePath := range pdfFiles {
    wg.Add(1)
    go func(i int, filePath string) {
      defer wg.Done()
      sem <- struct{}{}
      defer func() { <-sem }()
      results[i] = parseSingleFile(ctx, &client, filePath)
    }(i, filePath)
  }
  wg.Wait()

  for _, r := range results {
    if r.Status != "success" {
      fmt.Printf("✗ Error parsing %s: %v\n", r.File, r.Err)
    }
  }
}

Java

import ai.llamaindex.llamacloud.client.LlamaCloudClient;
import ai.llamaindex.llamacloud.client.okhttp.LlamaCloudOkHttpClient;
import ai.llamaindex.llamacloud.models.files.FileCreateParams;
import ai.llamaindex.llamacloud.models.files.FileCreateResponse;
import ai.llamaindex.llamacloud.models.parsing.ParsingCreateParams;
import ai.llamaindex.llamacloud.models.parsing.ParsingCreateResponse;
import ai.llamaindex.llamacloud.models.parsing.ParsingGetParams;
import ai.llamaindex.llamacloud.models.parsing.ParsingGetResponse;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class BatchParseFolder {

    // Cap on parse jobs in flight at once
    private static final int MAX_CONCURRENT = 2;

    public static void main(String[] args) throws Exception {
        List<Path> pdfFiles = new ArrayList<>();
        try (DirectoryStream<Path> pdfs = Files.newDirectoryStream(Paths.get("sample_files"), "*.pdf")) {
            for (Path pdf : pdfs) {
                pdfFiles.add(pdf);
            }
        }

        // Initialize parser
        LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();

        // A fixed thread pool is the semaphore: at most MAX_CONCURRENT submitted
        // tasks run at a time, the rest queue up
        ExecutorService pool = Executors.newFixedThreadPool(MAX_CONCURRENT);
        List<Future<String>> futures = new ArrayList<>();
        for (Path pdf : pdfFiles) {
            futures.add(pool.submit(() -> parseSingleFile(client, pdf)));
        }

        for (Future<String> future : futures) {
            System.out.println(future.get());
        }
        pool.shutdown();
    }

    // A helper method to parse a single file
    private static String parseSingleFile(LlamaCloudClient client, Path filePath) {
        String name = filePath.getFileName().toString();
        try {
            System.out.println("Starting parse: " + name);

            FileCreateResponse fileObj = client.files().create(FileCreateParams.builder()
                    .file(filePath)
                    .purpose("parse")
                    .externalFileId(filePath.toString())
                    .build());

            ParsingCreateResponse job = client.parsing().create(ParsingCreateParams.builder()
                    .fileId(fileObj.id())
                    .tier(ParsingCreateParams.Tier.AGENTIC)
                    .version(ParsingCreateParams.Version.LATEST)
                    .build());

            // Poll until the job reaches a terminal status
            ParsingGetParams getParams = ParsingGetParams.builder()
                    .jobId(job.id())
                    .addExpand("markdown")
                    .addExpand("text")
                    .addExpand("items")
                    .build();

            ParsingGetResponse result = client.parsing().get(getParams);
            while (!result.job().status().equals(ParsingGetResponse.Job.Status.COMPLETED)
                    && !result.job().status().equals(ParsingGetResponse.Job.Status.FAILED)
                    && !result.job().status().equals(ParsingGetResponse.Job.Status.CANCELLED)) {
                Thread.sleep(2000);
                result = client.parsing().get(getParams);
            }
            if (!result.job().status().equals(ParsingGetResponse.Job.Status.COMPLETED)) {
                return "✗ Error parsing " + name + ": parse ended as " + result.job().status();
            }

            int pages = result.items().map(items -> items.pages().size()).orElse(0);

            return "✓ Completed: " + name + " (" + pages + " pages)";
        } catch (Exception e) {
            return "✗ Error parsing " + name + ": " + e.getMessage();
        }
    }
}

CLI

# Cap on parse jobs in flight at once
MAX_CONCURRENT=2

# A helper function to parse a single file
parse_single_file() {
  FILE_PATH="$1"
  NAME=$(basename "$FILE_PATH")
  echo "Starting parse: $NAME"

  FILE_ID=$(llp files create \
    --file "$FILE_PATH" \
    --purpose parse \
    --external-file-id "$FILE_PATH" | jq -r '.id')

  JOB_ID=$(llp parsing create \
    --file-id "$FILE_ID" \
    --tier agentic \
    --version latest | jq -r '.id')

  # Poll until the job reaches a terminal status
  while true; do
    STATUS=$(llp parsing get --job-id "$JOB_ID" | jq -r '.job.status')
    case "$STATUS" in COMPLETED|FAILED|CANCELLED) break ;; esac
    sleep 2
  done
  if [ "$STATUS" != "COMPLETED" ]; then
    echo "✗ Error parsing $NAME: parse ended as $STATUS"
    return 1
  fi

  PAGES=$(llp parsing get --job-id "$JOB_ID" \
    --expand markdown \
    --expand text \
    --expand items | jq '.items.pages | length')
  echo "✓ Completed: $NAME ($PAGES pages)"
}

for FILE_PATH in sample_files/*.pdf; do
  # Throttle gate: block until a slot frees up, so no more than
  # MAX_CONCURRENT background jobs are ever in flight
  while [ "$(jobs -rp | wc -l)" -ge "$MAX_CONCURRENT" ]; do
    sleep 1
  done
  parse_single_file "$FILE_PATH" &
done
wait

Alternatively, you can use the batch_parse_folder.py script we’ve provided, which you can use with the sample_files directory you created before:

python batch_parse_folder.py --input-dir ./sample_files --max-concurrent 5

Parameters:

• --input-dir: Directory containing PDF files to parse
• --max-concurrent: Controls the maximum number of concurrent parse operations. Adjust based on:
  • Your API rate limits (typically 5-10 for most accounts)
  • Available network bandwidth
  • Server capacity
  • File sizes (larger files may require lower concurrency to avoid memory issues)

Example Output

Found 2 PDF files to parse
Processing 2 files with max 5 concurrent operations...
Starting parse: attention.pdf
Starting parse: bert.pdf
Started parsing the file under job_id 1a7b8f3b-9119-4e38-954d-b67b8e96b3d6
Started parsing the file under job_id 28123aeb-dd3e-4398-b754-0cb101a3b78b
✓ Completed: attention.pdf (15 pages)
✓ Completed: bert.pdf (16 pages)
PARSE SUMMARY

Total files: 2
Successful: 2
Failed: 0
Total time: 10.00 seconds
Average time per file: 5.00 seconds

How It Works

1. Semaphore-based Concurrency: Uses asyncio.Semaphore to limit concurrent requests, preventing API rate limit errors and managing resource usage.

2. Async Processing: Each file is parsed asynchronously using parser.aparse(), allowing multiple files to be processed concurrently up to the semaphore limit.

3. Result Aggregation: All results are collected and summarized at the end, providing a complete overview of the parsing operation.