Repetitive Agents

Handle batch and loop tasks with AgentFlow and the loop() helper

Run the same agent step over a list, CSV, or text file — then aggregate results in one workflow.

from praisonaiagents import Agent, AgentFlow, loop

processor = Agent(
    name="Processor",
    instructions="Process each item thoroughly.",
)

workflow = AgentFlow(
    steps=[
        loop(processor, over="topics"),
        Agent(name="Summariser", instructions="Summarise all processed results."),
    ],
    variables={"topics": ["AI Ethics", "Machine Learning", "Neural Networks"]},
)

result = workflow.start("Research and summarise these AI topics")

The user supplies a list of items; the workflow loops the agent over each entry then summarises results.

Quick Start

1. Simple Usage

   Loop an agent over a variable list:

   from praisonaiagents import Agent, AgentFlow, loop
   
   processor = Agent(name="Processor", instructions="Process each item thoroughly.")
   
   workflow = AgentFlow(
       steps=[
           loop(processor, over="topics"),
           Agent(name="Summariser", instructions="Summarise all processed results."),
       ],
       variables={"topics": ["AI Ethics", "Machine Learning", "Neural Networks"]},
   )
   
   result = workflow.start("Research and summarise these AI topics")
   print(result["output"][:500])
   
2. With Configuration

   Process rows from a CSV file with a custom handler:

   from praisonaiagents import Agent, AgentFlow, WorkflowContext, StepResult, loop
   
   support_agent = Agent(
       name="Support Agent",
       instructions="Resolve customer issues efficiently.",
       llm="gpt-4o-mini",
   )
   
   def handle_customer(ctx: WorkflowContext) -> StepResult:
       row = ctx.variables.get("item", {})
       name = row.get("name", "unknown")
       issue = row.get("issue", "unknown")
       response = support_agent.chat(f"Help {name} with: {issue}")
       return StepResult(output=f"{name}: {response}")
   
   workflow = AgentFlow(steps=[loop(handle_customer, from_csv="customers.csv")])
   result = workflow.start("Process all customer issues")
   

---

How It Works

Input source	Parameter	Behaviour
Variable list	over="topics"	Iterates variables["topics"]
CSV file	from_csv="data.csv"	One sub-step per row
Text file	from_file="urls.txt"	One sub-step per line
Parallel	parallel=True	Runs iterations concurrently

Loop outputs are stored in result["variables"]["loop_outputs"] by default.

---

Common Patterns

CSV customer support

name,issue
John,Billing problem
Jane,Technical issue
Sarah,Password reset

workflow = AgentFlow(steps=[loop(handle_customer, from_csv="customers.csv")])

Line-by-line URL analysis

def analyse_url(ctx: WorkflowContext) -> StepResult:
    url = ctx.variables.get("item", "")
    result = url_agent.chat(f"Analyse this website: {url}")
    return StepResult(output=f"{url}: {result}")

workflow = AgentFlow(steps=[loop(analyse_url, from_file="urls.txt")])

Parallel batch processing

workflow = AgentFlow(
    steps=[loop(process_item, from_csv="data.csv", parallel=True, max_workers=4)]
)

---

Best Practices

Validate input files before starting

Check that CSV files exist and have headers. Empty files produce zero iterations silently.

Use parallel=True for independent items

When sub-steps do not depend on each other, set parallel=True and cap workers with max_workers.

Aggregate with a final summariser step

Add a summariser agent after the loop step rather than embedding aggregation inside each iteration.

Keep loop handlers focused

Pass row data via ctx.variables["item"] and return a StepResult — avoid side effects outside the workflow context.

---

Related

Workflow Loop

Loop patterns inside multi-step workflows

Workflow Parallel Execution

Run independent workflow branches concurrently