Agent Pattern: Supervisor (Manager + Workers)

[yasha-dev1]

Overview

The Supervisor pattern is a multi-agent architecture where a central supervisor agent coordinates specialized worker agents. This approach excels when tasks require different types of expertise. Rather than building one agent that manages tool selection across domains, you create focused specialists coordinated by a supervisor who understands the overall workflow.

How It Works

1. Central Orchestrator: A supervisor agent receives the task and decides which worker agents to invoke
2. Specialized Workers: Each worker agent has specific tools and expertise for a domain
3. Dynamic Routing: The supervisor analyzes the task context and adaptively routes to appropriate workers
4. Result Synthesis: The supervisor collects and synthesizes results from multiple workers

Control Flow:

User Request → Supervisor Agent
                    ↓
    ┌───────────────┼───────────────┐
    ↓               ↓               ↓
Researcher      Analyst         Writer
(web_search)   (calculator)  (file_writer)
    ↓               ↓               ↓
    └───────────────┼───────────────┘
                    ↓
              Final Result

Reference Implementations

• LangGraph Supervisor - Official LangGraph library for hierarchical multi-agent systems
• LangGraph Multi-Agent Workflows - Tutorial on building supervisor architectures
• LangGraph Supervisor Documentation - API reference for supervisor patterns
• CrewAI Multi-Agent Teams - Comparison of supervisor approaches across frameworks

Proposed PyWorkflow Implementation

from pyworkflow_agents import SupervisorAgent, Agent
from pyworkflow_agents.providers import AnthropicProvider

# Define specialized worker agents
researcher = Agent(
    name="researcher",
    provider=AnthropicProvider(model="claude-sonnet-4-5-20250929"),
    instructions="You are a research specialist. Search the web for accurate information.",
    tools=[web_search_tool],
)

analyst = Agent(
    name="analyst",
    provider=AnthropicProvider(model="claude-sonnet-4-5-20250929"),
    instructions="You are a data analyst. Perform calculations and analysis.",
    tools=[calculator_tool, data_analysis_tool],
)

writer = Agent(
    name="writer",
    provider=AnthropicProvider(model="claude-sonnet-4-5-20250929"),
    instructions="You are a technical writer. Create clear documentation.",
    tools=[file_writer_tool, markdown_formatter_tool],
)

# Create supervisor agent
supervisor = SupervisorAgent(
    provider=AnthropicProvider(model="claude-opus-4-6"),
    workers=[researcher, analyst, writer],
    strategy="adaptive",  # Options: "sequential", "parallel", "adaptive"
    max_delegations=10,   # Prevent infinite loops
)

# Execute via pyworkflow
from pyworkflow import workflow

@workflow(durable=True)
async def research_report_workflow(topic: str):
    result = await supervisor.run(
        task=f"Create a comprehensive research report on {topic}",
        context={"format": "markdown", "word_count": 2000}
    )
    return result

Key Mapping to PyWorkflow Primitives:

• Supervisor = Main workflow function
• Worker invocation = Steps dispatched to Celery workers via @step decorator
• Routing decisions = Recorded as AGENT_ROUTING events
• Worker execution = Each worker runs as an isolated step on the steps queue
• Result aggregation = Standard workflow return value

Event Types

New events for supervisor pattern:

class EventType(str, Enum):
    # Existing events...
    AGENT_ROUTING = "agent_routing"           # Supervisor decides which agent to invoke
    AGENT_DELEGATION = "agent_delegation"     # Task delegated to worker
    AGENT_COMPLETED = "agent_completed"       # Worker finished task
    AGENT_SYNTHESIS = "agent_synthesis"       # Supervisor synthesizes results

Event Data Schema:

# AGENT_ROUTING
{
    "supervisor_id": "supervisor_main",
    "selected_agent": "researcher",
    "reasoning": "Task requires web search capability",
    "routing_strategy": "adaptive"
}

# AGENT_DELEGATION
{
    "supervisor_id": "supervisor_main",
    "worker_id": "researcher",
    "task": "Search for latest AI research papers",
    "step_id": "step_researcher_001"
}

# AGENT_COMPLETED
{
    "worker_id": "researcher",
    "result": {...},
    "execution_time_ms": 3450
}

Trade-offs

Pros

• Clear orchestration: Centralized decision-making is easier to reason about
• Specialization: Each worker focuses on specific capabilities
• Debuggability: Event log shows all routing decisions
• Replay-safe: Supervisor decisions are deterministic during event replay
• Scalable workers: PyWorkflow's Celery integration allows horizontal scaling of worker steps

Cons

• Routing overhead: Extra LLM calls for supervisor decision-making (~40% slower than peer handoffs)
• Single point of failure: If supervisor logic is flawed, entire workflow suffers
• Context limitations: Supervisor must fit all worker context in its prompt
• Cost: Additional LLM invocations for orchestration vs direct execution

When to Use

• Tasks require multiple distinct specializations
• Workflow requires adaptive routing based on intermediate results
• Debugging and observability are critical
• Team collaboration benefits from centralized control

When to Avoid

• Simple linear workflows (overhead not justified)
• Cost-sensitive applications (prefer swarm handoffs)
• Low-latency requirements (routing adds delay)

Implementation Checklist

• Create SupervisorAgent class in pyworkflow_agents/supervisor.py
• Implement routing strategies: sequential, parallel, adaptive
• Add AGENT_ROUTING, AGENT_DELEGATION, AGENT_COMPLETED event types
• Integrate with PyWorkflow's @step decorator for worker execution
• Add event replay support for supervisor decisions (deterministic routing)
• Implement max_delegations limit to prevent infinite loops
• Add supervisor state to WorkflowContext
• Create examples in examples/agents/supervisor_pattern.py
• Add tests for all routing strategies
• Document supervisor configuration options

Related Issues

• #TBD - Swarm Agent (Peer Handoffs) - Alternative to centralized orchestration
• #TBD - Hierarchical Multi-Agent - Nested supervisors for complex workflows
• #TBD - Parallel Agent (Scatter-Gather) - Can be used as a routing strategy

References

• LangGraph Supervisor Announcement
• Multi-Agent Workflows Tutorial
• LangGraph Personal Assistant with Subagents
• CrewAI vs LangGraph Comparison
• Multi-Agent System Tutorial with LangGraph