Orchestrator-Worker Pattern

Orchestrator-Worker Pattern

The orchestrator-worker pattern distributes independent tasks across multiple worker agents that execute in parallel, then aggregates their results. Best for parallelizable workloads.

When to Use

• Tasks can be decomposed into independent subtasks
• Subtasks can execute simultaneously
• Results need aggregation or synthesis
• Performance/throughput is critical

Architecture

Request → Orchestrator (split) → Worker 1 ┐
                               → Worker 2 ├→ Aggregator → Result
                               → Worker 3 ┘

Implementation

Python with Send API

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END, Send

class OrchestratorState(TypedDict):
    """State for orchestrator pattern."""
    task: str
    subtasks: list[dict]
    results: Annotated[list[dict], operator.add]
    final_result: str

class WorkerState(TypedDict):
    """State for individual worker."""
    subtask: dict

def orchestrator_node(state: OrchestratorState) -> dict:
    """Split task into subtasks and dispatch to workers."""
    task = state["task"]

    # Decompose into subtasks (could use LLM)
    subtasks = [
        {"id": 1, "query": "Research topic A"},
        {"id": 2, "query": "Research topic B"},
        {"id": 3, "query": "Research topic C"}
    ]

    return {"subtasks": subtasks}

def worker_node(state: WorkerState) -> dict:
    """Process individual subtask."""
    subtask = state["subtask"]

    # Process subtask
    result = f"Result for {subtask['query']}"

    return {"results": [{"result": result, "status": "success"}]}

def aggregator_node(state: OrchestratorState) -> dict:
    """Aggregate worker results."""
    results = state["results"]

    # Combine results (could use LLM for synthesis)
    final_result = "\n".join(r["result"] for r in results)

    return {"final_result": final_result}

# Graph assembly with parallel execution
def create_orchestrator_graph():
    """Build orchestrator-worker graph with parallel execution."""

    def dispatch_workers(state: OrchestratorState):
        """Create Send objects for parallel worker execution."""
        return [
            Send("worker", {"subtask": subtask})
            for subtask in state["subtasks"]
        ]

    graph = StateGraph(OrchestratorState)

    graph.add_node("orchestrator", orchestrator_node)
    graph.add_node("worker", worker_node)
    graph.add_node("aggregator", aggregator_node)

    graph.add_edge(START, "orchestrator")

    # Dispatch to workers in parallel
    graph.add_conditional_edges(
        "orchestrator",
        dispatch_workers
    )

    # Workers send results to aggregator
    graph.add_edge("worker", "aggregator")
    graph.add_edge("aggregator", END)

    return graph.compile()

JavaScript

import { StateGraph, START, END, Send } from "@langchain/langgraph";
import { Annotation } from "@langchain/langgraph";

const OrchestratorStateAnnotation = Annotation.Root({
  task: Annotation<string>(),
  subtasks: Annotation<Array<any>>({
    reducer: (_, right) => right,
    default: () => [],
  }),
  results: Annotation<Array<{ result: string; status?: string }>>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  finalResult: Annotation<string>(),
});

function orchestratorNode(state: any): Partial<any> {
  const task = state.task;

  const subtasks = [
    { id: 1, query: "Research topic A" },
    { id: 2, query: "Research topic B" },
    { id: 3, query: "Research topic C" }
  ];

  return { subtasks };
}

function workerNode(state: any): Partial<any> {
  const subtask = state.subtask;
  const result = `Result for ${subtask.query}`;

  return { results: [{ result, status: "success" }] };
}

function aggregatorNode(state: any): Partial<any> {
  const results = state.results;
  const finalResult = results.map((r: any) => r.result).join("\n");

  return { finalResult };
}

export function createOrchestratorGraph() {
  function dispatchWorkers(state: any) {
    return state.subtasks.map((subtask: any) =>
      new Send("worker", { subtask })
    );
  }

  const graph = new StateGraph(OrchestratorStateAnnotation)
    .addNode("orchestrator", orchestratorNode)
    .addNode("worker", workerNode)
    .addNode("aggregator", aggregatorNode)

    .addEdge(START, "orchestrator")

    .addConditionalEdges("orchestrator", dispatchWorkers)

    .addEdge("worker", "aggregator")
    .addEdge("aggregator", END);

  return graph.compile();
}

Task Decomposition Strategies

LLM-Based Decomposition

from langchain_openai import ChatOpenAI
from pydantic import BaseModel

class TaskDecomposition(BaseModel):
    """Structured task decomposition."""
    subtasks: list[str]
    dependencies: dict[str, list[str]]

def decompose_task(task: str) -> TaskDecomposition:
    """Use LLM to break down task."""
    model = ChatOpenAI(model="gpt-4")
    structured = model.with_structured_output(TaskDecomposition)

    prompt = f"""Break down this task into independent subtasks that can run in parallel:
    Task: {task}

    Identify:
    1. Subtasks that can execute simultaneously
    2. Any dependencies between subtasks
    """

    return structured.invoke(prompt)

Rule-Based Decomposition

def decompose_research_task(topic: str) -> list[dict]:
    """Decompose research task into subtasks."""
    return [
        {"type": "web_search", "query": f"{topic} overview"},
        {"type": "academic_search", "query": f"{topic} research papers"},
        {"type": "news_search", "query": f"{topic} recent news"},
        {"type": "expert_search", "query": f"{topic} experts"}
    ]

Result Aggregation

Simple Concatenation

def simple_aggregator(results: list[dict]) -> str:
    """Concatenate results."""
    return "\n\n".join(f"## Result {i+1}\n{r['result']}" for i, r in enumerate(results))

LLM-Based Synthesis

def llm_aggregator(results: list[dict], original_task: str) -> str:
    """Use LLM to synthesize results."""
    model = ChatOpenAI(model="gpt-4")

    prompt = f"""Synthesize these research results into a coherent answer:

    Original question: {original_task}

    Results:
    {chr(10).join(f"{i+1}. {r['result']}" for i, r in enumerate(results))}

    Provide a comprehensive, well-organized answer.
    """

    return model.invoke(prompt).content

Structured Aggregation

from pydantic import BaseModel

class AggregatedResult(BaseModel):
    """Structured aggregation."""
    summary: str
    key_findings: list[str]
    sources: list[str]
    confidence: float

def structured_aggregator(results: list[dict]) -> AggregatedResult:
    """Aggregate into structured format."""
    model = ChatOpenAI(model="gpt-4")
    structured = model.with_structured_output(AggregatedResult)

    return structured.invoke({
        "results": results,
        "instruction": "Synthesize these results"
    })

Handling Worker Failures

Retry Strategy

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def resilient_worker(state: WorkerState) -> dict:
    """Worker with automatic retry."""
    try:
        result = process_subtask(state["subtask"])
        return {"results": [{"result": result, "status": "success"}]}
    except Exception as e:
        return {"results": [{"result": str(e), "status": "failed"}]}

Fallback Handling

def robust_aggregator(state: OrchestratorState) -> dict:
    """Handle failed workers."""
    results = state.get("results", [])

    # Filter out failed results
    successful_results = [
        r for r in results
        if isinstance(r, dict) and r.get("status") == "success"
    ]

    if len(successful_results) == 0:
        return {"final_result": "All workers failed", "status": "error"}

    # Aggregate successful results
    final_result = aggregate_results([r["result"] for r in successful_results])

    # Include warning about failures
    if len(successful_results) < len(results):
        final_result += f"\n\nWarning: {len(results) - len(successful_results)} workers failed"

    return {"final_result": final_result, "status": "partial_success"}

Best Practices

1. Resource Management

Limit concurrent workers to avoid overwhelming resources:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import RetryPolicy

def create_orchestrator_graph():
    graph = StateGraph(OrchestratorState)

    # Configure with resource limits
    graph.add_node(
        "worker",
        worker_node,
        retry_policy=RetryPolicy(max_attempts=3)
    )

    # Add checkpointer for state persistence
    checkpointer = MemorySaver()

    return graph.compile(checkpointer=checkpointer)

2. Progress Tracking

Track worker progress in state:

from typing import TypedDict, Annotated
import operator

class OrchestratorState(TypedDict):
    task: str
    subtasks: list[dict]
    completed: int
    total: int
    results: Annotated[list[dict], operator.add]

def track_progress(state: OrchestratorState) -> dict:
    """Update progress tracking."""
    return {
        "completed": len(state.get("results", [])),
        "total": len(state.get("subtasks", []))
    }

3. Timeout Handling

Set timeouts for workers:

import asyncio
from concurrent.futures import TimeoutError

async def worker_with_timeout(state: WorkerState, timeout: int = 30) -> dict:
    """Worker with timeout."""
    try:
        result = await asyncio.wait_for(
            process_subtask_async(state["subtask"]),
            timeout=timeout
        )
        return {"results": [{"result": result, "status": "success"}]}
    except TimeoutError:
        return {"results": [{"result": "Task timed out", "status": "timeout"}]}

When to Use vs Alternatives

Use orchestrator-worker when:

• Tasks are independent and parallelizable
• Throughput is important
• Results need synthesis

Use supervisor pattern instead when:

• Tasks are sequential or dependent
• Agents need to collaborate
• Dynamic routing is needed

Use map-reduce pattern instead when:

• Processing large datasets
• Uniform processing across items
• No complex aggregation logic needed

Performance Considerations

Advantages:

• Parallel execution reduces total time
• Scales horizontally
• High throughput

Challenges:

• All workers consume resources simultaneously
• Aggregation waits for slowest worker
• Memory usage grows with worker count
• Cost scales with parallelism

Optimizations:

• Batch workers if too many subtasks
• Stream results to aggregator incrementally
• Use cheaper models for workers
• Implement worker pooling

Testing

def test_orchestrator_pattern():
    """Test orchestrator-worker pattern."""
    graph = create_orchestrator_graph()

    # Test with sample task
    result = graph.invoke({
        "task": "Research artificial intelligence"
    })

    assert "final_result" in result
    assert len(result["results"]) == len(result["subtasks"])

def test_worker_isolation():
    """Ensure workers don't interfere."""
    # Workers should process independently
    state1 = {"subtask": {"id": 1, "query": "Task 1"}}
    state2 = {"subtask": {"id": 2, "query": "Task 2"}}

    result1 = worker_node(state1)
    result2 = worker_node(state2)

    assert result1["results"][0]["result"] != result2["results"][0]["result"]

Additional Resources

• LangGraph Map-Reduce: https://docs.langchain.com/oss/python/langgraph/use-graph-api#map-reduce-and-the-send-api
• Send API Documentation: https://reference.langchain.com/python/langgraph/types/?h=send#langgraph.types.Send