Orchestrator Workflow

Decompose complex tasks into subtasks and coordinate execution across multiple agents

Overview

The Orchestrator workflow uses an LLM to break down complex tasks into manageable subtasks, assign them to appropriate specialized agents, and aggregate the results. It’s ideal for sophisticated, multi-step operations where the optimal task breakdown isn’t known in advance.

Orchestrators automatically generate planning prompts based on available agent capabilities, enabling dynamic task decomposition.

Key Features

• Dynamic Planning: LLM generates execution plans based on task requirements
• Task Decomposition: Breaks complex tasks into focused subtasks
• Agent Coordination: Assigns subtasks to appropriate specialized agents
• Two Planning Modes: Full upfront planning or iterative step-by-step
• Result Aggregation: Combines outputs into coherent final results

Basic Usage

import asyncio
from fast_agent import FastAgent

fast = FastAgent("Orchestrator-Workers")

# Define specialized worker agents
@fast.agent(
    name="finder",
    instruction="""You are an agent with access to the filesystem and URL fetching.
    Your job is to identify the closest match to a user's request, 
    make the appropriate tool calls, and return the URI and CONTENTS 
    of the closest match.""",
    servers=["fetch", "filesystem"],
    model="gpt-4.1",
)
@fast.agent(
    name="writer",
    instruction="""You are an agent that can write to the filesystem.
    You are tasked with taking the user's input, addressing it, and 
    writing the result to disk in the appropriate location.""",
    servers=["filesystem"],
)
@fast.agent(
    name="proofreader",
    instruction="""Review text for grammar, spelling, and punctuation errors.
    Identify any awkward phrasing or structural issues that could improve clarity. 
    Provide detailed feedback on corrections.""",
    servers=["fetch"],
    model="gpt-4.1",
)

# Define the orchestrator to coordinate the workers
@fast.iterative_planner(
    name="orchestrate",
    agents=["finder", "writer", "proofreader"],
    model="sonnet",
    plan_iterations=5,
)
async def main() -> None:
    async with fast.run() as agent:
        task = """Load the student's short story from short_story.md, 
        and generate a report with feedback across proofreading, 
        factuality/logical consistency and style adherence. Use the style rules from 
        https://apastyle.apa.org/learn/quick-guide-on-formatting and 
        https://apastyle.apa.org/learn/quick-guide-on-references.
        Write the graded report to graded_report.md in the same directory as short_story.md"""
        
        await agent.orchestrate(task)

if __name__ == "__main__":
    asyncio.run(main())

Configuration Parameters

name · string · required

Name of the orchestrator workflow

agents · list[str] · required

List of agent names the orchestrator can use

instruction · string

Base instruction for the orchestrator

model · string

Model for planning (recommend capable models like Sonnet or GPT-4)

use_history · bool · default: "false"

Whether orchestrator maintains chat history

human_input · bool · default: "false"

Whether orchestrator can request human input

plan_type · string · default: "full"

Planning approach: "full" (upfront) or "iterative" (step-by-step)

plan_iterations · int · default: "5"

Maximum number of planning attempts or iterations

Planning Modes

Full Planning

Generates complete execution plan upfront:

@fast.orchestrator(
    name="project_manager",
    agents=["researcher", "designer", "developer", "tester"],
    plan_type="full",
    plan_iterations=3,  # Max attempts to create valid plan
)

Workflow:

1. Analyze the complete task
2. Generate full execution plan
3. Execute all steps sequentially
4. Aggregate final results

Best for:

• Well-defined tasks with predictable steps
• Tasks where all subtasks are known upfront
• Scenarios requiring validated plans before execution

Iterative Planning

Plans and executes one step at a time:

@fast.iterative_planner(
    name="adaptive_manager",
    agents=["analyzer", "implementer", "validator"],
    plan_iterations=10,  # Max steps in the plan
)

Workflow:

1. Analyze task and determine next step
2. Execute that step
3. Evaluate results
4. Plan next step based on outcomes
5. Repeat until task complete or max iterations reached

Best for:

• Complex tasks with uncertain paths
• Tasks where later steps depend on earlier results
• Exploratory or research-oriented workflows

How It Works

Advanced Examples

Research and Analysis Pipeline

@fast.agent(
    "web_researcher",
    instruction="Search the web and gather relevant information",
    servers=["fetch"],
)
@fast.agent(
    "data_analyzer",
    instruction="Analyze data and identify patterns and insights",
)
@fast.agent(
    "report_writer",
    instruction="Write comprehensive, well-structured reports",
    servers=["filesystem"],
)
@fast.agent(
    "fact_checker",
    instruction="Verify facts and check for inconsistencies",
    servers=["fetch"],
)

@fast.orchestrator(
    name="research_orchestrator",
    agents=["web_researcher", "data_analyzer", "report_writer", "fact_checker"],
    model="sonnet",
    plan_type="full",
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.research_orchestrator.send(
            """Research the current state of quantum computing, 
            analyze key developments, verify all claims, 
            and produce a comprehensive report saved to quantum_report.md"""
        )

Software Development Orchestrator

@fast.agent(
    "requirements_analyst",
    instruction="Analyze requirements and create detailed specifications",
)
@fast.agent(
    "code_generator",
    instruction="Generate code based on specifications",
    servers=["filesystem"],
)
@fast.agent(
    "test_writer",
    instruction="Create comprehensive unit tests",
    servers=["filesystem"],
)
@fast.agent(
    "code_reviewer",
    instruction="Review code for quality, bugs, and best practices",
    servers=["filesystem"],
)
@fast.agent(
    "doc_writer",
    instruction="Write technical documentation",
    servers=["filesystem"],
)

@fast.iterative_planner(
    name="dev_orchestrator",
    agents=[
        "requirements_analyst",
        "code_generator",
        "test_writer",
        "code_reviewer",
        "doc_writer",
    ],
    plan_iterations=15,
    model="sonnet",
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.dev_orchestrator.send(
            "Create a Python module for parsing JSON configuration files with validation"
        )

Content Production Pipeline

@fast.agent(
    "topic_researcher",
    instruction="Research topics and gather source material",
    servers=["fetch"],
)
@fast.agent(
    "outline_creator",
    instruction="Create detailed content outlines",
)
@fast.agent(
    "content_writer",
    instruction="Write engaging, well-structured content",
)
@fast.agent(
    "editor",
    instruction="Edit content for clarity, style, and grammar",
)
@fast.agent(
    "seo_optimizer",
    instruction="Optimize content for search engines",
)
@fast.agent(
    "publisher",
    instruction="Format and publish content",
    servers=["filesystem"],
)

@fast.orchestrator(
    name="content_pipeline",
    agents=[
        "topic_researcher",
        "outline_creator",
        "content_writer",
        "editor",
        "seo_optimizer",
        "publisher",
    ],
    plan_type="full",
    model="sonnet",
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.content_pipeline.send(
            "Create a comprehensive blog post about FastAPI best practices"
        )

Orchestrator vs Other Patterns

Feature	Orchestrator	Router	Agents as Tools	Chain
Task Decomposition	✅ Dynamic	❌ None	✅ LLM-driven	❌ Fixed
Planning	✅ LLM generates	❌ None	✅ Implicit	❌ Predefined
Agent Selection	✅ Per subtask	✅ Single	✅ Per tool call	❌ Fixed
Parallel Execution	❌ Sequential	❌ Single	✅ Parallel	❌ Sequential
Complexity	High	Low	Medium	Low
Best For	Complex tasks	Simple routing	Dynamic composition	Known sequences

Best Practices

Capable Planner Model

Use a powerful model (Sonnet, GPT-4) for orchestrator planning

Clear Agent Roles

Define distinct, well-documented capabilities for worker agents

Appropriate Planning Mode

Use full planning for predictable tasks, iterative for exploratory work

Iteration Limits

Set reasonable plan_iterations to prevent infinite loops

Common Patterns

The Research Pattern

1. Research/Gather → 2. Analyze → 3. Synthesize → 4. Write → 5. Review

The Development Pattern

1. Requirements → 2. Design → 3. Implement → 4. Test → 5. Document

The Content Pattern

1. Topic Research → 2. Outline → 3. Draft → 4. Edit → 5. Optimize → 6. Publish

The Quality Assurance Pattern

1. Create → 2. Review → 3. Revise → 4. Validate → 5. Approve

Debugging Orchestrator Plans

Enable detailed logging to see the execution plan:

import logging
logging.basicConfig(level=logging.DEBUG)

# You'll see:
# DEBUG: Orchestrator generated plan:
# 1. Use 'finder' to locate short_story.md
# 2. Use 'proofreader' to review the story
# 3. Use 'writer' to create graded_report.md

Performance Considerations

Orchestrators add significant overhead:

• Multiple LLM calls for planning
• Sequential execution of subtasks
• Aggregation of results

For simpler tasks, consider Chain or Router instead.

Use Cases

• Complex Research: Multi-source research with analysis and synthesis
• Software Development: Requirements through deployment
• Content Production: Research, writing, editing, optimization
• Data Processing: Extract, transform, analyze, visualize, report
• Quality Workflows: Create, review, revise, approve cycles
• Multi-Step Analysis: Gather, analyze, validate, summarize

Related Patterns

• Agents as Tools - Similar capability with parallel execution and simpler implementation
• Chain - Fixed sequence execution (simpler but less flexible)
• Router - Single agent selection (simpler delegation)
• Iterative Planner - Step-by-step planning variant