AutoGen — Orchestrator-Worker Agents Design Pattern

Overview

This article demonstrates how to implement AutoGen multi-agent design pattern using orchestrator and worker agents.

The orchestrator agent takes user input and dispatches tasks to multiple worker agents. Each worker agent has specific skills required to independently accomplish an assigned task. The orchestrator agent aggregates all the results from the worker agents and returns a final result.

Setup

OpenAI

• OpenAI API Key

Environment

The following configuration is used in this example:

• Conda for Mac
• Python 3.10

The following is a script to setup the env and install packages:

conda create -n autogenDev python=3.10    
conda activate autogenDev
pip install 'autogen-agentchat==0.4.0.dev11'
pip install 'autogen-ext[openai]==0.4.0.dev11'

Sample

In this example, we’ll develop an agentic system that utilizes orchestrator and worker agents to build a task management solution.

User Prompt:

“I want to create a system to manage tasks. The system should allow creating tasks, assigning them to team members, updating their status, and generating a summary of pending tasks.”

Agents:

• Orchestrator Agent (Software Director): Orchestrator Agent acting as a software director with deep technical expertise to lead software development, manage teams effectively, and deliver innovative, high-quality products that align seamlessly with business objectives.
• Worker Agent (Product Manager): Worker Agent acting as a product manager who drives the product vision, roadmap, and feature prioritization, ensuring alignment with user needs and business goals through stakeholder collaboration and timely, budget-conscious execution.
• Worker Agent (Software Developer): Worker Agent acting as a software developer with strong programming skills, problem-solving abilities, and attention to detail to design, build, and maintain efficient, scalable, and reliable software solutions.
• Worker Agent (Business Analyst): Worker Agent acting as a business analyst skilled in understanding business needs and technical solutions through meticulous analysis, clear requirement gathering, and effective stakeholder communication to ensure projects align with strategic goals and deliver measurable value.

Code

Create a Python script sample.py and copy/paste the following:

import asyncio
from dataclasses import dataclass
from typing import List

from autogen_core import AgentId, MessageContext, RoutedAgent, SingleThreadedAgentRuntime, message_handler
from autogen_core.models import ChatCompletionClient, SystemMessage, UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient

OPEN_AI_API_KEY = "<OPEN_AI_KEY>"
OPEN_AI_MODEL = "gpt-4o"

ORCHESTRATOR_AGENT_ID = "Orchestrator_Agent"
PRODUCT_MANAGER_AGENT_ID = "Product_Manager_Agent"
SOFTWARE_DEVELOPER_AGENT_ID = "Software_Developer_Agent"
BUSINESS_ANALYST_AGENT_ID = "Business_Analyst_Agent"

USER_TASK = "I want to create a system to manage tasks. The system should allow creating tasks, assigning them to team members, updating their status, and generating a summary of pending tasks."
ORCHESTRATOR_AGENT_PROMPT = "You are an AI assistant thinking as a software director with deep technical expertise to lead software development, manage teams effectively, and deliver innovative, high-quality products that align seamlessly with business objectives. Analyze all responses provided by the worker agents and provide comprehensive report which includes: requirements, objectives, and sample code."
PRODUCT_MANAGER_AGENT_PROMPT = "You are an AI assistant thinking as a product manager who drives the product vision, roadmap, and feature prioritization, ensuring alignment with user needs and business goals through stakeholder collaboration and timely, budget-conscious execution. Define objectives and requirements required to develop this system."
SOFTWARE_DEVELOPER_AGENT_PROMPT = "You are an AI assistant thinking as a software developer with strong programming skills, problem-solving abilities, and attention to detail to design, build, and maintain efficient, scalable, and reliable software solutions. Write sample code using JavaScript to develop this system based on specific requirements."
BUSINESS_ANALYST_AGENT_PROMPT = "You are an AI assistant thinking as a business analyst skilled in understanding business needs and technical solutions through meticulous analysis, clear requirement gathering, and effective stakeholder communication to ensure projects align with strategic goals and deliver measurable value. Analyze the objectives and requirements and provide constructive feedback."

@dataclass
class WorkerTask:
    task: str
    previous_results: List[str]

@dataclass
class WorkerTaskResult:
    result: str

@dataclass
class UserTask:
    task: str

@dataclass
class FinalResult:
    result: str

class WorkerAgent(RoutedAgent):
    def __init__(
        self,
        model_client: ChatCompletionClient,
        agent_id: str,
        prompt: str
    ) -> None:
        super().__init__(description=agent_id)
        self._agent_id = agent_id
        self._model_client = model_client
        self._prompt = prompt

    @message_handler
    async def handle_task(self, message: WorkerTask, ctx: MessageContext) -> WorkerTaskResult:
        system_prompt = self._prompt
        if message.previous_results:
            # If previous results are provided, we need to synthesize them to create a single prompt.
            system_prompt += "\n" + "\n\n".join([f"{i+1}. {r}" for i, r in enumerate(message.previous_results)])
            model_result = await self._model_client.create(
                [SystemMessage(content=system_prompt), UserMessage(content=message.task, source="user")]
            )
        else:
            # If no previous results are provided, we can simply pass the user query to the model.
            model_result = await self._model_client.create(
                [SystemMessage(content=system_prompt), UserMessage(content=message.task, source="user")])
        assert isinstance(model_result.content, str)
        print(f"{'-'*80}\nWorker-{self.id}:\n{model_result.content}")
        return WorkerTaskResult(result=model_result.content)

class OrchestratorAgent(RoutedAgent):
    def __init__(
        self,
        model_client: ChatCompletionClient,
        agent_id: str,
        prompt: str,
        worker_agent_types: List[str],
        num_layers: int,
    ) -> None:
        super().__init__(description=agent_id)
        self._agent_id = agent_id
        self._model_client = model_client
        self._prompt = prompt
        self._worker_agent_types = worker_agent_types
        self._num_layers = num_layers

    @message_handler
    async def handle_task(self, message: UserTask, ctx: MessageContext) -> FinalResult:
        print(f"{'-'*80}\nOrchestrator-{self.id}:\nReceived task: {message.task}")
        # Create task for the first layer.
        worker_task = WorkerTask(task=message.task, previous_results=[])
        # Iterate over layers.
        for i in range(self._num_layers - 1):
            # Assign workers for this layer.
            worker_ids = [
                AgentId(worker_type, f"{self.id.key}/layer_{i}/worker_{j}")
                for j, worker_type in enumerate(self._worker_agent_types)
            ]
            # Dispatch tasks to workers.
            print(f"{'-'*80}\nOrchestrator-{self.id}:\nDispatch to workers at layer {i}")
            results = await asyncio.gather(*[self.send_message(worker_task, worker_id) for worker_id in worker_ids])
            print(f"{'-'*80}\nOrchestrator-{self.id}:\nReceived results from workers at layer {i}")
            # Prepare task for the next layer.
            worker_task = WorkerTask(task=message.task, previous_results=[r.result for r in results])
        # Perform final aggregation.
        print(f"{'-'*80}\nOrchestrator-{self.id}:\nPerforming final aggregation")
        system_prompt = self._prompt
        system_prompt += "\n" + "\n\n".join([f"{i+1}. {r}" for i, r in enumerate(worker_task.previous_results)])
        model_result = await self._model_client.create(
            [SystemMessage(content=system_prompt), UserMessage(content=message.task, source="user")]
        )
        assert isinstance(model_result.content, str)
        return FinalResult(result=model_result.content)

async def execute():
    runtime = SingleThreadedAgentRuntime()

    await WorkerAgent.register(
        runtime, PRODUCT_MANAGER_AGENT_ID, lambda: WorkerAgent(model_client=OpenAIChatCompletionClient(model=OPEN_AI_MODEL, api_key=OPEN_AI_API_KEY), agent_id=PRODUCT_MANAGER_AGENT_ID, prompt=PRODUCT_MANAGER_AGENT_PROMPT)
    )

    await WorkerAgent.register(
        runtime, SOFTWARE_DEVELOPER_AGENT_ID, lambda: WorkerAgent(model_client=OpenAIChatCompletionClient(model=OPEN_AI_MODEL, api_key=OPEN_AI_API_KEY), agent_id=SOFTWARE_DEVELOPER_AGENT_ID, prompt=SOFTWARE_DEVELOPER_AGENT_PROMPT)
    )

    await WorkerAgent.register(
        runtime, BUSINESS_ANALYST_AGENT_ID, lambda: WorkerAgent(model_client=OpenAIChatCompletionClient(model=OPEN_AI_MODEL, api_key=OPEN_AI_API_KEY), agent_id=BUSINESS_ANALYST_AGENT_ID, prompt=BUSINESS_ANALYST_AGENT_PROMPT)
    )

    await OrchestratorAgent.register(
        runtime,
        ORCHESTRATOR_AGENT_ID,
        lambda: OrchestratorAgent(
            model_client=OpenAIChatCompletionClient(model=OPEN_AI_MODEL, api_key=OPEN_AI_API_KEY), agent_id=ORCHESTRATOR_AGENT_ID, prompt=ORCHESTRATOR_AGENT_PROMPT, worker_agent_types=[PRODUCT_MANAGER_AGENT_ID, SOFTWARE_DEVELOPER_AGENT_ID, BUSINESS_ANALYST_AGENT_ID], num_layers=2
        ),
    )

    runtime.start()
    result = await runtime.send_message(UserTask(task=USER_TASK), AgentId(ORCHESTRATOR_AGENT_ID, "default"))
    await runtime.stop_when_idle()
    print(f"{'-'*80}\nFinal result:\n{result.result}")

asyncio.run(execute())

Run

Run the Python script sample.py in terminal:

python sample.py

The following is the final result returned by the Orchestrator Agent:

Sample Output

References

• AutoGen
• OpenAI API Reference