Merge branch 'main' into chore/main

CHANGELOG.md

@@ -12,6 +12,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 - Parsing streaming response with some OpenAi compatible services.
 
+## [0.30.2] - 2024-08-26
+
+### Fixed
+- Ensure thread safety when publishing events by adding a thread lock to batch operations in `BaseEventListenerDriver`. 
+- `FileManagerTool` failing to save Artifacts created by `ExtractionTool` with a `CsvExtractionEngine`.
+
 ## [0.30.1] - 2024-08-21
 
 ### Fixed

griptape/drivers/event_listener/base_event_listener_driver.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import threading
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING
 
@@ -18,6 +19,7 @@
 class BaseEventListenerDriver(FuturesExecutorMixin, ABC):
     batched: bool = field(default=True, kw_only=True)
     batch_size: int = field(default=10, kw_only=True)
+    thread_lock: threading.Lock = field(default=Factory(lambda: threading.Lock()))
 
     _batch: list[dict] = field(default=Factory(list), kw_only=True)
 
@@ -39,10 +41,11 @@ def _safe_try_publish_event(self, event: BaseEvent | dict, *, flush: bool) -> No
             event_payload = event if isinstance(event, dict) else event.to_dict()
 
             if self.batched:
-                self._batch.append(event_payload)
-                if len(self.batch) >= self.batch_size or flush:
-                    self.try_publish_event_payload_batch(self.batch)
-                    self._batch = []
+                with self.thread_lock:
+                    self._batch.append(event_payload)
+                    if len(self.batch) >= self.batch_size or flush:
+                        self.try_publish_event_payload_batch(self.batch)
+                        self._batch = []
                 return
             else:
                 self.try_publish_event_payload(event_payload)

griptape/tools/file_manager/tool.py

@@ -93,7 +93,8 @@ def save_memory_artifacts_to_disk(self, params: dict) -> ErrorArtifact | InfoArt
 
         for artifact in list_artifact.value:
             formatted_file_name = f"{artifact.name}-{file_name}" if len(list_artifact) > 1 else file_name
-            result = self.file_manager_driver.save_file(os.path.join(dir_name, formatted_file_name), artifact.value)
+            value = artifact.value if isinstance(artifact.value, (str, bytes)) else artifact.to_text()
+            result = self.file_manager_driver.save_file(os.path.join(dir_name, formatted_file_name), value)
             if isinstance(result, ErrorArtifact):
                 return result
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "griptape"
-version = "0.30.1"
+version = "0.30.2"
 description = "Modular Python framework for LLM workflows, tools, memory, and data."
 authors = ["Griptape <[email protected]>"]
 license = "Apache 2.0"

tests/unit/tools/test_file_manager.py

@@ -6,6 +6,7 @@
 import pytest
 
 from griptape.artifacts import ListArtifact, TextArtifact
+from griptape.artifacts.csv_row_artifact import CsvRowArtifact
 from griptape.artifacts.error_artifact import ErrorArtifact
 from griptape.drivers.file_manager.local_file_manager_driver import LocalFileManagerDriver
 from griptape.loaders.text_loader import TextLoader
@@ -108,6 +109,29 @@ def test_save_memory_artifacts_to_disk_for_multiple_artifacts(self, temp_dir):
         assert Path(os.path.join(temp_dir, "test", f"{artifacts[1].name}-{file_name}")).read_text() == "baz"
         assert result.value == "Successfully saved memory artifacts to disk"
 
+    def test_save_memory_artifacts_to_disk_for_non_string_artifact(self, temp_dir):
+        memory = defaults.text_task_memory("Memory1")
+        artifact = CsvRowArtifact({"foo": "bar"})
+
+        memory.store_artifact("foobar", artifact)
+
+        file_manager = FileManagerTool(
+            input_memory=[memory], file_manager_driver=LocalFileManagerDriver(workdir=temp_dir)
+        )
+        result = file_manager.save_memory_artifacts_to_disk(
+            {
+                "values": {
+                    "dir_name": "test",
+                    "file_name": "foobar.txt",
+                    "memory_name": memory.name,
+                    "artifact_namespace": "foobar",
+                }
+            }
+        )
+
+        assert Path(os.path.join(temp_dir, "test", "foobar.txt")).read_text() == "bar"
+        assert result.value == "Successfully saved memory artifacts to disk"
+
     def test_save_content_to_file(self, temp_dir):
         file_manager = FileManagerTool(file_manager_driver=LocalFileManagerDriver(workdir=temp_dir))
         result = file_manager.save_content_to_file(