Map Node | Vellum | Documentation

from vellum.workflows.inputs.base import BaseInputs
from vellum.workflows.state import BaseState
from vellum.workflows import BaseWorkflow
from vellum.workflows.nodes import MapNode, BaseNode
from typing import List
class MapNodeInputs(BaseInputs):
    my_list: List[str]
class Iteration(BaseNode):
    item = MapNode.SubworkflowInputs.item
    index = MapNode.SubworkflowInputs.index
    class Outputs(BaseNode.Outputs):
        result: str
    def run(self) -> Outputs:
        return self.Outputs(result=self.item + str(self.index))
    
class IterationSubworkflow(BaseWorkflow[MapNode.SubworkflowInputs, BaseState]):
    graph = Iteration
    class Outputs(BaseWorkflow.Outputs):
        result = Iteration.Outputs.result
class MyMapNode(MapNode):
    items = MapNodeInputs.my_list
    subworkflow = IterationSubworkflow
    class Outputs(BaseNode.Outputs):
        result: List[str] # this MUST match the Subworkflow's Outputs.result attribute

MyMapNode.Outputs(
    result=[
        "text0",
        "text1",
        "text2"
    ]
)

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.

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