community[patch]: Adding try-except block for GCSDirectoryLoader (lan…

…gchain-ai#19591)

- **Description:** Implemented try-except block for
`GCSDirectoryLoader`. Reason: Users processing large number of
unstructured files in a folder may experience many different errors. A
try-exception block is added to capture these errors. A new argument
`use_try_except=True` is added to enable *silent failure* so that error
caused by processing one file does not break the whole function.
- **Issue:** N/A
- **Dependencies:** no new dependencies
- **Twitter handle:** timothywong731

---------

Co-authored-by: Bagatur <[email protected]>

docs/docs/integrations/document_loaders/google_cloud_storage_directory.ipynb

@@ -125,18 +125,41 @@
     "loader.load()"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "id": "f9c0734f",
+   "metadata": {},
+   "source": [
+    "## Continue on failure to load a single file\n",
+    "Files in a GCS bucket may cause errors during processing. Enable the `continue_on_failure=True` argument to allow silent failure. This means failure to process a single file will not break the function, it will log a warning instead. "
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "f9c0734f",
+   "id": "3d774795",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "loader = GCSDirectoryLoader(\n",
+    "    project_name=\"aist\", bucket=\"testing-hwc\", continue_on_failure=True\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "0d15f536",
    "metadata": {},
    "outputs": [],
-   "source": []
+   "source": [
+    "loader.load()"
+   ]
   }
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "Python 3 (ipykernel)",
+   "display_name": "Python 3.10.6 64-bit",
    "language": "python",
    "name": "python3"
   },
@@ -151,6 +174,11 @@
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
    "version": "3.10.6"
+  },
+  "vscode": {
+   "interpreter": {
+    "hash": "5f90d085fc70553c85f15dd96b84c64a94d58988a621c9dbc38cac6a7e6079b3"
+   }
   }
  },
  "nbformat": 4,

libs/community/langchain_community/document_loaders/gcs_directory.py

@@ -1,3 +1,4 @@
+import logging
 from typing import Callable, List, Optional
 
 from langchain_core.documents import Document
@@ -6,6 +7,8 @@
 from langchain_community.document_loaders.gcs_file import GCSFileLoader
 from langchain_community.utilities.vertexai import get_client_info
 
+logger = logging.getLogger(__name__)
+
 
 class GCSDirectoryLoader(BaseLoader):
     """Load from GCS directory."""
@@ -16,6 +19,7 @@ def __init__(
         bucket: str,
         prefix: str = "",
         loader_func: Optional[Callable[[str], BaseLoader]] = None,
+        continue_on_failure: bool = False,
     ):
         """Initialize with bucket and key name.
 
@@ -26,11 +30,15 @@ def __init__(
             loader_func: A loader function that instantiates a loader based on a
                 file_path argument. If nothing is provided, the  GCSFileLoader
                 would use its default loader.
+            continue_on_failure: To use try-except block for each file within the GCS
+                directory. If set to `True`, then failure to process a file will not
+                cause an error.
         """
         self.project_name = project_name
         self.bucket = bucket
         self.prefix = prefix
         self._loader_func = loader_func
+        self.continue_on_failure = continue_on_failure
 
     def load(self) -> List[Document]:
         """Load documents."""
@@ -55,4 +63,19 @@ def load(self) -> List[Document]:
                 self.project_name, self.bucket, blob.name, loader_func=self._loader_func
             )
             docs.extend(loader.load())
+            # Use the try-except block here
+            try:
+                loader = GCSFileLoader(
+                    self.project_name,
+                    self.bucket,
+                    blob.name,
+                    loader_func=self._loader_func,
+                )
+                docs.extend(loader.load())
+            except Exception as e:
+                if self.continue_on_failure:
+                    logger.warning(f"Problem processing blob {blob.name}, message: {e}")
+                    continue
+                else:
+                    raise e
         return docs