Map Node | Vellum | Documentation

1 from vellum.workflows.inputs.base import BaseInputs 2 from vellum.workflows.state import BaseState 3 from vellum.workflows import BaseWorkflow 4 from vellum.workflows.nodes import MapNode, BaseNode 5 from typing import List 6 7 class MapNodeInputs(BaseInputs): 8 my_list: List[str] 9 10 class Iteration(BaseNode): 11 item = MapNode.SubworkflowInputs.item 12 index = MapNode.SubworkflowInputs.index 13 14 class Outputs(BaseNode.Outputs): 15 result: str 16 17 def run(self) -> Outputs: 18 return self.Outputs(result=self.item + str(self.index)) 19 20 class IterationSubworkflow(BaseWorkflow[MapNode.SubworkflowInputs, BaseState]): 21 graph = Iteration 22 23 class Outputs(BaseWorkflow.Outputs): 24 result = Iteration.Outputs.result 25 26 class MyMapNode(MapNode): 27 items = MapNodeInputs.my_list 28 subworkflow = IterationSubworkflow 29 30 class Outputs(BaseNode.Outputs): 31 result: List[str] # this MUST match the Subworkflow's Outputs.result attribute

1 MyMapNode.Outputs( 2 result=[ 3 "text0", 4 "text1", 5 "text2" 6 ] 7 )

vellum.workflows.nodes.MapNode

Used to map over a list of items and execute a Subworkflow for each item. This enables parallel processing of multiple items through the same workflow logic.

Attributes

items

List[Any]Required

The list of items to map over. Each item will be processed by the subworkflow.

subworkflow

Type[BaseWorkflow[WorkflowInputsType, InnerStateType]]Required

The Subworkflow class to execute for each item

max_concurrency

Optional[int]Defaults to None

The maximum number of concurrent subworkflow executions.

Outputs

The outputs are determined by the subworkflow’s outputs, with each output field becoming a list containing results from all iterations.