Orchestrator-Worker Decomposition: Splitting Complex Tasks Across Agents

Complex user requests—such as “Perform a full competitive analysis of three cloud providers and write a 10-page executive summary”—cannot be solved reliably in a single LLM prompt pass. Massive prompts suffer from context fragmentation, missed constraints, and hallucinated details.

The Orchestrator-Worker Pattern breaks monolithic goals down into structured sub-tasks. An Orchestrator agent acts as the project manager, planning and delegating execution to specialized Worker agents before synthesizing their outputs.

Architecture of Orchestrator-Worker Systems

The design consists of two distinct functional roles:

1. Orchestrator Agent: Receives high-level user goals, performs task decomposition, dynamically instantiates sub-tasks, assigns them to specialized workers, and verifies quality upon completion.
2. Worker Agents: Domain-specific agents equipped with focused tools (e.g., Code Runner, Database Search, Web Scraper) that execute single scoped sub-tasks with high accuracy.

                     [User Goal]
                          |
                  [Orchestrator Agent]
             (Creates Task Plan / Schedule)
                          |
     +--------------------+--------------------+
     |                    |                    |
[Worker: Research]  [Worker: Coding]   [Worker: Writer]
     |                    |                    |
     +--------------------+--------------------+
                          |
             (Collects & Synthesizes)
                          v
                 [Orchestrator Agent]
                          |
                  [Final Output]

Implementing Orchestrator-Worker Loops in Google ADK

Using Google Agent Development Kit (ADK) in Python, we construct an Orchestrator that delegates sub-tasks to worker instances:

from adk import Agent, Task, Workflow
from pydantic import BaseModel, Field

# Define structured task decomposition schema
class SubTaskPlan(BaseModel):
    task_id: str
    target_worker: str = Field(..., description="Worker type: 'researcher', 'coder', or 'writer'")
    instructions: str = Field(..., description="Detailed instructions for worker")

class DecompositionPlan(BaseModel):
    goal_summary: str
    subtasks: list[SubTaskPlan]

# Specialized Worker Agents
research_worker = Agent(
    name="Research Worker",
    model="gemini-1.5-pro",
    instructions="You are a research specialist. Search and summarize facts concisely."
)

coding_worker = Agent(
    name="Coding Worker",
    model="gemini-1.5-pro",
    instructions="You are a senior developer. Write and verify clean Python code."
)

# Orchestrator Agent
orchestrator = Agent(
    name="Master Orchestrator",
    model="gemini-1.5-pro",
    instructions="You manage complex workflows. Decompose goals into clear subtasks and delegate to workers."
)

async def execute_decomposed_workflow(user_prompt: str):
    # Step 1: Orchestrator creates structured plan
    plan_response = await orchestrator.generate_structured(
        prompt=f"Decompose this goal into subtasks: {user_prompt}",
        response_schema=DecompositionPlan
    )
    
    plan: DecompositionPlan = plan_response.structured_data
    worker_outputs = {}

    # Step 2: Execute sub-tasks with assigned workers
    for subtask in plan.subtasks:
        print(f"Delegating Subtask '{subtask.task_id}' to {subtask.target_worker}...")
        
        if subtask.target_worker == "researcher":
            res = await research_worker.run(subtask.instructions)
            worker_outputs[subtask.task_id] = res.text
        elif subtask.target_worker == "coder":
            res = await coding_worker.run(subtask.instructions)
            worker_outputs[subtask.task_id] = res.text

    # Step 3: Orchestrator synthesizes worker results
    synthesis_prompt = f"Original Goal: {user_prompt}\n\nWorker Results:\n{worker_outputs}\n\nSynthesize into final report."
    final_report = await orchestrator.run(synthesis_prompt)
    return final_report.text

Key Guidelines for Effective Task Decomposition

• Granular Task Boundaries: Each worker sub-task should focus on a single clear deliverable (e.g., “Search Q3 earnings report” rather than “Analyze company financial history”).
• Quality Verification Gates: The Orchestrator should evaluate worker outputs against acceptance criteria before moving to the next workflow phase.
• Dynamic Replanning: If a worker sub-task fails or returns insufficient data, allow the Orchestrator to generate alternative sub-tasks dynamically.

To learn more about ADK agent patterns, tools, and multi-agent workflows, explore the official Google ADK Repository.