async batch

upd demo results

README.md

@@ -1,6 +1,6 @@
 # LlamaParse (Preview)
 
-LlamaParse is an API created by LlamaIndex to effeciently parse and represent files for effecient retrieval and context augmentation using LlamaIndex frameworks.
+LlamaParse is an API created by LlamaIndex to efficiently parse and represent files for efficient retrieval and context augmentation using LlamaIndex frameworks.
 
 LlamaParse directly integrates with [LlamaIndex](https://github.com/run-llama/llama_index).
 
@@ -26,14 +26,22 @@ from llama_parse import LlamaParse
 
 parser = LlamaParse(
     api_key="llx-...",  # can also be set in your env as LLAMA_CLOUD_API_KEY
-    result_type="markdown"  # "markdown" and "text" are available
+    result_type="markdown",  # "markdown" and "text" are available
+    num_workers=4, # if multiple files passed, split in `num_workers` API calls
+    verbose=True
 )
 
 # sync
 documents = parser.load_data("./my_file.pdf")
 
+# sync batch
+documents = parser.load_data(["./my_file1.pdf", "./my_file2.pdf"])
+
 # async
 documents = await parser.aload_data("./my_file.pdf")
+
+# async batch
+documents = await parser.aload_data(["./my_file1.pdf", "./my_file2.pdf"])
 ```
 
 ## Using with `SimpleDirectoryReader`
@@ -49,7 +57,8 @@ from llama_index.core import SimpleDirectoryReader
 
 parser = LlamaParse(
     api_key="llx-...",  # can also be set in your env as LLAMA_CLOUD_API_KEY
-    result_type="markdown"  # "markdown" and "text" are available
+    result_type="markdown",  # "markdown" and "text" are available
+    verbose=True
 )
 
 file_extractor = {".pdf": parser}
@@ -60,8 +69,12 @@ Full documentation for `SimpleDirectoryReader` can be found on the [LlamaIndex D
 
 ## Examples
 
-Serveral end-to-end indexing examples can be found in the examples folder
+Several end-to-end indexing examples can be found in the examples folder
 
 - [Getting Started](examples/demo_basic.ipynb)
 - [Advanced RAG Example](examples/demo_advanced.ipynb)
 - [Raw API Usage](examples/demo_api.ipynb)
+
+## Terms of Service
+
+See the [Terms of Service Here](./TOS.pdf).

llama_parse/base.py

@@ -1,14 +1,19 @@
+import os
 import asyncio
 import httpx
 import mimetypes
 import time
 from enum import Enum
-from typing import List, Optional
+from typing import List, Optional, Union
 
-from llama_index.bridge.pydantic import Field, validator
-from llama_index.readers.base import BasePydanticReader
-from llama_index.schema import Document
+from llama_index.core.async_utils import run_jobs
+from llama_index.core.bridge.pydantic import Field, validator
+from llama_index.core.constants import DEFAULT_BASE_URL
+from llama_index.core.readers.base import BasePydanticReader
+from llama_index.core.schema import Document
 
+nest_asyncio_err = "cannot be called from a running event loop"
+nest_asyncio_msg = "The event loop is already running. Add `import nest_asyncio; nest_asyncio.apply()` to your code to fix this issue."
 
 class ResultType(str, Enum):
     """The result type for the parser."""
@@ -22,12 +27,18 @@ class LlamaParse(BasePydanticReader):
 
     api_key: str = Field(default="", description="The API key for the LlamaParse API.")
     base_url: str = Field(
-        default="https://api.cloud.llamaindex.ai/api/parsing",
+        default=DEFAULT_BASE_URL,
         description="The base URL of the Llama Parsing API.",
     )
     result_type: ResultType = Field(
         default=ResultType.TXT, description="The result type for the parser."
     )
+    num_workers: int = Field(
+        default=4,
+        gt=0,
+        lt=10, 
+        description="The number of workers to use sending API requests for parsing."
+    )
     check_interval: int = Field(
         default=1,
         description="The interval in seconds to check if the parsing is done.",
@@ -36,6 +47,9 @@ class LlamaParse(BasePydanticReader):
         default=2000,
         description="The maximum timeout in seconds to wait for the parsing to finish.",
     )
+    verbose: bool = Field(
+        default=True, description="Whether to print the progress of the parsing."
+    )
 
     @validator("api_key", pre=True, always=True)
     def validate_api_key(cls, v: str) -> str:
@@ -48,55 +62,102 @@ class LlamaParse(BasePydanticReader):
             return api_key
         
         return v
+    
+    @validator("base_url", pre=True, always=True)
+    def validate_base_url(cls, v: str) -> str:
+        """Validate the base URL."""
+        url = os.getenv("LLAMA_CLOUD_BASE_URL", None)
+        return url or v or DEFAULT_BASE_URL
 
-    def load_data(self, file_path: str, extra_info: Optional[dict] = None) -> List[Document]:
+    async def _aload_data(self, file_path: str, extra_info: Optional[dict] = None) -> List[Document]:
         """Load data from the input path."""
-        return asyncio.run(self.aload_data(file_path, extra_info))
+        try:
+            file_path = str(file_path)
+            if not file_path.endswith(".pdf"):
+                raise Exception("Currently, only PDF files are supported.")
 
-    async def aload_data(self, file_path: str, extra_info: Optional[dict] = None) -> List[Document]:
-        """Load data from the input path."""
-        file_path = str(file_path)
-        if not file_path.endswith(".pdf"):
-            raise Exception("Currently, only PDF files are supported.")
+            extra_info = extra_info or {}
+            extra_info["file_path"] = file_path
 
-        extra_info = extra_info or {}
-        extra_info["file_path"] = file_path
+            headers = {"Authorization": f"Bearer {self.api_key}"}
 
-        headers = {"Authorization": f"Bearer {self.api_key}"}
+            # load data, set the mime type
+            with open(file_path, "rb") as f:
+                mime_type = mimetypes.guess_type(file_path)[0]
+                files = {"file": (f.name, f, mime_type)}
 
-        # load data, set the mime type
-        with open(file_path, "rb") as f:
-            mime_type = mimetypes.guess_type(file_path)[0]
-            files = {"file": (f.name, f, mime_type)}
+                # send the request, start job
+                url = f"{self.base_url}/api/parsing/upload"
+                async with httpx.AsyncClient(timeout=self.max_timeout) as client:
+                    response = await client.post(url, files=files, headers=headers)
+                    if not response.is_success:
+                        raise Exception(f"Failed to parse the PDF file: {response.text}")
 
-            # send the request, start job
-            url = f"{self.base_url}/upload"
-            async with httpx.AsyncClient() as client:
-                response = await client.post(url, files=files, headers=headers)
-                if not response.is_success:
-                    raise Exception(f"Failed to parse the PDF file: {response.text}")
+            # check the status of the job, return when done
+            job_id = response.json()["id"]
+            if self.verbose:
+                print("Started parsing the file under job_id %s" % job_id)
+            
+            result_url = f"{self.base_url}/api/parsing/job/{job_id}/result/{self.result_type.value}"
 
-        # check the status of the job, return when done
-        job_id = response.json()["id"]
-        print("Started parsing the file under job_id %s" % job_id)
-        result_url = f"{self.base_url}/job/{job_id}/result/{self.result_type.value}"
+            start = time.time()
+            tries = 0
+            while True:
+                await asyncio.sleep(self.check_interval)
+                async with httpx.AsyncClient(timeout=self.max_timeout) as client: 
+                    tries += 1   
+                    
+                    result = await client.get(result_url, headers=headers)
 
-        start = time.time()
-        while True:
-            await asyncio.sleep(self.check_interval)
-            async with httpx.AsyncClient() as client:    
-                result = await client.get(result_url, headers=headers)
+                    if result.status_code == 404:
+                        end = time.time()
+                        if end - start > self.max_timeout:
+                            raise Exception(
+                                f"Timeout while parsing the PDF file: {response.text}"
+                            )
+                        if self.verbose and tries % 10 == 0:
+                            print(".", end="", flush=True)
+                        continue
 
-                if not result.is_success:
-                    if time.time() - start > self.max_timeout:
-                        raise Exception(
-                            f"Timeout while parsing the PDF file: {response.text}"
+                    if result.status_code == 400:
+                        detail = result.json().get("detail", "Unknown error")
+                        raise Exception(f"Failed to parse the PDF file: {detail}")
+
+                    return [
+                        Document(
+                            text=result.json()[self.result_type.value],
+                            metadata=extra_info,
                         )
-                    continue
+                    ]
+        except Exception as e:
+            print("Error while parsing the PDF file: ", e)
+            return []
+    
+    async def aload_data(self, file_path: Union[List[str], str], extra_info: Optional[dict] = None) -> List[Document]:
+        """Load data from the input path."""
+        if isinstance(file_path, str):
+            return await self._aload_data(file_path, extra_info=extra_info)
+        elif isinstance(file_path, list):
+            jobs = [self._aload_data(f, extra_info=extra_info) for f in file_path]
+            try:
+                results = await run_jobs(jobs, workers=self.num_workers)
+                
+                # return flattened results
+                return [item for sublist in results for item in sublist]
+            except RuntimeError as e:
+                if nest_asyncio_err in str(e):
+                    raise RuntimeError(nest_asyncio_msg)
+                else:
+                    raise e
+        else:
+            raise ValueError("The input file_path must be a string or a list of strings.")
 
-                return [
-                    Document(
-                        text=result.json()[self.result_type.value],
-                        metadata=extra_info,
-                    )
-                ]
+    def load_data(self, file_path: Union[List[str], str], extra_info: Optional[dict] = None) -> List[Document]:
+        """Load data from the input path."""
+        try:
+            return asyncio.run(self.aload_data(file_path, extra_info))
+        except RuntimeError as e:
+            if nest_asyncio_err in str(e):
+                raise RuntimeError(nest_asyncio_msg)
+            else:
+                raise e

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "llama-parse"
-version = "0.1.3"
+version = "0.3.3"
 description = "Parse files into RAG-Optimized formats."
 authors = ["Logan Markewich <logan@llamaindex.ai>"]
 license = "MIT"
@@ -9,8 +9,7 @@ packages = [{include = "llama_parse"}]
 
 [tool.poetry.dependencies]
 python = ">=3.8.1,<4.0"
-llama-index = "^0.9.40"
-
+llama-index-core = ">=0.10.7"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.0.0"