Routing Tasks Across a Fleet of Worker Agents

As AI systems grow more complex, single agents often become bottlenecks. The solution? Distribute work across a fleet of specialized worker agents. But routing tasks efficiently across multiple agents requires careful orchestration.

Why Route Tasks Across Multiple Workers?

The Limitations of Single Agents

Single agents face several challenges as systems scale:

1. Context Window Limits: Loading all tools and knowledge into one agent overwhelms its context
2. Decision Fatigue: Too many tools lead to poor tool selection
3. No Parallelism: Sequential execution wastes time on independent tasks
4. Single Point of Failure: One failure stops everything

The Fleet Approach

A fleet of worker agents offers:

• Specialization: Each worker excels at specific tasks
• Parallelism: Independent tasks execute concurrently
• Scalability: Add workers without redesigning the system
• Resilience: One worker’s failure doesn’t stop others

Routing Patterns in LangGraph

LangGraph provides several patterns for routing tasks across worker agents.

1. The Router Pattern

A routing step classifies input and directs it to specialized agents:

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

class State(TypedDict):
    query: str
    classifications: List[dict]
    results: Annotated[List[str], operator.add]
    final_answer: str

def classify_query(state: State):
    """Classify the query and determine which agents to invoke."""
    # This could be an LLM call or rule-based classification
    query = state["query"].lower()
    
    classifications = []
    if "code" in query or "programming" in query:
        classifications.append({"agent": "code_expert", "query": state["query"]})
    if "data" in query or "analytics" in query:
        classifications.append({"agent": "data_expert", "query": state["query"]})
    if "design" in query or "ui" in query:
        classifications.append({"agent": "design_expert", "query": state["query"]})
    
    # Default to general agent if no specific match
    if not classifications:
        classifications.append({"agent": "general_expert", "query": state["query"]})
    
    return {"classifications": classifications}

def route_to_workers(state: State):
    """Route to relevant agents based on query classification."""
    return [
        Send(c["agent"], {"query": c["query"]})
        for c in state["classifications"]
    ]

# Worker nodes
def code_expert(state: State):
    """Handle code-related queries."""
    # Specialized code analysis logic
    return {"results": [f"Code analysis: {state['query']}"]}

def data_expert(state: State):
    """Handle data-related queries."""
    # Specialized data analysis logic
    return {"results": [f"Data analysis: {state['query']}"]}

def design_expert(state: State):
    """Handle design-related queries."""
    # Specialized design analysis logic
    return {"results": [f"Design analysis: {state['query']}"]}

def general_expert(state: State):
    """Handle general queries."""
    return {"results": [f"General analysis: {state['query']}"]}

def synthesize_results(state: State):
    """Combine results from all workers."""
    combined = "\n".join(state["results"])
    return {"final_answer": f"Combined analysis:\n{combined}"}

# Build the workflow
workflow = StateGraph(State)

# Add nodes
workflow.add_node("classifier", classify_query)
workflow.add_node("code_expert", code_expert)
workflow.add_node("data_expert", data_expert)
workflow.add_node("design_expert", design_expert)
workflow.add_node("general_expert", general_expert)
workflow.add_node("synthesizer", synthesize_results)

# Add edges
workflow.set_entry_point("classifier")
workflow.add_conditional_edges(
    "classifier",
    route_to_workers,
    ["code_expert", "data_expert", "design_expert", "general_expert"]
)
workflow.add_edge("code_expert", "synthesizer")
workflow.add_edge("data_expert", "synthesizer")
workflow.add_edge("design_expert", "synthesizer")
workflow.add_edge("general_expert", "synthesizer")
workflow.add_edge("synthesizer", END)

# Compile
app = workflow.compile()

2. The Orchestrator-Worker Pattern

The orchestrator breaks down tasks and delegates to workers:

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langgraph.types import Send

class OrchestratorState(TypedDict):
    task: str
    subtasks: List[str]
    worker_results: List[str]
    final_output: str

def orchestrator(state: OrchestratorState):
    """Break down the task into subtasks."""
    task = state["task"]
    
    # This could be an LLM call to decompose the task
    subtasks = [
        f"Research aspect 1 of: {task}",
        f"Analyze aspect 2 of: {task}",
        f"Synthesize findings for: {task}"
    ]
    
    return {"subtasks": subtasks}

def assign_workers(state: OrchestratorState):
    """Assign subtasks to workers using Send API."""
    return [
        Send("worker", {"subtask": subtask})
        for subtask in state["subtasks"]
    ]

def worker(state: dict):
    """Process a subtask."""
    subtask = state["subtask"]
    # Worker-specific logic
    return {"worker_results": [f"Completed: {subtask}"]}

def synthesizer(state: OrchestratorState):
    """Combine all worker results."""
    combined = "\n".join(state["worker_results"])
    return {"final_output": f"Final synthesis:\n{combined}"}

# Build the orchestrator-worker workflow
workflow = StateGraph(OrchestratorState)

workflow.add_node("orchestrator", orchestrator)
workflow.add_node("worker", worker)
workflow.add_node("synthesizer", synthesizer)

workflow.set_entry_point("orchestrator")
workflow.add_conditional_edges(
    "orchestrator",
    assign_workers,
    ["worker"]
)
workflow.add_edge("worker", "synthesizer")
workflow.add_edge("synthesizer", END)

app = workflow.compile()

3. The Supervisor Pattern

A supervisor agent dynamically decides which workers to call:

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

class SupervisorState(TypedDict):
    messages: Annotated[List[dict], operator.add]
    next_agent: str
    task: str

def supervisor(state: SupervisorState):
    """Decide which agent should handle the next step."""
    # This would typically be an LLM call
    task = state["task"].lower()
    
    if "research" in task:
        return {"next_agent": "researcher"}
    elif "write" in task:
        return {"next_agent": "writer"}
    elif "review" in task:
        return {"next_agent": "reviewer"}
    else:
        return {"next_agent": "generalist"}

def researcher(state: SupervisorState):
    """Perform research tasks."""
    return {
        "messages": [{"role": "assistant", "content": "Research completed"}]
    }

def writer(state: SupervisorState):
    """Write content."""
    return {
        "messages": [{"role": "assistant", "content": "Writing completed"}]
    }

def reviewer(state: SupervisorState):
    """Review work."""
    return {
        "messages": [{"role": "assistant", "content": "Review completed"}]
    }

def generalist(state: SupervisorState):
    """Handle general tasks."""
    return {
        "messages": [{"role": "assistant", "content": "General task completed"}]
    }

def route_supervisor(state: SupervisorState):
    """Route based on supervisor decision."""
    return state["next_agent"]

# Build supervisor workflow
workflow = StateGraph(SupervisorState)

workflow.add_node("supervisor", supervisor)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)
workflow.add_node("generalist", generalist)

workflow.set_entry_point("supervisor")
workflow.add_conditional_edges(
    "supervisor",
    route_supervisor,
    {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        "generalist": "generalist"
    }
)

# All agents route back to supervisor or end
for agent in ["researcher", "writer", "reviewer", "generalist"]:
    workflow.add_edge(agent, "supervisor")

app = workflow.compile()

Load Balancing Strategies

1. Round-Robin Distribution

Simple distribution across workers:

class RoundRobinRouter:
    def __init__(self, workers: List[str]):
        self.workers = workers
        self.current_index = 0
    
    def route(self, task: dict) -> str:
        worker = self.workers[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.workers)
        return worker

2. Capability-Based Routing

Route based on worker specializations:

class CapabilityRouter:
    def __init__(self):
        self.worker_capabilities = {
            "worker_a": {"code", "debugging", "testing"},
            "worker_b": {"data", "analytics", "visualization"},
            "worker_c": {"writing", "documentation", "editing"}
        }
    
    def route(self, task: dict) -> str:
        task_requirements = set(task.get("requirements", []))
        
        best_worker = None
        best_overlap = 0
        
        for worker, capabilities in self.worker_capabilities.items():
            overlap = len(task_requirements.intersection(capabilities))
            if overlap > best_overlap:
                best_overlap = overlap
                best_worker = worker
        
        return best_worker or "worker_a"  # Default fallback

3. Load-Aware Routing

Consider current worker load:

import asyncio
from typing import Dict
from dataclasses import dataclass
import time

@dataclass
class WorkerStatus:
    active_tasks: int = 0
    max_concurrent: int = 5
    avg_response_time: float = 0.0
    last_task_time: float = 0.0

class LoadAwareRouter:
    def __init__(self):
        self.workers: Dict[str, WorkerStatus] = {}
    
    def add_worker(self, worker_id: str, max_concurrent: int = 5):
        self.workers[worker_id] = WorkerStatus(max_concurrent=max_concurrent)
    
    def route(self, task: dict) -> str:
        """Route to the least loaded worker."""
        available_workers = [
            (wid, status) for wid, status in self.workers.items()
            if status.active_tasks < status.max_concurrent
        ]
        
        if not available_workers:
            # All workers at capacity, route to least loaded anyway
            return min(
                self.workers.items(),
                key=lambda x: x[1].active_tasks
            )[0]
        
        # Choose worker with lowest load
        return min(
            available_workers,
            key=lambda x: x[1].active_tasks
        )[0]
    
    def start_task(self, worker_id: str):
        self.workers[worker_id].active_tasks += 1
        self.workers[worker_id].last_task_time = time.time()
    
    def complete_task(self, worker_id: str):
        self.workers[worker_id].active_tasks -= 1

Best Practices

1. Design Clear Worker Boundaries

Each worker should have a single responsibility:

# Good: Specialized workers
workers = {
    "code_analyzer": "Analyzes code quality and suggests improvements",
    "data_processor": "Processes and transforms data",
    "document_writer": "Writes documentation and reports"
}

# Bad: Overlapping responsibilities
workers = {
    "helper": "Does everything",
    "assistant": "Helps with various tasks",
    "agent": "General purpose agent"
}

2. Implement Proper Error Handling

async def safe_worker_execution(worker_func, task, fallback=None):
    try:
        result = await worker_func(task)
        return {"success": True, "result": result}
    except Exception as e:
        if fallback:
            return await fallback(task)
        return {"success": False, "error": str(e)}

3. Monitor Worker Performance

class WorkerMonitor:
    def __init__(self):
        self.metrics = {}
    
    def record_execution(self, worker_id: str, duration: float, 
                        success: bool):
        if worker_id not in self.metrics:
            self.metrics[worker_id] = {
                "total_tasks": 0,
                "successful": 0,
                "failed": 0,
                "avg_duration": 0.0
            }
        
        metrics = self.metrics[worker_id]
        metrics["total_tasks"] += 1
        
        if success:
            metrics["successful"] += 1
        else:
            metrics["failed"] += 1
        
        # Update average duration
        n = metrics["total_tasks"]
        metrics["avg_duration"] = (
            (metrics["avg_duration"] * (n - 1) + duration) / n
        )

4. Use Structured Output

Ensure workers return consistent, parseable output:

from pydantic import BaseModel

class WorkerOutput(BaseModel):
    worker_id: str
    success: bool
    result: str
    metadata: dict = {}
    
    def to_dict(self):
        return self.model_dump()

Conclusion

Routing tasks across a fleet of worker agents unlocks parallelism, specialization, and scalability. The key patterns are:

1. Router Pattern: For classification-based dispatch
2. Orchestrator-Worker: For task decomposition and parallel execution
3. Supervisor Pattern: For dynamic, context-aware routing

Choose the pattern based on your needs:

• Router: When you have distinct input categories
• Orchestrator-Worker: When tasks need decomposition
• Supervisor: When routing depends on conversation context

With proper load balancing, error handling, and monitoring, a fleet of worker agents can handle complex workflows that would overwhelm any single agent. The result is faster, more resilient AI systems that scale with your needs.