Subagents: Agent Orchestration & Scaling

Overview

Subagents are smaller agents orchestrated by a parent agent. This enables solving complex problems through delegation and coordination.

---

Subagent Pattern

Parent Delegates to Children

class ParentAgent:
    """Agent that manages subagents"""

    def __init__(self):
        self.subagents = {
            'researcher': ResearchSubagent(),
            'analyzer': AnalysisSubagent(),
            'writer': WritingSubagent()
        }

    def solve_complex_task(self, task: str):
        """Break task into subagent work"""

        # Step 1: Research phase
        research_task = self.formulate_research(task)
        research_result = self.subagents['researcher'].run(research_task)

        # Step 2: Analysis phase
        analysis_task = self.formulate_analysis(task, research_result)
        analysis_result = self.subagents['analyzer'].run(analysis_task)

        # Step 3: Writing phase
        writing_task = self.formulate_writing(task, analysis_result)
        final_result = self.subagents['writer'].run(writing_task)

        return final_result


class ResearchSubagent:
    """Specialized research agent"""

    def __init__(self):
        self.tools = [WebSearchTool(), PaperAnalyzer()]
        self.specialization = "research"

    def run(self, task: str):
        """Research a topic"""

        # Focused on research
        # Uses research-specific tools

        findings = self.search(task)
        analyzed = self.analyze_sources(findings)

        return {
            'task': task,
            'findings': findings,
            'analysis': analyzed
        }

---

Parallel Subagents

Concurrent Execution

import asyncio

class ParallelOrchestration:
    """Run multiple subagents in parallel"""

    def __init__(self):
        self.agents = {
            'market_research': MarketResearchAgent(),
            'competitive_analysis': CompetitiveAnalysisAgent(),
            'technical_assessment': TechnicalAssessmentAgent()
        }

    async def run_parallel(self, task: str):
        """Execute agents concurrently"""

        tasks = [
            self.agents['market_research'].run_async(task),
            self.agents['competitive_analysis'].run_async(task),
            self.agents['technical_assessment'].run_async(task)
        ]

        # Wait for all to complete
        results = await asyncio.gather(*tasks)

        # Aggregate results
        return self.aggregate_results(results)

    def aggregate_results(self, results):
        """Combine parallel results"""

        return {
            'market': results[0],
            'competitive': results[1],
            'technical': results[2],
            'consensus': self.find_consensus(results)
        }

---

Hierarchical Agents

Multi-Level Orchestration

class HierarchicalOrchestration:
    """Tree of agents"""

    class Agent:
        def __init__(self, name, level):
            self.name = name
            self.level = level
            self.children = []  # Subagents

        def add_child(self, agent):
            self.children.append(agent)

    def __init__(self):
        # Create hierarchy
        self.root = self.Agent("root", 0)

        # Level 1
        analysis = self.Agent("analysis", 1)
        synthesis = self.Agent("synthesis", 1)

        self.root.add_child(analysis)
        self.root.add_child(synthesis)

        # Level 2 (under analysis)
        market = self.Agent("market", 2)
        tech = self.Agent("tech", 2)

        analysis.add_child(market)
        analysis.add_child(tech)

    def solve(self, task):
        """Solve using hierarchy"""

        # Recursively solve
        return self.solve_recursive(self.root, task)

    def solve_recursive(self, agent, task):
        """Recursively delegate"""

        if not agent.children:
            # Leaf node - execute
            return agent.execute(task)

        # Has children - delegate
        results = {}

        for child in agent.children:
            results[child.name] = self.solve_recursive(child, task)

        # Aggregate
        return agent.synthesize_results(results)

---

Error Handling in Hierarchies

Cascading Failures

class ResilientOrchestration:
    """Handle subagent failures"""

    def run_with_fallback(self, task: str):
        """Execute with fallback strategy"""

        # Try primary agent
        try:
            result = self.primary_agent.run(task)
            return result
        except Exception as e:
            # Failed! Try alternative

            # Option 1: Retry with different agent
            try:
                result = self.fallback_agent.run(task)
                self.log_failover(task, self.primary_agent, self.fallback_agent)
                return result
            except Exception as e2:
                # Both failed
                self.alert_human(task, [e, e2])
                raise

    def monitor_subagent_health(self):
        """Detect failures early"""

        for name, agent in self.subagents.items():
            health = agent.check_health()

            if not health['ok']:
                # Subagent unhealthy
                self.remove_from_pool(name)
                self.spawn_replacement()

---

3 Warnings ⚠️

Warning 1: Communication Overhead

# ❌ WRONG
# Subagents communicate constantly
subagent_1.notify_all_agents(update)
subagent_2.notify_all_agents(update)
# Bottleneck in communication

# ✅ RIGHT
# Parent orchestrates communication
parent.collect_results_from_subagents()
parent.distribute_next_tasks()
# Cleaner coordination

Warning 2: State Inconsistency

# ❌ WRONG
# Subagents have conflicting state
subagent_1.set_value('budget', 100)
subagent_2.set_value('budget', 200)
# Inconsistent

# ✅ RIGHT
# Parent manages shared state
state = {'budget': 100}
subagent_1.run_with_state(state)
subagent_2.run_with_state(state)

Warning 3: Cascading Errors

# ❌ WRONG
# Error in subagent crashes parent
result = subagent.run()  # Fails
parent.continue()  # Also fails!

# ✅ RIGHT
# Isolate errors
try:
    result = subagent.run()
except Exception as e:
    result = get_fallback()
    log_error(e)

parent.continue(result)

---

Last Updated: August 9, 2026