Skip to main content

Orchestrator Pattern

A central orchestrator agent dynamically delegates tasks to specialized worker agents.

Overview

                    ┌─────────────────┐
                    │   Orchestrator  │
                    │    (Manager)    │
                    └────────┬────────┘

         ┌───────────────────┼───────────────────┐
         │                   │                   │
         ▼                   ▼                   ▼
   ┌───────────┐      ┌───────────┐      ┌───────────┐
   │  Worker 1 │      │  Worker 2 │      │  Worker 3 │
   │(Researcher)│      │ (Analyst) │      │ (Writer)  │
   └───────────┘      └───────────┘      └───────────┘

Implementation

from praisonaiagents import Agent, Task, PraisonAIAgents

# Create worker agents
researcher = Agent(
    name="Researcher",
    instructions="Research topics when asked."
)

analyst = Agent(
    name="Analyst",
    instructions="Analyze data when asked."
)

writer = Agent(
    name="Writer",
    instructions="Write content when asked."
)

# Create orchestrator
orchestrator = Agent(
    name="Orchestrator",
    instructions="""
    You are a project manager. Break down complex tasks and delegate to:
    - Researcher: For gathering information
    - Analyst: For analyzing data
    - Writer: For creating content
    
    Coordinate their work and synthesize results.
    """,
    handoff=["Researcher", "Analyst", "Writer"]
)

# Run with hierarchical process
agents = PraisonAIAgents(
    agents=[orchestrator, researcher, analyst, writer],
    tasks=[
        Task(
            description="Create a comprehensive report on AI trends",
            agent=orchestrator
        )
    ],
    process="hierarchical"
)

result = agents.start()

Dynamic Task Delegation

from praisonaiagents import Agent

class Orchestrator:
    def __init__(self):
        self.workers = {
            "research": Agent(name="Researcher"),
            "analyze": Agent(name="Analyst"),
            "write": Agent(name="Writer")
        }
        
        self.manager = Agent(
            name="Manager",
            instructions="Decide which worker to use for each subtask."
        )
    
    def run(self, task: str) -> str:
        # Manager decides the plan
        plan = self.manager.start(f"""
        Break down this task into subtasks and assign workers:
        Task: {task}
        
        Available workers: research, analyze, write
        
        Return a JSON list of subtasks with worker assignments.
        """)
        
        # Execute subtasks
        results = []
        for subtask in self.parse_plan(plan):
            worker = self.workers[subtask["worker"]]
            result = worker.start(subtask["description"])
            results.append(result)
        
        # Synthesize results
        return self.manager.start(f"Synthesize these results: {results}")

With Feedback Loop

from praisonaiagents import Agent

orchestrator = Agent(
    name="Orchestrator",
    instructions="""
    Manage the workflow:
    1. Delegate tasks to workers
    2. Review their output
    3. Request revisions if needed
    4. Synthesize final result
    """
)

# Orchestrator can request revisions
result = orchestrator.start("""
Complete this task with quality checks:
- Research AI trends
- Analyze findings
- Write a report
- Review and revise until quality is satisfactory
""")