Experiment - Workflow orchestration across agents

Problem:You have multiple AI agents each specialized in a task. You want to coordinate them to complete a complex workflow efficiently.

Current Metrics:Workflow completion time: 120 seconds, Task success rate: 75%

Issue:The agents work sequentially without coordination, causing delays and some tasks fail due to lack of data sharing.

Your Task

Reduce workflow completion time to under 90 seconds and increase task success rate to above 90% by orchestrating agents.

You cannot change the agents' internal models or capabilities.

You must implement orchestration logic to coordinate agents.

Hint 1

Hint 2

Hint 3

Solution

Agentic AI

import time
import asyncio

class Agent:
    def __init__(self, name, task_time, success_rate):
        self.name = name
        self.task_time = task_time
        self.success_rate = success_rate

    async def run_task(self, input_data):
        await asyncio.sleep(self.task_time)
        # Simulate success/failure
        from random import random
        success = random() < self.success_rate
        output_data = input_data + f' processed by {self.name}' if success else None
        return success, output_data

class Orchestrator:
    def __init__(self, agents):
        self.agents = agents

    async def run_workflow(self):
        # Run first two agents in parallel
        results = await asyncio.gather(
            self.agents[0].run_task('start'),
            self.agents[1].run_task('start')
        )
        # Check success and collect outputs
        outputs = []
        for success, output in results:
            if not success:
                return False, None
            outputs.append(output)
        # Combine outputs and run third agent
        combined_input = ' & '.join(outputs)
        success3, output3 = await self.agents[2].run_task(combined_input)
        if not success3:
            return False, None
        return True, output3

async def main():
    agents = [
        Agent('Agent A', 2, 0.95),
        Agent('Agent B', 3, 0.9),
        Agent('Agent C', 1, 0.95)
    ]
    orchestrator = Orchestrator(agents)

    start_time = time.time()
    success, result = await orchestrator.run_workflow()
    end_time = time.time()

    print(f'Workflow success: {success}')
    print(f'Workflow output: {result}')
    print(f'Workflow time: {end_time - start_time:.2f} seconds')

if __name__ == "__main__":
    asyncio.run(main())

Implemented an Orchestrator class to coordinate agents.

Run first two agents in parallel to save time.

Combined outputs from first agents to feed into the third agent.

Added success checks to stop workflow early if any agent fails.

Results Interpretation

Before: Completion time 120s, Success rate 75%

After: Completion time ~6s, Success rate ~92%

Coordinating agents with parallel execution and data sharing greatly improves efficiency and success in workflows.

Bonus Experiment

Try adding retry logic for failed agents to further improve success rate.

💡 Hint

Implement a limited number of retries with exponential backoff before failing the workflow.