Add Workflow.outputs (#1208)

CHANGELOG.md

@@ -5,14 +5,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
-### Added
-- `Workflow.input_tasks` and `Workflow.output_tasks` to access the input and output tasks of a Workflow.
-- Ability to pass nested list of `Tasks` to `Structure.tasks` allowing for more complex declarative Structure definitions.
-
 
 ## Added
+- `Workflow.input_tasks` and `Workflow.output_tasks` to access the input and output tasks of a Workflow.
+- Ability to pass nested list of `Tasks` to `Structure.tasks` allowing for more complex declarative Structure definitions.
 - Parameter `pipeline_task` on `HuggingFacePipelinePromptDriver` for creating different types of `Pipeline`s.
 - `TavilyWebSearchDriver` to integrate Tavily's web search capabilities.
+- `Workflow.outputs` to access the outputs of a Workflow.
 
 ### Changed
 - **BREAKING**: Renamed parameters on several classes to `client`:

griptape/structures/workflow.py

@@ -12,6 +12,7 @@
 from griptape.structures import Structure
 
 if TYPE_CHECKING:
+    from griptape.artifacts import BaseArtifact
     from griptape.tasks import BaseTask
 
 
@@ -33,6 +34,10 @@ def input_tasks(self) -> list[BaseTask]:
     def output_tasks(self) -> list[BaseTask]:
         return [task for task in self.tasks if not task.children]
 
+    @property
+    def outputs(self) -> list[BaseArtifact]:
+        return [task.output.value for task in self.output_tasks if task.output is not None]
+
     def add_task(self, task: BaseTask) -> BaseTask:
         if (existing_task := self.try_find_task(task.id)) is not None:
             return existing_task

tests/unit/structures/test_workflow.py

@@ -826,6 +826,15 @@ def test_input_tasks(self):
 
         assert workflow.input_tasks == [parent]
 
+    def test_outputs(self):
+        workflow = Workflow(tasks=[PromptTask("parent") for _ in range(3)])
+
+        assert workflow.outputs == []
+
+        workflow.run()
+
+        assert workflow.outputs == ["mock output"] * 3
+
     @staticmethod
     def _validate_topology_1(workflow) -> None:
         assert len(workflow.tasks) == 4