Loop Tasks
Run a task once per row of a CSV or text file, in sync or async workflows
Loop tasks turn a single task into one subtask per row of a CSV or text file — no manual iteration.
Quick Start
- Loop a task over a CSV
python from praisonaiagents import Agent, Task, PraisonAIAgents researcher = Agent(name="Researcher", instructions="Summarise the topic in one line") loop_task = Task( name="research_each", description="Research: {input}", # {input} substitutes per row expected_output="One-line summary", agent=researcher, task_type="loop", input_file="topics.csv", is_start=True, ) workflow = PraisonAIAgents(agents=[researcher], tasks=[loop_task], process="workflow") workflow.start() - Same task, async entry point
python import asyncio asyncio.run(workflow.astart()) # per-row expansion works here too (SDK ≥ 2026-07-27)
How It Works
The workflow reads input_file before the run loop begins and expands the loop task into one subtask per row.
| Step | What happens |
|---|---|
| Read file | Workflow reads input_file once, before the first task runs |
| Expand | Each row becomes a subtask named <original_name>_<row_index> |
| Run | Subtasks execute sequentially, each seeing its row as {input} |
| Aggregate | Result returned when all subtasks complete |
Choose the Right Loop
| Use | When | Import |
|---|---|---|
Task(task_type="loop", input_file=...) | Fixed CSV/text file, workflow process, one row per subtask | from praisonaiagents import Task, PraisonAIAgents |
AgentFlow(steps=[loop(from_csv=...)]) | Deterministic step pipeline with loops as a step | from praisonaiagents import AgentFlow, loop |
AgentFlow(steps=[repeat(step, until=...)]) | Repeat a step until a condition is met (no file needed) | from praisonaiagents import AgentFlow, repeat |
Sync vs Async
Common Patterns
Summarise a list of URLs
from praisonaiagents import Agent, Task, PraisonAIAgents
summariser = Agent(name="Summariser", instructions="Summarise the page at the given URL in one line")
loop_task = Task(
name="summarise_url",
description="Summarise this URL: {input}",
expected_output="One-line summary",
agent=summariser,
task_type="loop",
input_file="urls.csv",
is_start=True,
)
workflow = PraisonAIAgents(agents=[summariser], tasks=[loop_task], process="workflow")
workflow.start()Loop then post-process each row
from praisonaiagents import Agent, Task, PraisonAIAgents
extractor = Agent(name="Extractor", instructions="Extract the key fact from the row")
formatter = Agent(name="Formatter", instructions="Format the fact as a bullet point")
extract = Task(
name="extract_row",
description="Extract the key fact: {input}",
expected_output="A single fact",
agent=extractor,
task_type="loop",
input_file="scraped.csv",
is_start=True,
next_tasks=["format_row"],
)
format_row = Task(
name="format_row",
description="Format as a bullet point",
expected_output="A markdown bullet",
agent=formatter,
)
workflow = PraisonAIAgents(agents=[extractor, formatter], tasks=[extract, format_row], process="workflow")
workflow.start()Best Practices
Omit input_file to default to tasks.csv
When task_type="loop" and no input_file is set, the workflow reads tasks.csv from the working directory. This applies to both the start loop task and loop tasks reached mid-workflow via next_tasks / condition routing — in sync (workflow.start()) and async (workflow.astart()) engines equally. See the Sync vs Async note above for the version history.
Task(name="loop_it", description="Process: {input}", agent=agent, task_type="loop", is_start=True)
# reads tasks.csv (start task)
Task(name="post_process", description="Handle: {input}", agent=agent, task_type="loop")
# also reads tasks.csv when reached via next_tasksOne row = one string — use JSON for structured fields
Each row is passed to {input} as a single string. Put structured data in JSON inside the row and parse it in the agent instructions.
Loop subtasks run sequentially by design
Subtasks execute one after another. Use AgentFlow.parallel when you want fan-out across rows.
A single failing row doesn't abort the workflow
Other rows still run. Check per-subtask TaskOutput to find which rows failed.