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"]},
   )

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

Features

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

Best Practices

Related