A workflow optimization pattern where agents handle repetitive tasks through automated loops, processing multiple instances efficiently while maintaining consistency.

Quick Start

1

Install Package

First, install the PraisonAI Agents package:
pip install praisonaiagents
2

Set API Key

Set your OpenAI API key as an environment variable in your terminal:
export OPENAI_API_KEY=your_api_key_here
3

Create a file

Create a new file repetitive_agent.py with the basic setup:
from praisonaiagents import Agent, Task, PraisonAIAgents

agent = Agent(
    instructions="You are a loop agent that creating a loop of tasks."
)

task = Task(
    description="Create the list of tasks to be looped through.",
    agent=agent,
    task_type="loop",
    input_file="tasks.csv"
)

agents = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    process="workflow",
    max_iter=30
)

agents.start()
4

Start Agents

Type this in your terminal to run your agents:
python repetitive_agent.py
Requirements
  • Python 3.10 or higher
  • OpenAI API key. Generate OpenAI API key here. Use Other models using this guide.

Understanding Repetitive Agents

What are Repetitive Agents?

Repetitive agents enable:
  • Automated task loops
  • Batch processing
  • Consistent task execution
  • Efficient handling of multiple similar tasks

Features

Task Looping

Process multiple tasks through automated loops.

Batch Processing

Handle multiple similar tasks efficiently.

Input Management

Process tasks from structured input files.

Progress Tracking

Monitor task completion and progress.

Loop Tasks with File Input

Process batches of tasks from CSV or other structured files:
from praisonaiagents import Agent, Task, PraisonAIAgents

# Create agent for processing questions
qa_agent = Agent(
    name="QA Bot",
    role="Answer questions",
    goal="Provide accurate answers to user questions"
)

# Create loop task that processes questions from CSV
loop_task = Task(
    name="process_questions",
    description="Answer each question",
    expected_output="Answer for each question",
    agent=qa_agent,
    task_type="loop",
    input_file="questions.csv"  # Each row becomes a subtask
)

# Create workflow
agents = PraisonAIAgents(
    agents=[qa_agent],
    tasks=[loop_task],
    process="workflow"  # Use workflow for loop tasks
)

# Run the batch processing
result = agents.start()

CSV File Format

The input CSV file should have headers that correspond to task parameters:
question,context,priority
"What is Python?","Programming language context","high"
"Explain machine learning","AI and ML context","medium"
"How does Docker work?","Container technology context","high"

Advanced File Processing

Processing with Multiple Columns

# Agent that uses multiple CSV columns
analyzer = Agent(
    name="Data Analyzer",
    role="Analyze data entries",
    goal="Process and analyze each data entry"
)

# Task that uses multiple columns from CSV
analysis_task = Task(
    name="analyze_entries",
    description="Analyze data: {title} with context: {context}",
    expected_output="Analysis report for each entry",
    agent=analyzer,
    task_type="loop",
    input_file="data_entries.csv",
    # Map CSV columns to task parameters
    column_mapping={
        "title": "title",
        "context": "context",
        "category": "metadata.category"
    }
)

Processing Different File Types

# Define the processor agent
processor = Agent(
    name="DataProcessor",
    role="Data processing specialist",
    goal="Process various data formats efficiently"
)

# JSON file processing
json_task = Task(
    name="process_json_data",
    description="Process JSON entries",
    expected_output="Processed results",
    agent=processor,
    task_type="loop",
    input_file="data.json",
    file_format="json"  # Specify file format
)

# Text file processing (one task per line)
text_task = Task(
    name="process_lines",
    description="Process text: {line}",
    expected_output="Processed line",
    agent=processor,
    task_type="loop",
    input_file="tasks.txt",
    file_format="text"
)

Batch Processing Patterns

Parallel Processing

# Configure parallel processing for better performance
agents = PraisonAIAgents(
    agents=[qa_agent],
    tasks=[loop_task],
    process="workflow",
    max_workers=5,  # Process 5 items in parallel
    batch_size=10   # Process in batches of 10
)

Sequential Processing with Dependencies

# Define the required agents
extractor = Agent(
    name="DataExtractor",
    role="Data extraction specialist",
    goal="Extract data from various sources"
)

transformer = Agent(
    name="DataTransformer",
    role="Data transformation expert",
    goal="Transform data to required format"
)

# First loop task processes data
extract_task = Task(
    name="extract_data",
    description="Extract data from {source}",
    expected_output="Extracted data",
    agent=extractor,
    task_type="loop",
    input_file="sources.csv"
)

# Second loop task uses results from first
transform_task = Task(
    name="transform_data",
    description="Transform extracted data",
    expected_output="Transformed data",
    agent=transformer,
    task_type="loop",
    depends_on=["extract_data"]  # Uses output from extract_data
)

Error Handling and Recovery

# Define processor agent
processor = Agent(
    name="SafeProcessor",
    role="Error-tolerant processor",
    goal="Process items with error recovery"
)

# Configure error handling for batch processing
loop_task = Task(
    name="process_with_recovery",
    description="Process item safely",
    expected_output="Processed result",
    agent=processor,
    task_type="loop",
    input_file="items.csv",
    error_handling={
        "continue_on_error": True,  # Don't stop on errors
        "max_retries": 3,          # Retry failed items
        "log_errors": True         # Log all errors
    }
)

Progress Tracking

Monitor batch processing progress:
from praisonaiagents.callbacks import Callback

class BatchProgressTracker(Callback):
    def __init__(self):
        self.processed = 0
        self.total = 0
        
    def on_task_start(self, task, **kwargs):
        if task.task_type == "loop" and self.total == 0:
            # Count total items
            import csv
            try:
                with open(task.input_file, 'r', encoding='utf-8') as f:
                    # More efficient for counting lines in a CSV
                    self.total = sum(1 for _ in f) - 1
            except FileNotFoundError:
                print(f"Warning: Input file not found at {task.input_file}. Progress will not be shown.")
                self.total = 0
    
    def on_subtask_complete(self, subtask, result, **kwargs):
        self.processed += 1
        print(f"Progress: {self.processed}/{self.total} ({self.processed/self.total*100:.1f}%)")

# Use progress tracker
agents = PraisonAIAgents(
    agents=[qa_agent],
    tasks=[loop_task],
    callbacks=[BatchProgressTracker()]
)

Output Aggregation

Collect and aggregate results from loop tasks:
# Define the summarizer agent
summarizer = Agent(
    name="Summarizer",
    role="Results aggregator",
    goal="Create comprehensive summaries from processed data"
)

# Task that aggregates all loop results
summary_task = Task(
    name="summarize_results",
    description="Create summary of all processed items",
    expected_output="Comprehensive summary report",
    agent=summarizer,
    depends_on=["process_questions"],  # Depends on loop task
    aggregate_results=True  # Receives all loop results
)

# Complete workflow
# Note: Assuming qa_agent and loop_task are defined from previous examples
agents = PraisonAIAgents(
    agents=[qa_agent, summarizer],
    tasks=[loop_task, summary_task],
    process="workflow"
)

Best Practices

  1. File Validation: Always validate input files before processing
import os
import csv

def validate_input_file(filepath):
    if not os.path.exists(filepath):
        raise FileNotFoundError(f"Input file not found: {filepath}")
    
    with open(filepath, 'r') as f:
        reader = csv.reader(f)
        headers = next(reader, None)
        if not headers:
            raise ValueError("CSV file is empty or has no headers")
    
    return True
  1. Memory Management: For large files, use streaming
loop_task = Task(
    name="process_large_file",
    description="Process item",
    expected_output="Result",
    agent=processor,
    task_type="loop",
    input_file="large_data.csv",
    streaming=True,  # Process one item at a time
    chunk_size=100   # Read 100 rows at a time
)
  1. Result Storage: Save results progressively
loop_task = Task(
    name="process_and_save",
    description="Process and save",
    expected_output="Saved result",
    agent=processor,
    task_type="loop",
    input_file="data.csv",
    output_file="results.csv",  # Save results to file
    append_mode=True  # Append results as processed
)

Troubleshooting

Loop Issues

If loops aren’t working as expected:
  • Verify input file format
  • Check task configurations
  • Enable verbose mode for debugging

Performance Issues

If processing is slow:
  • Check batch sizes
  • Verify resource allocation
  • Monitor memory usage

Next Steps

For optimal results, ensure your input files are properly formatted and your task configurations are appropriate for your use case.