Skip to main content
The loop() helper runs a step for each item in a list, CSV, or text file — ideal for batch processing and data pipelines. 
from praisonaiagents import Agent, AgentFlow, loop, WorkflowContext, StepResult

worker = Agent(name="Worker", instructions="Process one batch item at a time")

def process_item(ctx: WorkflowContext) -> StepResult:
    item = ctx.variables["item"]
    return StepResult(output=f"Processed: {item}")

workflow = AgentFlow(
    agents=[worker],
    steps=[loop(process_item, over="items")],
    variables={"items": ["alpha", "beta"]},
)
workflow.start("Run the batch loop")
The user triggers a batch job; each list item runs through the loop step.

​
Quick Start

1

Loop Over a List

from praisonaiagents import AgentFlow, WorkflowContext, StepResult
from praisonaiagents import loop

def process_item(ctx: WorkflowContext) -> StepResult:
    item = ctx.variables["item"]
    index = ctx.variables["loop_index"]
    return StepResult(output=f"[{index}] Processed: {item}")

workflow = AgentFlow(
    steps=[loop(process_item, over="items")],
    variables={"items": ["apple", "banana", "cherry"]}
)

result = workflow.start("Process fruits")
Output:
🔁 Looping over 3 items...
✅ process_item: [0] Processed: apple...
✅ process_item: [1] Processed: banana...
✅ process_item: [2] Processed: cherry...
2

Loop Over a CSV

def process_row(ctx: WorkflowContext) -> StepResult:
    row = ctx.variables["item"]  # Dict with column names as keys
    name = row["name"]
    email = row["email"]
    return StepResult(output=f"Processed: {name} ({email})")

workflow = AgentFlow(steps=[
    loop(process_row, from_csv="users.csv")
])

result = workflow.start("Process users")
3

Loop Over a Text File

def process_line(ctx: WorkflowContext) -> StepResult:
    line = ctx.variables["item"]  # String
    return StepResult(output=f"Line: {line}")

workflow = AgentFlow(steps=[
    loop(process_line, from_file="tasks.txt")
])

​
Choosing an Iteration Source

Pick the parameter that matches where your items live.

​
API Reference

​
loop()

loop(
    step: Any,                       # Step to execute for each item
    over: Optional[str] = None,      # Variable name containing list
    from_csv: Optional[str] = None,  # CSV file path
    from_file: Optional[str] = None, # Text file path (one item per line)
    var_name: str = "item"           # Variable name for current item
) -> Loop

​
Parameters

ParameterTypeDefaultDescription
stepAnyrequiredStep to execute (function, Agent, Task)
overOptional[str]NoneVariable name containing list to iterate
from_csvOptional[str]NonePath to CSV file
from_fileOptional[str]NonePath to text file
var_namestr"item"Variable name for current item

​
Context Variables

Inside the loop, these variables are available:
VariableTypeDescription
item (or custom)AnyCurrent item being processed
loop_indexintCurrent iteration index (0-based)

​
Result Variables

After loop completion:
VariableTypeDescription
loop_outputsList[str]All outputs from loop iterations

​
Examples

​
Custom Variable Name

workflow = AgentFlow(
    steps=[loop(process_user, over="users", var_name="user")],
    variables={"users": [{"name": "Alice"}, {"name": "Bob"}]}
)

def process_user(ctx: WorkflowContext) -> StepResult:
    user = ctx.variables["user"]  # Custom variable name
    return StepResult(output=f"Hello, {user['name']}!")

​
CSV with Headers

Given data.csv: 
name,email,role
Alice,alice@example.com,Admin
Bob,bob@example.com,User

def process_csv_row(ctx: WorkflowContext) -> StepResult:
    row = ctx.variables["item"]
    # row = {"name": "Alice", "email": "alice@example.com", "role": "Admin"}
    return StepResult(output=f"{row['name']} is {row['role']}")

workflow = AgentFlow(steps=[
    loop(process_csv_row, from_csv="data.csv")
])

​
With Agents

from praisonaiagents import Agent

processor = Agent(
    name="Processor",
    role="Process items",
    instructions="Process the item: {{item}}"
)

workflow = AgentFlow(
    steps=[loop(processor, over="items")],
    variables={"items": ["task1", "task2", "task3"]}
)

​
Chained Loops

workflow = AgentFlow(steps=[
    loop(fetch_data, over="sources"),      # Fetch from each source
    loop(process_record, over="records"),  # Process each record
    summarize_all
])

​
Loop with Aggregation

def aggregate_results(ctx: WorkflowContext) -> StepResult:
    outputs = ctx.variables["loop_outputs"]
    total = len(outputs)
    return StepResult(output=f"Processed {total} items")

workflow = AgentFlow(steps=[
    loop(process_item, over="items"),
    aggregate_results
])

​
Use Cases

Use CaseDescription
Batch ProcessingProcess files, records, or data items
Data MigrationTransform and migrate data row by row
Report GenerationGenerate reports for each entity
Email CampaignsSend personalized emails to list
API CallsCall API for each item in list

​
How It Works

PhaseWhat happens
1. Resolveloop() reads the over variable (list, CSV, or text file) from ctx.variables
2. IterateHandler is called once per item; ctx.variables["item"] holds the current value
3. CollectAll StepResult objects are gathered; loop_index tracks the current position
4. ReturnAgentFlow returns after the last item completes

​
Best Practices

Catch exceptions inside the loop body so one bad row does not stop the entire batch.
Enable verbose mode or custom logging when iterating thousands of records.
Split massive CSVs into batches to limit memory and simplify retries.
Close files and connections even when individual items fail.

​
Error Handling

def safe_process(ctx: WorkflowContext) -> StepResult:
    try:
        item = ctx.variables["item"]
        # Process item...
        return StepResult(output=f"Success: {item}")
    except Exception as e:
        return StepResult(output=f"Error: {e}")

Workflow Patterns

Overview of routing, parallel, loop, and repeat patterns

Workflow Routing

Decision-based branching in workflows

Workflow Parallel

Run independent steps concurrently

Workflow Repeat

Repeat steps until a condition is met