02 · WorkflowCoordinator-WorkersManager-WorkerDynamic Task DecompositionMulti-Agent Collaboration

Orchestrator-Workers.
Decompose. Dispatch. Fold.

An orchestrator dynamically decomposes a task and assigns subtasks to specialized workers, managing the aggregation centrally.

When to reach for it

Subtasks only become apparent at runtime.
Workers handle highly specialized functions.
Aggregation must remain centrally controlled.

When it backfires

A static workflow suffices.
Workers lack clear responsibilities.
The orchestrator becomes a severe bottleneck.

The tradeoff

Highly flexible delegation is gained against significant coordination and integration overhead.

The mental model

A shape you can draw on a napkin.

An orchestrator decomposes the job, dispatches to workers in parallel, then folds the results.

One central LLM plans; every specialist is exposed as a typed tool call.
The orchestrator picks a tool, invokes it, and receives a return value.
After every call control returns to the center — specialists never talk to each other.

Compare all six coordination patterns →

Walk it through

A real run, step by step.

Task"Research three competitors and produce a comparison report."

1 / 6

Decomposeorchestrator(task) → ['research_acme', 'research_globalcorp', 'research_stark']

DispatchSend(subtask_1) → Worker A, Send(subtask_2) → Worker B, Send(subtask_3) → Worker C

Parallel workWorkers fetch and summarize in parallel (no shared state until fold).

Foldorchestrator([summary_A, summary_B, summary_C]) → comparison_table + narrative

Renderrespond(folded) → Markdown report

AnswerComparison report with table and narrative. Total: 1 decompose + 3 parallel worker calls + 1 fold.

In code

The loop is already built in.

LangGraphpython

from langgraph.graph import StateGraph, END
from langgraph.constants import Send

class State(TypedDict):
    task: str
    subtasks: list[str]
    results: dict[str, str]
    answer: str | None

def decompose(state):
    return {"subtasks": planner_llm(state["task"])}

def worker(state):
    # Each invocation receives one subtask
    return {"results": {state["subtask"]: research_llm(state["subtask"])}}

def fold(state):
    return {"answer": synthesizer_llm(state["results"])}

g = StateGraph(State)
g.add_node("decompose", decompose)
g.add_node("worker", worker)
g.add_node("fold", fold)

# Fan-out: one Send per subtask
g.add_conditional_edges("decompose", lambda s: [Send("worker", {"subtask": t}) for t in s["subtasks"]])
g.add_edge("worker", "fold")
g.add_edge("fold", END)
g.set_entry_point("decompose")
orchestrator = g.compile()