Orchestrator Pattern
A central orchestrator agent dynamically delegates tasks to specialized worker agents.Overview
Copy
┌─────────────────┐
│ Orchestrator │
│ (Manager) │
└────────┬────────┘
│
┌───────────────────┼───────────────────┐
│ │ │
▼ ▼ ▼
┌───────────┐ ┌───────────┐ ┌───────────┐
│ Worker 1 │ │ Worker 2 │ │ Worker 3 │
│(Researcher)│ │ (Analyst) │ │ (Writer) │
└───────────┘ └───────────┘ └───────────┘
Implementation
Copy
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
Copy
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
Copy
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
""")
Related
- Hierarchical Process - Process types
- Handoff - Agent handoffs
- Workflows Module - Workflows reference

