Agentic Orchestrator Worker

Learn how to create AI agents that orchestrate and distribute tasks among specialized workers.

A workflow with a central orchestrator directing multiple worker LLMs to perform subtasks, synthesizing their outputs for complex, coordinated operations.

from praisonaiagents import Agent, AgentFlow

orchestrator = Agent(
    name="Orchestrator",
    instructions="Route each task to the research, code, or writing worker.",
)
flow = AgentFlow(start=orchestrator)
flow.start("Research AI trends and draft a one-page summary.")

The user submits a complex task; the orchestrator routes subtasks to workers and synthesises the final answer.

Quick Start

1. Simple Usage

   pip install praisonaiagents
   export OPENAI_API_KEY="${OPENAI_API_KEY:?Set OPENAI_API_KEY in your shell}"
   

   Create app.py:

   from praisonaiagents import Agent, AgentFlow
   from praisonaiagents import route
   
   # Create orchestrator agent that decides task routing
   orchestrator = Agent(
       name="Orchestrator",
       role="Task Orchestrator",
       goal="Analyze tasks and route to appropriate workers",
       instructions="Analyze the task. Respond with ONLY 'research', 'code', or 'writing' based on task type."
   )
   
   # Create specialized worker agents
   research_worker = Agent(
       name="ResearchWorker",
       role="Research Specialist",
       goal="Conduct thorough research",
       instructions="You are a research specialist. Provide detailed research findings."
   )
   
   code_worker = Agent(
       name="CodeWorker",
       role="Software Developer",
       goal="Write and review code",
       instructions="You are a software developer. Write clean, efficient code."
   )
   
   writing_worker = Agent(
       name="WritingWorker",
       role="Content Writer",
       goal="Create written content",
       instructions="You are a content writer. Write clear, engaging content."
   )
   
   # Create synthesizer agent
   synthesizer = Agent(
       name="Synthesizer",
       role="Result Synthesizer",
       goal="Synthesize and summarize results",
       instructions="Summarize the work completed and provide a final polished output."
   )
   
   # Create orchestrated workflow
   workflow = AgentFlow(
       steps=[
           orchestrator,  # First, orchestrator decides routing
           route({
               "research": [research_worker],
               "code": [code_worker],
               "writing": [writing_worker]
           }),
           synthesizer  # Finally, synthesize results
       ]
   )
   
   # Run orchestrated workflow
   result = workflow.start("Create a Python function to calculate fibonacci numbers")
   print(f"Result: {result['output'][:500]}...")
   

   Run it:

   python app.py
   
2. With Configuration

   Add routing tools and explicit task wiring for finer control:

   from praisonaiagents import Agent, Task
   
   router = Agent(
       name="Router",
       role="Task router",
       goal="Distribute tasks based on conditions",
       tools=[get_time_check],
   )
   
   worker = Agent(
       name="Worker",
       role="Specialized worker",
       goal="Handle specific task type",
       instructions="Processing instructions",
   )
   
   router_task = Task(
       name="route_task",
       description="Route tasks to workers",
       agent=router,
       is_start=True,
       task_type="decision",
       condition={"1": ["worker1_task"], "2": ["worker2_task"]},
   )
   

Requirements

• Python 3.10 or higher
• OpenAI API key. Generate OpenAI API key here. Use Other models using this guide.
• Basic understanding of Python

How It Works

The user submits a task; the orchestrator picks a worker via route(), the worker executes, and the synthesizer merges the result.

---

Understanding Orchestrator-Worker Pattern

What is Orchestrator-Worker?

Orchestrator-Worker pattern enables:

• Dynamic task distribution and routing
• Specialized worker execution
• Result synthesis and aggregation
• Coordinated workflow management

Features

Task Routing

Intelligently distribute tasks to specialized workers.

Worker Specialization

Dedicated agents for specific task types.

Result Synthesis

Combine and process worker outputs effectively.

Process Control

Monitor and manage the orchestrated workflow.

Configuration Options

# Create an orchestrator agent
router = Agent(
    name="Router",
    role="Task router",
    goal="Distribute tasks based on conditions",
    tools=[get_time_check],  # Tools for routing decisions
      # Enable detailed logging
)

# Create a worker agent
worker = Agent(
    name="Worker",
    role="Specialized worker",
    goal="Handle specific task type",
    instructions="Processing instructions"
)

# Create routing task
router_task = Task(
    name="route_task",
    description="Route tasks to workers",
    agent=router,
    is_start=True,
    task_type="decision",
    condition={
        "1": ["worker1_task"],
        "2": ["worker2_task"]
    }
)

# Create synthesis task
synthesis_task = Task(
    name="synthesize",
    description="Combine worker results",
    agent=synthesizer,
    context=[worker1_task, worker2_task]  # Reference worker tasks
)

Troubleshooting

Routing Issues

If task routing fails:

• Check routing conditions
• Verify worker availability
• Enable verbose mode for debugging

Synthesis Flow

If result synthesis is incorrect:

• Review worker outputs
• Check context connections
• Verify synthesis logic

Best Practices

Keep orchestrator instructions routing-only

The orchestrator should return a single routing keyword — avoid asking it to perform worker tasks directly.

Specialise each worker

Give workers narrow roles (research, code, writing) so the orchestrator can route confidently.

Always add a synthesizer step

A final synthesizer agent merges worker outputs into one coherent response for the user.

Related

AutoAgents

Automatically created and managed AI agents

Mini Agents

Lightweight, focused AI agents

Multi-Agent Pipelines

Chain agents into production pipelines

Nested Workflows

Compose workflows inside workflows