fix: enable parallel execution in graph workflow (#485)

Author: ahmetatalay

src/strands/multiagent/graph.py

@@ -338,13 +338,18 @@ async def _execute_graph(self) -> None:
             current_batch = ready_nodes.copy()
             ready_nodes.clear()
 
-            # Execute current batch of ready nodes
-            for node in current_batch:
-                if node not in self.state.completed_nodes:
-                    await self._execute_node(node)
-
-                    # Find newly ready nodes after this execution
-                    ready_nodes.extend(self._find_newly_ready_nodes())
+            # Execute current batch of ready nodes concurrently
+            tasks = [
+                asyncio.create_task(self._execute_node(node))
+                for node in current_batch
+                if node not in self.state.completed_nodes
+            ]
+
+            for task in tasks:
+                await task
+
+            # Find newly ready nodes after batch execution
+            ready_nodes.extend(self._find_newly_ready_nodes())
 
     def _find_newly_ready_nodes(self) -> list["GraphNode"]:
         """Find nodes that became ready after the last execution."""

tests_integ/test_multiagent_graph.py

@@ -131,7 +131,10 @@ def proceed_to_second_summary(state):
 
     # Verify execution order - extract node_ids from GraphNode objects
     execution_order_ids = [node.node_id for node in result.execution_order]
-    assert execution_order_ids == ["computation_subgraph", "secondary_math", "validator", "primary_summary"]
+    # With parallel execution, secondary_math and validator can complete in any order
+    assert execution_order_ids[0] == "computation_subgraph"  # First
+    assert execution_order_ids[3] == "primary_summary"  # Last
+    assert set(execution_order_ids[1:3]) == {"secondary_math", "validator"}  # Middle two in any order
 
     # Verify specific nodes completed
     assert "computation_subgraph" in result.results