tools-only 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