> ## Documentation Index
> Fetch the complete documentation index at: https://developer.watson-orchestrate.ibm.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Foreach node
Use a foreach node to create a sub agentic workflow that repeats nodes for each item in the input schema.
To configure a foreach node in an agentic workflow, call the `foreach()` method to define a nested agentic workflow. In this method, define the following parameters:
The schema that defines each item to iterate over.
The input schema for the nested agentic workflow executed by the foreach node.
The output schema for the nested agentic workflow executed by the foreach node.
```py Example theme={null}
foreach_flow: Flow = aflow.foreach(item_schema = CustomerRecord)
```
Next, call the foreach agentic workflow for the node you want to repeat. You can include more than one node in the same foreach flow.
```py Example theme={null}
node2 = foreach_flow.tool(send_invitation_email)
foreach_flow.sequence(START, node2, END)
```
The example below shows how to configure a foreach node:
```py Python [expandable] theme={null}
"""
In this example, the user is retrieving a set of email addresses from a contact list, and
for each email address, sending out an invitation.
"""
from typing import List
from pydantic import BaseModel, Field
from ibm_watsonx_orchestrate.agent_builder.tools import tool, ToolPermission
from ibm_watsonx_orchestrate.flow_builder.flows import Flow, flow, START, END
from ibm_watsonx_orchestrate.flow_builder.types import ForeachPolicy
from .send_invitation_email import send_invitation_email
from .get_emails_from_customer import get_emails_from_customer, CustomerRecord
class CustomerName(BaseModel):
name: str = Field(description="The name of the customer")
class Invitations(BaseModel):
invitations: List[str]
@flow(
name="send_invitation_to_customer",
input_schema=CustomerName
)
def build_send_invitation_to_customer_flow(aflow: Flow) -> Flow:
""" Given a list of customers, we will iterate through the list and send email to each """
get_customer_list_node = aflow.tool(get_emails_from_customer)
# calling add_foreach will create a subflow, and we can add more node to the subflow
foreach_flow: Flow = aflow.foreach(item_schema = CustomerRecord) \
.policy(kind=ForeachPolicy.SEQUENTIAL)
# .policy(kind=ForeachPolicy.PARALLEL) # replace with 'ForeachPolicy.PARALLEL' if need to run .foreach in parallel
node2 = foreach_flow.tool(send_invitation_email)
foreach_flow.sequence(START, node2, END)
# add the foreach flow to the main flow
aflow.edge(START, get_customer_list_node)
aflow.edge(get_customer_list_node, foreach_flow)
aflow.edge(foreach_flow, END)
return aflow
```
### Foreach processing method
You can configure how your foreach node processes input items. There are two processing methods:
* **Sequential**
A sequential foreach processes items one after another. The next item starts only after the previous one finishes. Use this method when the order of item processing affects the result.
* **Parallel**
A parallel foreach processes one or more items at the same time. In most cases, this method runs faster than the sequential approach.
To set the processing method, call the `policy()` method in your foreach node. This method accepts the following parameter:
Defines the processing method. Use `ForeachPolicy.SEQUENTIAL` for sequential processing or `ForeachPolicy.PARALLEL` for parallel processing.
```py Python {7-9} theme={null}
def build_send_invitation_to_customer_flow(aflow: Flow) -> Flow:
""" Given a list of customers, we will iterate through the list and send email to each """
get_customer_list_node = aflow.tool(get_emails_from_customer)
# calling add_foreach will create a subflow, and we can add more node to the subflow
foreach_flow: Flow = aflow.foreach(item_schema = CustomerRecord) \
.policy(kind=ForeachPolicy.SEQUENTIAL)
# .policy(kind=ForeachPolicy.PARALLEL) # replace with 'ForeachPolicy.PARALLEL' if need to run .foreach in parallel
```