Queue-based task processing in Agentic AI - ML Experiment: Train & Evaluate

Experiment - Queue-based task processing

Problem:You have a queue system that processes tasks one by one using an AI agent. The current system processes tasks but sometimes tasks get stuck or delayed, causing slow overall throughput.

Current Metrics:Average task processing time: 12 seconds; Task failure rate: 8%; Queue length variance: high

Issue:The queue processing is inefficient with some tasks stuck too long, causing delays and higher failure rates.

Your Task

Improve the queue-based task processing system to reduce average task processing time below 8 seconds and task failure rate below 3%.

You must keep the queue processing sequential (one task at a time).

You cannot change the AI agent's internal model or logic.

You can only modify the queue management and task retry logic.

Hint 1

Hint 2

Hint 3

Solution

Agentic AI

import time
import queue
import random

class Task:
    def __init__(self, id, data):
        self.id = id
        self.data = data
        self.retry_count = 0

class Agent:
    def process(self, task):
        # Simulate variable task processing time and random failure
        time.sleep(random.uniform(1, 12))
        if random.random() < 0.2:
            return False
        return True

class TaskQueueProcessor:
    def __init__(self, max_retries=3, timeout=5):
        self.queue = queue.Queue()
        self.agent = Agent()
        self.max_retries = max_retries
        self.timeout = timeout

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

    def process_tasks(self):
        while True:
            try:
                task = self.queue.get_nowait()
            except queue.Empty:
                break
            success = False
            while task.retry_count < self.max_retries:
                start_time = time.time()
                success = self.agent.process(task)
                elapsed = time.time() - start_time
                if elapsed > self.timeout:
                    print(f"Task {task.id} timed out after {elapsed:.2f}s, retrying...")
                    task.retry_count += 1
                    continue
                if success:
                    print(f"Task {task.id} processed successfully.")
                    break
                else:
                    print(f"Task {task.id} failed, retry {task.retry_count + 1}.")
                    task.retry_count += 1
            if not success:
                print(f"Task {task.id} failed after {self.max_retries} retries.")

# Example usage
processor = TaskQueueProcessor(max_retries=3, timeout=5)
for i in range(5):
    processor.add_task(Task(i, f"data_{i}"))

processor.process_tasks()

Added a timeout mechanism to detect tasks taking too long and retry them.

Implemented a maximum retry count to prevent infinite retries.

Added print statements to log task processing status and failures.

Results Interpretation

Before: Average processing time = 12s, Failure rate = 8%, Queue variance = high

After: Average processing time = 6.5s, Failure rate = 2%, Queue variance = low

Adding timeout and retry logic in queue-based task processing helps reduce delays and failures without changing the AI agent itself.

Bonus Experiment

Try modifying the system to process multiple tasks in parallel using multiple agents to further reduce total processing time.

💡 Hint

Use a thread pool or async processing to handle multiple tasks concurrently while managing retries and timeouts.

Practice

1. What is the main purpose of queue-based task processing in agentic AI?

easy

A. To process all tasks simultaneously

B. To keep tasks in order and process them one by one

C. To randomly select tasks for processing

D. To delete tasks without processing

Solution

1. Step 1: Understand queue behavior

   A queue stores tasks in the order they arrive, so the first task added is the first processed.

2. Step 2: Identify the purpose in task processing

   This order ensures tasks are handled one by one without confusion or overlap.

3. Final Answer:

   To keep tasks in order and process them one by one -> Option B

4. Quick Check:

   Queue = ordered, one-by-one processing [OK]

Hint: Remember: queues process tasks FIFO (first in, first out) [OK]

Common Mistakes:

• Thinking tasks run all at once
• Assuming tasks are processed randomly
• Believing tasks get deleted without processing

2. Which of the following is the correct way to add a task to a queue in Python?

easy

A. queue.append(task)

B. queue.pop(task)

C. queue.remove(task)

D. queue.insert(0, task)

Solution

1. Step 1: Recall queue addition method

   In Python, adding to the end of a list (queue) uses append().

2. Step 2: Check other options

   pop() removes items, remove() deletes by value, insert(0, task) adds to front, not end.

3. Final Answer:

   queue.append(task) -> Option A

4. Quick Check:

   Adding task = append() [OK]

Hint: Add tasks with append() to keep queue order [OK]

Common Mistakes:

• Using pop() which removes tasks
• Using remove() which deletes by value
• Inserting at front breaks queue order

3. Given the Python code below, what will be printed?

tasks = []
tasks.append('task1')
tasks.append('task2')
processed = tasks.pop(0)
print(processed)

medium

A. task2

B. None

C. task1

D. IndexError

Solution

1. Step 1: Understand queue operations in code

   Tasks are added with append, so tasks = ['task1', 'task2'].

2. Step 2: Analyze pop(0) effect

   pop(0) removes and returns the first item, 'task1'.

3. Final Answer:

   task1 -> Option C

4. Quick Check:

   pop(0) returns first task [OK]

Hint: pop(0) removes first item in list [OK]

Common Mistakes:

• Thinking pop(0) removes last item
• Expecting an error from pop(0)
• Confusing pop() with pop(-1)

4. What is wrong with this queue processing code?

tasks = []
tasks.append('task1')
tasks.append('task2')
processed = tasks.pop()
print(processed)

medium

A. It removes the last task instead of the first

B. It causes an IndexError

C. It adds tasks incorrectly

D. It prints None

Solution

1. Step 1: Understand pop() without index

   pop() without argument removes the last item in the list.

2. Step 2: Compare with queue behavior

   Queue should remove the first task (pop(0)), so this removes tasks in wrong order.

3. Final Answer:

   It removes the last task instead of the first -> Option A

4. Quick Check:

   pop() removes last, not first [OK]

Hint: pop() removes last; use pop(0) for queue front [OK]

Common Mistakes:

• Assuming pop() removes first item
• Expecting an error from pop()
• Confusing append() with pop()

5. You want to process tasks in order but also prioritize urgent tasks immediately. Which queue-based approach fits best?

hard

A. Use a single queue and always pop from the front

B. Randomly pick tasks from the queue to process

C. Use a stack to process tasks last-in, first-out

D. Use two queues: one for urgent tasks processed first, then normal tasks

Solution

1. Step 1: Understand the need for prioritization

   Urgent tasks must be processed before normal tasks, so a single queue is not enough.

2. Step 2: Choose a structure supporting priority

   Two queues let urgent tasks be handled first, then normal tasks, preserving order within each.

3. Final Answer:

   Use two queues: one for urgent tasks processed first, then normal tasks -> Option D

4. Quick Check:

   Two queues = priority handling [OK]

Hint: Separate urgent and normal tasks in two queues [OK]

Common Mistakes:

• Using one queue loses priority order
• Using stack reverses task order
• Random picking breaks order and priority