Skip to main content
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

The orchestrator should return a single routing keyword — avoid asking it to perform worker tasks directly.
Give workers narrow roles (research, code, writing) so the orchestrator can route confidently.
A final synthesizer agent merges worker outputs into one coherent response for the user.

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