异步解析一个目录里的所有 PDF

本例演示如何用 Parse 处理目录中的多个 PDF，并通过 asyncio 与信号量控制并发。你可以边读这份教程边看那个处理异步解析的示例脚本（在 llama_cloud_services 仓库里，给它一个目录名即可）：batch_parse_folder.py

设置

环境变量

设置你的 LLAMA_CLOUD_API_KEY 环境变量：

终端窗口

export LLAMA_CLOUD_API_KEY='llx-...'

或者创建一个 .env 文件：

终端窗口

LLAMA_CLOUD_API_KEY=llx-...

安装依赖

终端窗口

pip install llama-cloud python-dotenv requests

快速开始

下载示例 PDF

先下载用于测试的样例 PDF：

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

用 asyncio 与信号量配合 Parse

批量解析一个目录，意味着同时保持多个解析任务在跑、又不能超出 API 速率限制。下面每个标签页都会解析 sample_files 目录里的每个 PDF，并把并发任务数上限设为 2 —— Python 用 asyncio.Semaphore、TypeScript 用工作线程池、Go 用带缓冲通道的信号量、Java 用固定线程池，CLI 则分批启动后台任务：

Python
TypeScript
Go
Java
CLI

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)

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

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

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

终端窗口

# 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 -er '.job.status') || break
    case "$STATUS" in COMPLETED|FAILED|CANCELLED) break ;; esac
    sleep 2
  done
  if [ "$STATUS" != "COMPLETED" ]; then
    llp parsing get --job-id "$JOB_ID" \
      | jq -r --arg name "$NAME" '"✗ Error parsing \($name): \(.job.error_message // .)"'
    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)"
}

STARTED=0
for FILE_PATH in sample_files/*.pdf; do
  parse_single_file "$FILE_PATH" &
  # Run the files in batches: once MAX_CONCURRENT jobs are in flight,
  # wait for all of them to finish before starting more
  STARTED=$((STARTED + 1))
  if [ $((STARTED % MAX_CONCURRENT)) -eq 0 ]; then wait; fi
done
wait

另外，你也可以用我们提供的 batch_parse_folder.py 脚本，直接作用在你先前建好的 sample_files 目录上：

终端窗口

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

参数：

--input-dir：存放待解析 PDF 的目录
--max-concurrent：控制并发解析操作的最大数量。请依据下列因素调整：
- 你的 API 速率限制（多数账户通常是 5-10）
- 可用的网络带宽
- 服务器容量
- 文件大小（文件较大时可能需要更低并发，以免出现内存问题）

示例输出

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

工作原理

基于信号量的并发控制：用 asyncio.Semaphore 限制并发请求，既防止触发 API 速率限制错误，也便于管理资源占用。
异步处理：每个文件都用 parser.aparse() 异步解析，因此在信号量上限内可有多个文件同时处理。
结果汇总：最后收集并总结全部结果，给出这次解析操作的完整概览。

索引

数据加载

查询

存储

成本分析

应用

聊天机器人

问答

结构化数据

智能体

Chat Engines

Query Engine

连接器

文档与节点

摄取流水线

节点解析器

大语言模型

提示词

回调

节点后处理器

响应合成器

检索器

路由

结构化输出

现有数据

Presets and Modes

能力

起步

能力

Examples

Azure Deployment

Databases and Queues

LLM Integrations

设置

环境变量

安装依赖

快速开始

下载示例 PDF

用 asyncio 与信号量配合 Parse

示例输出

工作原理

友情链接