Queue-based task processing in Agentic Ai

Introduction

Queue-based task processing helps organize and handle tasks one by one in order. It makes sure tasks don't get lost and are done step-by-step.

When you have many tasks to do and want to handle them one at a time.

When tasks come in at different times and you want to keep them in order.

When you want to avoid doing many tasks all at once to prevent confusion.

When you want to track which tasks are waiting and which are done.

When you want to share tasks between different parts of a program safely.

Syntax

Agentic_ai

class TaskQueue:
    def __init__(self):
        self.queue = []  # List to hold tasks

    def add_task(self, task):
        self.queue.append(task)  # Add task to the end

    def process_task(self):
        if self.queue:
            task = self.queue.pop(0)  # Remove task from front
            return task
        else:
            return None

    def is_empty(self):
        return len(self.queue) == 0

    def size(self):
        return len(self.queue)

The queue uses a list to store tasks in order.

Adding a task puts it at the end; processing removes from the front.

Examples

Empty queue: no tasks yet.

Agentic_ai

task_queue = TaskQueue()
print(task_queue.is_empty())  # True
print(task_queue.size())      # 0

Queue with one task added.

Agentic_ai

task_queue = TaskQueue()
task_queue.add_task('Task 1')
print(task_queue.is_empty())  # False
print(task_queue.size())      # 1

Processing removes the first task added.

Agentic_ai

task_queue = TaskQueue()
task_queue.add_task('Task 1')
task_queue.add_task('Task 2')
print(task_queue.process_task())  # 'Task 1'
print(task_queue.size())          # 1

Processing from empty queue returns None.

Agentic_ai

task_queue = TaskQueue()
print(task_queue.process_task())  # None

Sample Program

This program creates a task queue, adds three tasks, then processes each task in order. It prints the number of tasks before and after processing.

Agentic_ai

class TaskQueue:
    def __init__(self):
        self.queue = []

    def add_task(self, task):
        self.queue.append(task)

    def process_task(self):
        if self.queue:
            task = self.queue.pop(0)
            return task
        else:
            return None

    def is_empty(self):
        return len(self.queue) == 0

    def size(self):
        return len(self.queue)

# Create a queue
my_task_queue = TaskQueue()

# Add tasks
my_task_queue.add_task('Download data')
my_task_queue.add_task('Clean data')
my_task_queue.add_task('Train model')

print(f"Tasks in queue before processing: {my_task_queue.size()}")

# Process tasks one by one
while not my_task_queue.is_empty():
    current_task = my_task_queue.process_task()
    print(f"Processing: {current_task}")

print(f"Tasks in queue after processing: {my_task_queue.size()}")

Important Notes

Time complexity: Adding tasks is fast (O(1)), but processing tasks by removing from the front is slower (O(n)) because lists shift elements.

Space complexity: The queue uses space proportional to the number of tasks stored.

Common mistake: Removing tasks from the front of a list can be slow; for many tasks, consider using collections.deque for faster operations.

Use queue processing when task order matters and you want to handle tasks one at a time.

Summary

Queue-based task processing keeps tasks in order and handles them one by one.

Adding tasks puts them at the end; processing removes from the front.

This method helps organize work and avoid confusion when many tasks arrive.