factor out retry decision

* batch parse sctript with asyncio

* lint

---------

Co-authored-by: Logan Markewich <logan.markewich@live.com>

.changeset/silly-dryers-scream.md

@@ -0,0 +1,5 @@
+---
+"llama-cloud-services": patch
+---
+
+Improve parse results polling

examples/parse/batch_parse_folder.py

@@ -0,0 +1,183 @@
+"""
+Example: Batch Processing a Folder of PDFs with LlamaParse
+
+This script demonstrates how to process multiple PDFs from a folder
+using LlamaParse with controlled concurrency using asyncio and semaphores.
+
+Usage:
+    python batch_parse_folder.py --input-dir ./pdfs --max-concurrent 5
+"""
+
+import asyncio
+import argparse
+from pathlib import Path
+from typing import List, Dict, Any
+from datetime import datetime
+from dotenv import load_dotenv
+import os
+
+from llama_cloud_services import LlamaParse
+
+# Load environment variables from .env file
+load_dotenv()
+
+
+async def parse_single_file(
+    parser: LlamaParse,
+    file_path: Path,
+    semaphore: asyncio.Semaphore,
+) -> Dict[str, Any]:
+    """
+    Parse a single PDF file with concurrency control.
+
+    Args:
+        parser: LlamaParse instance
+        file_path: Path to the PDF file
+        semaphore: Semaphore to control concurrent requests
+
+    Returns:
+        Dictionary with file info and parse result
+    """
+    async with semaphore:
+        try:
+            print(f"Starting parse: {file_path.name}")
+
+            result = await parser.aparse(str(file_path))
+
+            print(f"✓ Completed: {file_path.name} ({len(result.pages)} pages)")
+
+            return {
+                "file": file_path.name,
+                "status": "success",
+                "result": result,
+                "pages": len(result.pages) if result.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),
+            }
+
+
+async def parse_folder(
+    input_dir: Path,
+    max_concurrent: int = 5,
+    api_key: str = None,
+) -> List[Dict[str, any]]:
+    """
+    Parse all PDFs in a folder with controlled concurrency.
+
+    Args:
+        input_dir: Directory containing PDF files
+        max_concurrent: Maximum number of concurrent parse operations
+        api_key: LlamaCloud API key (loaded from .env file)
+
+    Returns:
+        List of parse results for each file
+    """
+    # Find all PDF files
+    pdf_files = list(input_dir.glob("*.pdf"))
+
+    if not pdf_files:
+        print(f"No PDF files found in {input_dir}")
+        return []
+
+    print(f"Found {len(pdf_files)} PDF files to parse")
+
+    # Initialize parser
+    parser = LlamaParse(
+        api_key=api_key,
+        num_workers=1,  # We control concurrency with semaphore
+        show_progress=False,  # We'll show our own progress
+    )
+
+    # Create semaphore to limit concurrent requests
+    semaphore = asyncio.Semaphore(max_concurrent)
+
+    # Create tasks for all files
+    tasks = [parse_single_file(parser, pdf_file, semaphore) for pdf_file in pdf_files]
+
+    # Run all tasks concurrently (but limited by semaphore)
+    print(
+        f"Processing {len(tasks)} files with max {max_concurrent} concurrent operations..."
+    )
+    start_time = datetime.now()
+
+    results = await asyncio.gather(*tasks)
+
+    end_time = datetime.now()
+    duration = (end_time - start_time).total_seconds()
+
+    # Process results
+    successful = [
+        r for r in results if isinstance(r, dict) and r.get("status") == "success"
+    ]
+    failed = [r for r in results if isinstance(r, dict) and r.get("status") == "error"]
+
+    # Print summary
+    print("PARSE SUMMARY \n")
+    print(f"Total files: {len(pdf_files)}")
+    print(f"Successful: {len(successful)}")
+    print(f"Failed: {len(failed)}")
+    print(f"Total time: {duration:.2f} seconds")
+    print(f"Average time per file: {duration / len(pdf_files):.2f} seconds")
+
+    if failed:
+        print("\nFailed files:")
+        for result in failed:
+            print(f"  - {result['file']}: {result.get('error', 'Unknown error')}")
+
+    return results
+
+
+def main():
+    """Main entry point for the script."""
+    parser = argparse.ArgumentParser(
+        description="Batch process PDFs in a folder with LlamaParse"
+    )
+    parser.add_argument(
+        "--input-dir",
+        type=str,
+        required=True,
+        help="Directory containing PDF files to parse",
+    )
+    parser.add_argument(
+        "--max-concurrent",
+        type=int,
+        default=5,
+        help="Maximum number of concurrent parse operations (default: 5)",
+    )
+
+    args = parser.parse_args()
+
+    input_dir = Path(args.input_dir)
+
+    # Validate input directory
+    if not input_dir.exists():
+        print(f"Error: Input directory does not exist: {input_dir}")
+        return
+
+    if not input_dir.is_dir():
+        print(f"Error: Input path is not a directory: {input_dir}")
+        return
+
+    # Get API key from environment (loaded from .env file)
+    api_key = os.getenv("LLAMA_CLOUD_API_KEY")
+    if not api_key:
+        print("Error: LLAMA_CLOUD_API_KEY not found. Please set it in your .env file")
+        return
+
+    # Run async function
+    asyncio.run(
+        parse_folder(
+            input_dir=input_dir,
+            max_concurrent=args.max_concurrent,
+            api_key=api_key,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

ts/llama_cloud_services/src/reader.ts

@@ -1,9 +1,10 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 import { type Client, createClient, createConfig } from "@hey-api/client-fetch";
+import { type FailedAttemptError } from "p-retry";
 import { Document, FileReader } from "@llamaindex/core/schema";
 import { fs, getEnv, path } from "@llamaindex/env";
 import {
-  type BodyUploadFileApiParsingUploadPost,
+  type BodyUploadFileApiV1ParsingUploadPost,
   type FailPageMode,
   type ParserLanguages,
   type ParsingMode,
@@ -32,6 +33,33 @@ type WriteStream = {
 // eslint-disable-next-line no-var
 var process: any;
 
+function handleFailedAttempt(
+  error: FailedAttemptError,
+  jobId: string,
+  verbose: boolean,
+) {
+  // Retry only on 5XX or socket errors.
+  const status = (error.cause as any)?.response?.status;
+  if (
+    !(
+      (status && status >= 500 && status < 600) ||
+      ((error.cause as any)?.code &&
+        ((error.cause as any).code === "ECONNRESET" ||
+          (error.cause as any).code === "ETIMEDOUT" ||
+          (error.cause as any).code === "ECONNREFUSED")) ||
+      (status && status === 404)
+    )
+  ) {
+    throw error;
+  }
+
+  if (verbose) {
+    console.warn(
+      `Attempting to get job ${jobId} result (attempt ${error.attemptNumber}) failed. Retrying...`,
+    );
+  }
+}
+
 /**
  * Represents a reader for parsing files using the LlamaParse API.
  * See https://github.com/run-llama/llama_parse
@@ -188,6 +216,16 @@ export class LlamaParseReader extends FileReader {
   extract_printed_page_number?: boolean | undefined;
   tier?: string | undefined;
   version?: string | undefined;
+  layout_aware?: boolean | undefined;
+  line_level_bounding_box?: boolean | undefined;
+  specialized_image_parsing?: boolean | undefined;
+  aggressive_table_extraction?: boolean | undefined;
+  preserve_very_small_text?: boolean | undefined;
+  spreadsheet_force_formula_computation?: boolean | undefined;
+  inline_images_in_markdown?: boolean | undefined;
+  keep_page_separator_when_merging_tables?: boolean | undefined;
+  remove_hidden_text?: boolean | undefined;
+  presentation_out_of_bounds_content?: boolean | undefined;
 
   constructor(
     params: Partial<Omit<LlamaParseReader, "language" | "apiKey">> & {
@@ -387,11 +425,25 @@ export class LlamaParseReader extends FileReader {
       extract_printed_page_number: this.extract_printed_page_number,
       tier: this.tier,
       version: this.version,
+      layout_aware: this.layout_aware,
+      line_level_bounding_box: this.line_level_bounding_box,
+      specialized_image_parsing: this.specialized_image_parsing,
+      aggressive_table_extraction: this.aggressive_table_extraction,
+      preserve_very_small_text: this.preserve_very_small_text,
+      spreadsheet_force_formula_computation:
+        this.spreadsheet_force_formula_computation,
+      inline_images_in_markdown: this.inline_images_in_markdown,
+      webhook_configurations: undefined,
+      keep_page_separator_when_merging_tables:
+        this.keep_page_separator_when_merging_tables,
+      remove_hidden_text: this.remove_hidden_text,
+      presentation_out_of_bounds_content:
+        this.presentation_out_of_bounds_content,
     } satisfies {
-      [Key in keyof BodyUploadFileApiParsingUploadPost]-?:
-        | BodyUploadFileApiParsingUploadPost[Key]
+      [Key in keyof BodyUploadFileApiV1ParsingUploadPost]-?:
+        | BodyUploadFileApiV1ParsingUploadPost[Key]
         | undefined;
-    } as unknown as BodyUploadFileApiParsingUploadPost;
+    } as unknown as BodyUploadFileApiV1ParsingUploadPost;
 
     const response = await uploadFileApiV1ParsingUploadPost({
       client: this.#client,
@@ -443,26 +495,8 @@ export class LlamaParseReader extends FileReader {
             }),
           {
             retries: this.maxErrorCount,
-            onFailedAttempt: (error) => {
-              // Retry only on 5XX or socket errors.
-              const status = (error.cause as any)?.response?.status;
-              if (
-                !(
-                  (status && status >= 500 && status < 600) ||
-                  ((error.cause as any)?.code &&
-                    ((error.cause as any).code === "ECONNRESET" ||
-                      (error.cause as any).code === "ETIMEDOUT" ||
-                      (error.cause as any).code === "ECONNREFUSED"))
-                )
-              ) {
-                throw error;
-              }
-              if (this.verbose) {
-                console.warn(
-                  `Attempting to get job ${jobId} result (attempt ${error.attemptNumber}) failed. Retrying...`,
-                );
-              }
-            },
+            onFailedAttempt: (error) =>
+              handleFailedAttempt(error, jobId, this.verbose),
           },
         );
       } catch (e: any) {
@@ -475,49 +509,69 @@ export class LlamaParseReader extends FileReader {
       const status = (data as Record<string, unknown>)["status"];
 
       if (status === "SUCCESS") {
-        let resultData;
         switch (resultType) {
           case "json": {
-            resultData =
-              await getJobJsonResultApiV1ParsingJobJobIdResultJsonGet({
-                client: this.#client,
-                throwOnError: true,
-                path: { job_id: jobId },
-                query: {
-                  organization_id: this.organization_id ?? null,
-                },
-                signal: AbortSignal.timeout(this.maxTimeout * 1000),
-              });
-            break;
+            const resultData = await pRetry(
+              () =>
+                getJobJsonResultApiV1ParsingJobJobIdResultJsonGet({
+                  client: this.#client,
+                  throwOnError: true,
+                  path: { job_id: jobId },
+                  query: {
+                    organization_id: this.organization_id ?? null,
+                  },
+                  signal: AbortSignal.timeout(this.maxTimeout * 1000),
+                }),
+              {
+                retries: this.maxErrorCount,
+                onFailedAttempt: (error) =>
+                  handleFailedAttempt(error, jobId, this.verbose),
+              },
+            );
+            return resultData.data;
           }
           case "markdown": {
-            resultData =
-              await getJobResultApiV1ParsingJobJobIdResultMarkdownGet({
-                client: this.#client,
-                throwOnError: true,
-                path: { job_id: jobId },
-                query: {
-                  organization_id: this.organization_id ?? null,
-                },
-                signal: AbortSignal.timeout(this.maxTimeout * 1000),
-              });
-            break;
+            const resultData = await pRetry(
+              () =>
+                getJobResultApiV1ParsingJobJobIdResultMarkdownGet({
+                  client: this.#client,
+                  throwOnError: true,
+                  path: { job_id: jobId },
+                  query: {
+                    organization_id: this.organization_id ?? null,
+                  },
+                  signal: AbortSignal.timeout(this.maxTimeout * 1000),
+                }),
+              {
+                retries: this.maxErrorCount,
+                onFailedAttempt: (error) =>
+                  handleFailedAttempt(error, jobId, this.verbose),
+              },
+            );
+            return resultData.data;
           }
           case "text": {
-            resultData =
-              await getJobTextResultApiV1ParsingJobJobIdResultTextGet({
-                client: this.#client,
-                throwOnError: true,
-                path: { job_id: jobId },
-                query: {
-                  organization_id: this.organization_id ?? null,
-                },
-                signal: AbortSignal.timeout(this.maxTimeout * 1000),
-              });
-            break;
+            const resultData = await pRetry(
+              () =>
+                getJobTextResultApiV1ParsingJobJobIdResultTextGet({
+                  client: this.#client,
+                  throwOnError: true,
+                  path: { job_id: jobId },
+                  query: {
+                    organization_id: this.organization_id ?? null,
+                  },
+                  signal: AbortSignal.timeout(this.maxTimeout * 1000),
+                }),
+              {
+                retries: this.maxErrorCount,
+                onFailedAttempt: (error) =>
+                  handleFailedAttempt(error, jobId, this.verbose),
+              },
+            );
+
+            return resultData.data;
           }
         }
-        return resultData.data;
       } else if (status === "PENDING") {
         if (this.verbose && tries % 10 === 0) {
           this.stdout?.write(".");