Subagents

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