Add base

llama_cloud_services/extract/extract.py

@@ -22,7 +22,6 @@ from llama_cloud import (
     Project,
     ExtractTarget,
     LlamaExtractSettings,
-    PaginatedExtractRunsResponse,
 )
 from llama_cloud.client import AsyncLlamaCloud
 from llama_cloud_services.extract.utils import JSONObjectType, augment_async_errors
@@ -57,8 +56,6 @@ class ExtractionAgent:
         num_workers: int = 4,
         show_progress: bool = True,
         verbose: bool = False,
-        verify: Optional[bool] = True,
-        httpx_timeout: Optional[float] = 60,
     ):
         self._client = client
         self._agent = agent
@@ -68,8 +65,6 @@ class ExtractionAgent:
         self.max_timeout = max_timeout
         self.num_workers = num_workers
         self.show_progress = show_progress
-        self.verify = verify
-        self.httpx_timeout = httpx_timeout
         self._verbose = verbose
         self._data_schema: Union[JSONObjectType, None] = None
         self._config: Union[ExtractConfig, None] = None
@@ -82,20 +77,14 @@ class ExtractionAgent:
 
         def run_coro() -> T:
             async def wrapped_coro() -> T:
-                # Get the original client to preserve its configuration
-                original_client = self._client._client_wrapper.httpx_client
-
-                # Create a new client with the same configuration as the original
                 async with httpx.AsyncClient(
-                    verify=self.verify,
-                    timeout=self.httpx_timeout,
+                    timeout=self._client._client_wrapper.httpx_client.timeout,
                 ) as client:
-                    # Temporarily replace the client
+                    original_client = self._client._client_wrapper.httpx_client
                     self._client._client_wrapper.httpx_client = client
                     try:
                         return await coro
                     finally:
-                        # Restore the original client
                         self._client._client_wrapper.httpx_client = original_client
 
             return asyncio.run(wrapped_coro())
@@ -391,29 +380,15 @@ class ExtractionAgent:
             )
         )
 
-    def delete_extraction_run(self, run_id: str) -> None:
-        """Delete an extraction run by ID.
-
-        Args:
-            run_id (str): The ID of the extraction run to delete
-        """
-        self._run_in_thread(
-            self._client.llama_extract.delete_extraction_run(run_id=run_id)
-        )
-
-    def list_extraction_runs(
-        self, page: int = 0, limit: int = 100
-    ) -> PaginatedExtractRunsResponse:
+    def list_extraction_runs(self) -> List[ExtractRun]:
         """List extraction runs for the extraction agent.
 
         Returns:
-            PaginatedExtractRunsResponse: Paginated list of extraction runs
+            List[ExtractRun]: List of extraction runs
         """
         return self._run_in_thread(
             self._client.llama_extract.list_extract_runs(
                 extraction_agent_id=self.id,
-                skip=page * limit,
-                limit=limit,
             )
         )
 
@@ -446,12 +421,6 @@ class LlamaExtract(BaseComponent):
     verbose: bool = Field(
         default=False, description="Show verbose output when extracting files."
     )
-    verify: Optional[bool] = Field(
-        default=True, description="Simple SSL verification option."
-    )
-    httpx_timeout: Optional[float] = Field(
-        default=60, description="Timeout for the httpx client."
-    )
     _async_client: AsyncLlamaCloud = PrivateAttr()
     _thread_pool: ThreadPoolExecutor = PrivateAttr()
     _project_id: Optional[str] = PrivateAttr()
@@ -467,8 +436,6 @@ class LlamaExtract(BaseComponent):
         show_progress: bool = True,
         project_id: Optional[str] = None,
         organization_id: Optional[str] = None,
-        verify: Optional[bool] = True,
-        httpx_timeout: Optional[float] = 60,
         verbose: bool = False,
     ):
         if not api_key:
@@ -486,18 +453,11 @@ class LlamaExtract(BaseComponent):
             max_timeout=max_timeout,
             num_workers=num_workers,
             show_progress=show_progress,
-            verify=verify,
-            httpx_timeout=httpx_timeout,
             verbose=verbose,
         )
-        self._httpx_client = httpx.AsyncClient(verify=verify, timeout=httpx_timeout)
-        self.verify = verify
-        self.httpx_timeout = httpx_timeout
 
         self._async_client = AsyncLlamaCloud(
-            token=self.api_key,
-            base_url=self.base_url,
-            httpx_client=self._httpx_client,
+            token=self.api_key, base_url=self.base_url, timeout=None
         )
         self._thread_pool = ThreadPoolExecutor(
             max_workers=min(10, (os.cpu_count() or 1) + 4)
@@ -526,22 +486,17 @@ class LlamaExtract(BaseComponent):
         def run_coro() -> T:
             # Create a new client for this thread
             async def wrapped_coro() -> T:
-                assert (
-                    self._httpx_client is not None
-                ), "httpx_client should be initialized"
-                # Create a new client with the same configuration as the original
                 async with httpx.AsyncClient(
-                    verify=self.verify,
-                    timeout=self.httpx_timeout,
+                    timeout=self._async_client._client_wrapper.httpx_client.timeout,
                 ) as client:
-                    # Temporarily replace the client
+                    # Replace the client in the coro's context
+                    original_client = self._async_client._client_wrapper.httpx_client
                     self._async_client._client_wrapper.httpx_client = client
                     try:
                         return await coro
                     finally:
-                        # Restore the original client
                         self._async_client._client_wrapper.httpx_client = (
-                            self._httpx_client
+                            original_client
                         )
 
             return asyncio.run(wrapped_coro())
@@ -644,8 +599,6 @@ class LlamaExtract(BaseComponent):
             num_workers=self.num_workers,
             show_progress=self.show_progress,
             verbose=self.verbose,
-            verify=self.verify,
-            httpx_timeout=self.httpx_timeout,
         )
 
     def list_agents(self) -> List[ExtractionAgent]:

llama_cloud_services/parse/base.py

@@ -4,6 +4,7 @@ import os
 import time
 from contextlib import asynccontextmanager
 from copy import deepcopy
+from enum import Enum
 from io import BufferedIOBase
 from pathlib import Path, PurePath, PurePosixPath
 from typing import Any, AsyncGenerator, Dict, List, Optional, Union
@@ -52,6 +53,14 @@ def build_url(
     return base_url
 
 
+class BackoffPattern(str, Enum):
+    """Backoff pattern for polling."""
+
+    CONSTANT = "constant"
+    LINEAR = "linear"
+    EXPONENTIAL = "exponential"
+
+
 class LlamaParse(BasePydanticReader):
     """A smart-parser for files."""
 
@@ -78,6 +87,15 @@ class LlamaParse(BasePydanticReader):
         description="The interval in seconds to check if the parsing is done.",
     )
 
+    backoff_pattern: BackoffPattern = Field(
+        default=BackoffPattern.LINEAR,
+        description="Controls the backoff pattern when retrying failed requests: 'constant', 'linear', or 'exponential'.",
+    )
+    max_check_interval: int = Field(
+        default=5,
+        description="Maximum interval in seconds between polling attempts when checking job status.",
+    )
+
     custom_client: Optional[httpx.AsyncClient] = Field(
         default=None, description="A custom HTTPX client to use for sending requests."
     )
@@ -276,7 +294,7 @@ class LlamaParse(BasePydanticReader):
         default=None,
         description="A templated suffix to add to the beginning of each page. If it contain `{page_number}`, it will be replaced by the page number.",
     )
-    parse_mode: Optional[str] = Field(
+    parsing_mode: Optional[str] = Field(
         default=None,
         description="The parsing mode to use, see ParsingMode enum for possible values ",
     )
@@ -284,10 +302,6 @@ class LlamaParse(BasePydanticReader):
         default=False,
         description="Use our best parser mode if set to True.",
     )
-    preserve_layout_alignment_across_pages: Optional[bool] = Field(
-        default=False,
-        description="Preserve grid alignment across page in text mode.",
-    )
     skip_diagonal_text: Optional[bool] = Field(
         default=False,
         description="If set to true, the parser will ignore diagonal text (when the text rotation in degrees modulo 90 is not 0).",
@@ -710,17 +724,9 @@ class LlamaParse(BasePydanticReader):
             )
             data["parsing_instruction"] = self.parsing_instruction
 
-        if self.parse_mode:
-            data["parse_mode"] = self.parse_mode
-
         if self.premium_mode:
             data["premium_mode"] = self.premium_mode
 
-        if self.preserve_layout_alignment_across_pages:
-            data[
-                "preserve_layout_alignment_across_pages"
-            ] = self.preserve_layout_alignment_across_pages
-
         if self.skip_diagonal_text:
             data["skip_diagonal_text"] = self.skip_diagonal_text
 
@@ -794,54 +800,87 @@ class LlamaParse(BasePydanticReader):
             if file_handle is not None:
                 file_handle.close()
 
+    def _calculate_backoff(self, current_interval: float) -> float:
+        """Calculate the next backoff interval based on the backoff pattern.
+
+        Args:
+            current_interval: The current interval in seconds
+
+        Returns:
+            The next interval in seconds
+        """
+        if self.backoff_pattern == BackoffPattern.CONSTANT:
+            return current_interval
+        elif self.backoff_pattern == BackoffPattern.LINEAR:
+            return min(current_interval + 1, float(self.max_check_interval))
+        elif self.backoff_pattern == BackoffPattern.EXPONENTIAL:
+            return min(current_interval * 2, float(self.max_check_interval))
+        return current_interval  # Default fallback
+
     async def _get_job_result(
         self, job_id: str, result_type: str, verbose: bool = False
     ) -> Dict[str, Any]:
         start = time.time()
         tries = 0
+        error_count = 0
+        current_interval: float = float(self.check_interval)
 
         # so we're not re-setting the headers & stuff on each
         # usage... assume that there is not some other
         # coro also modifying base_url and the other client related configs.
         client = self.aclient
         while True:
-            await asyncio.sleep(self.check_interval)
-            tries += 1
-            result = await client.get(JOB_STATUS_ROUTE.format(job_id=job_id))
-            if result.status_code != 200:
+            try:
+                await asyncio.sleep(current_interval)
+                tries += 1
+                result = await client.get(JOB_STATUS_ROUTE.format(job_id=job_id))
+                result.raise_for_status()  # this raises if status is not 2xx
+                # Allowed values "PENDING", "SUCCESS", "ERROR", "CANCELED"
+                result_json = result.json()
+                status = result_json["status"]
+                if status == "SUCCESS":
+                    parsed_result = await client.get(
+                        JOB_RESULT_URL.format(job_id=job_id, result_type=result_type),
+                    )
+                    return parsed_result.json()
+                elif status == "PENDING":
+                    end = time.time()
+                    if end - start > self.max_timeout:
+                        raise Exception(f"Timeout while parsing the file: {job_id}")
+                    if verbose and tries % 10 == 0:
+                        print(".", end="", flush=True)
+                    current_interval = self._calculate_backoff(current_interval)
+                else:
+                    error_code = result_json.get("error_code", "No error code found")
+                    error_message = result_json.get(
+                        "error_message", "No error message found"
+                    )
+                    exception_str = (
+                        f"Job ID: {job_id} failed with status: {status}, "
+                        f"Error code: {error_code}, Error message: {error_message}"
+                    )
+                    raise Exception(exception_str)
+            except (
+                httpx.ConnectError,
+                httpx.ReadError,
+                httpx.WriteError,
+                httpx.ConnectTimeout,
+                httpx.ReadTimeout,
+                httpx.WriteTimeout,
+                httpx.HTTPStatusError,
+            ) as err:
+                error_count += 1
                 end = time.time()
                 if end - start > self.max_timeout:
-                    raise Exception(f"Timeout while parsing the file: {job_id}")
+                    raise Exception(
+                        f"Timeout while parsing the file: {job_id}"
+                    ) from err
                 if verbose and tries % 10 == 0:
-                    print(".", end="", flush=True)
-                await asyncio.sleep(self.check_interval)
-                continue
-
-            # Allowed values "PENDING", "SUCCESS", "ERROR", "CANCELED"
-            result_json = result.json()
-            status = result_json["status"]
-            if status == "SUCCESS":
-                parsed_result = await client.get(
-                    JOB_RESULT_URL.format(job_id=job_id, result_type=result_type),
-                )
-                return parsed_result.json()
-
-            elif status == "PENDING":
-                end = time.time()
-                if end - start > self.max_timeout:
-                    raise Exception(f"Timeout while parsing the file: {job_id}")
-                if verbose and tries % 10 == 0:
-                    print(".", end="", flush=True)
-                await asyncio.sleep(self.check_interval)
-
-            else:
-                error_code = result_json.get("error_code", "No error code found")
-                error_message = result_json.get(
-                    "error_message", "No error message found"
-                )
-
-                exception_str = f"Job ID: {job_id} failed with status: {status}, Error code: {error_code}, Error message: {error_message}"
-                raise Exception(exception_str)
+                    print(
+                        f"HTTP error: {err}...",
+                        flush=True,
+                    )
+                current_interval = self._calculate_backoff(current_interval)
 
     async def _aload_data(
         self,

llama_parse/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "llama-parse"
-version = "0.6.8"
+version = "0.6.5"
 description = "Parse files into RAG-Optimized formats."
 authors = ["Logan Markewich <logan@llamaindex.ai>"]
 license = "MIT"
@@ -13,7 +13,7 @@ packages = [{include = "llama_parse"}]
 
 [tool.poetry.dependencies]
 python = ">=3.9,<4.0"
-llama-cloud-services = ">=0.6.7"
+llama-cloud-services = ">=0.6.5"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.0.0"

pyproject.toml

@@ -8,7 +8,7 @@ python_version = "3.10"
 
 [tool.poetry]
 name = "llama-cloud-services"
-version = "0.6.8"
+version = "0.6.5"
 description = "Tailored SDK clients for LlamaCloud services."
 authors = ["Logan Markewich <logan@runllama.ai>"]
 license = "MIT"
@@ -18,7 +18,7 @@ packages = [{include = "llama_cloud_services"}]
 [tool.poetry.dependencies]
 python = ">=3.9,<4.0"
 llama-index-core = ">=0.11.0"
-llama-cloud = "^0.1.16"
+llama-cloud = "^0.1.14"
 pydantic = "!=2.10"
 click = "^8.1.7"
 python-dotenv = "^1.0.1"

tests/extract/test_extract_api.py

@@ -182,14 +182,7 @@ class TestExtractionAgent:
         assert "new_field" in updated_agent.data_schema["properties"]
 
     def test_list_extraction_runs(self, test_agent: ExtractionAgent):
-        assert test_agent.list_extraction_runs().total == 0
+        assert len(test_agent.list_extraction_runs()) == 0
         test_agent.extract(TEST_PDF)
         runs = test_agent.list_extraction_runs()
-        assert runs.total > 0
-
-    def test_delete_extraction_run(self, test_agent: ExtractionAgent):
-        assert test_agent.list_extraction_runs().total == 0
-        run = test_agent.extract(TEST_PDF)
-        test_agent.delete_extraction_run(run.id)
-        runs = test_agent.list_extraction_runs()
-        assert runs.total == 0
+        assert len(runs) > 0

tests/report/test_llama_report.py

@@ -81,11 +81,6 @@ async def test_create_and_delete_report(
 
 
 @pytest.mark.asyncio
-@pytest.mark.xfail(
-    condition=lambda: os.getenv("CI"),
-    reason="Report plan sometimes times out",
-    raises=TimeoutError,
-)
 async def test_report_plan_workflow(report: ReportClient) -> None:
     """Test the report planning workflow."""
     # Wait for the plan