Subagents and Agent Hierarchies

Subagents are an architectural pattern where a main agent (orchestrator) delegates tasks to specialized sub-agents. It is like a corporate organizational structure - the CEO delegates tasks to managers, who delegate further!

Subagent Architecture

1┌─────────────────────────────────────────────────────────────┐
2│                   Agent Hierarchy                           │
3├─────────────────────────────────────────────────────────────┤
4│                                                             │
5│                    ┌───────────────┐                        │
6│                    │  Orchestrator │                        │
7│                    │   (Main AI)   │                        │
8│                    └───────┬───────┘                        │
9│                            │                                │
10│           ┌────────────────┼────────────────┐               │
11│           │                │                │               │
12│           ▼                ▼                ▼               │
13│    ┌────────────┐   ┌────────────┐   ┌────────────┐        │
14│    │ Research   │   │ Analysis   │   │ Execution  │        │
15│    │ Subagent   │   │ Subagent   │   │ Subagent   │        │
16│    └─────┬──────┘   └─────┬──────┘   └─────┬──────┘        │
17│          │                │                │                │
18│    ┌─────┴─────┐    ┌─────┴─────┐    ┌─────┴─────┐         │
19│    │  Tools    │    │  Tools    │    │  Tools    │         │
20│    └───────────┘    └───────────┘    └───────────┘         │
21│                                                             │
22└─────────────────────────────────────────────────────────────┘

Basic Subagent Implementation

1from openai import OpenAI
2from abc import ABC, abstractmethod
3from dataclasses import dataclass
4from typing import Optional
5from enum import Enum
6
7client = OpenAI()
8
9class SubagentRole(Enum):
10    RESEARCHER = "researcher"
11    ANALYST = "analyst"
12    EXECUTOR = "executor"
13    VALIDATOR = "validator"
14
15@dataclass
16class SubagentTask:
17    """Task for a subagent."""
18    description: str
19    context: str
20    expected_output: str
21    priority: int = 1
22
23@dataclass
24class SubagentResult:
25    """Result of subagent work."""
26    success: bool
27    output: str
28    metadata: dict
29
30class Subagent(ABC):
31    """Base subagent class."""
32
33    def __init__(self, name: str, role: SubagentRole, model: str = "gpt-4o-mini"):
34        self.name = name
35        self.role = role
36        self.model = model
37        self.tools: list[dict] = []
38        self.memory: list[str] = []
39
40    @property
41    @abstractmethod
42    def system_prompt(self) -> str:
43        """Subagent's system prompt."""
44        pass
45
46    def add_tool(self, tool: dict) -> None:
47        """Adds a tool to the subagent."""
48        self.tools.append(tool)
49
50    async def execute(self, task: SubagentTask) -> SubagentResult:
51        """Executes a task."""
52        messages = [
53            {"role": "system", "content": self.system_prompt},
54            {"role": "user", "content": f"""
55Task: {task.description}
56Context: {task.context}
57Expected output: {task.expected_output}
58"""}
59        ]
60
61        response = client.chat.completions.create(
62            model=self.model,
63            messages=messages,
64            tools=self.tools if self.tools else None
65        )
66
67        output = response.choices[0].message.content
68        self.memory.append(f"Task: {task.description} -> Result: {output[:100]}...")
69
70        return SubagentResult(
71            success=True,
72            output=output,
73            metadata={"model": self.model, "role": self.role.value}
74        )

Specialized Subagents

1class ResearchSubagent(Subagent):
2    """Research subagent - gathers information."""
3
4    @property
5    def system_prompt(self) -> str:
6        return """You are a specialized research agent.
7
8Your tasks:
91. Gather information from available sources
102. Verify facts
113. Structure knowledge
124. Identify information gaps
13
14Always:
15- Provide information sources
16- Flag uncertainties
17- Suggest further research directions"""
18
19
20class AnalysisSubagent(Subagent):
21    """Analytical subagent - analyzes data."""
22
23    @property
24    def system_prompt(self) -> str:
25        return """You are a specialized analytical agent.
26
27Your tasks:
281. Analyze provided data
292. Identify patterns and trends
303. Formulate conclusions
314. Assess risks and opportunities
32
33Always:
34- Use numerical data
35- Present alternative interpretations
36- Highlight key discoveries"""
37
38
39class ExecutorSubagent(Subagent):
40    """Executor subagent - carries out tasks."""
41
42    @property
43    def system_prompt(self) -> str:
44        return """You are a specialized executor agent.
45
46Your tasks:
471. Carry out planned actions
482. Use tools to complete tasks
493. Report progress
504. Handle errors
51
52Always:
53- Execute tasks step by step
54- Verify the results of each action
55- Report problems immediately"""
56
57
58class ValidatorSubagent(Subagent):
59    """Validator subagent - checks quality."""
60
61    @property
62    def system_prompt(self) -> str:
63        return """You are a specialized validation agent.
64
65Your tasks:
661. Verify result correctness
672. Check compliance with requirements
683. Identify errors and shortcomings
694. Suggest corrections
70
71Always:
72- Be critical but constructive
73- Use specific criteria
74- Propose solutions to problems"""

Orchestrator - Managing Subagents

1from typing import Dict, List
2import asyncio
3
4class AgentOrchestrator:
5    """Main orchestrator managing subagents."""
6
7    def __init__(self, model: str = "gpt-4o-mini"):
8        self.model = model
9        self.subagents: Dict[str, Subagent] = {}
10        self.task_history: List[dict] = []
11
12    def register_subagent(self, subagent: Subagent) -> None:
13        """Registers a subagent."""
14        self.subagents[subagent.name] = subagent
15        print(f"Registered subagent: {subagent.name} ({subagent.role.value})")
16
17    def get_available_subagents(self) -> str:
18        """Returns a description of available subagents."""
19        return "\n".join([
20            f"- {name}: {agent.role.value}"
21            for name, agent in self.subagents.items()
22        ])
23
24    async def plan_execution(self, task: str) -> List[SubagentTask]:
25        """Plans task execution through appropriate subagents."""
26        planning_prompt = f"""You have a task to execute: {task}
27
28Available subagents:
29{self.get_available_subagents()}
30
31Plan the task execution. For each step specify:
321. Which subagent should execute the step
332. Task description for the subagent
343. Expected output
354. Priority (1-5)
36
37Reply in JSON format:
38[{{"subagent": "name", "task": "description", "expected_output": "...", "priority": 1}}, ...]"""
39
40        response = client.chat.completions.create(
41            model=self.model,
42            messages=[
43                {"role": "system", "content": "You are a task planner. Reply only in JSON."},
44                {"role": "user", "content": planning_prompt}
45            ],
46            response_format={"type": "json_object"}
47        )
48
49        import json
50        plan = json.loads(response.choices[0].message.content)
51
52        tasks = []
53        for step in plan.get("steps", plan):
54            tasks.append(SubagentTask(
55                description=step["task"],
56                context=task,
57                expected_output=step["expected_output"],
58                priority=step.get("priority", 1)
59            ))
60
61        return tasks
62
63    async def execute_task(self, task: str) -> str:
64        """Executes a task using subagents."""
65        # 1. Planning
66        planned_tasks = await self.plan_execution(task)
67        print(f"Planned {len(planned_tasks)} steps")
68
69        # 2. Execution
70        results = []
71        for i, subtask in enumerate(planned_tasks):
72            # Find the appropriate subagent
73            subagent = self._select_subagent(subtask)
74
75            if subagent:
76                print(f"Step {i+1}: {subagent.name} executing: {subtask.description[:50]}...")
77                result = await subagent.execute(subtask)
78                results.append({
79                    "step": i + 1,
80                    "subagent": subagent.name,
81                    "task": subtask.description,
82                    "result": result.output
83                })
84
85        # 3. Synthesis
86        final_result = await self._synthesize_results(task, results)
87
88        return final_result
89
90    def _select_subagent(self, task: SubagentTask) -> Optional[Subagent]:
91        """Selects the best subagent for the task."""
92        # Simple heuristic - can be extended with ML
93        keywords = {
94            "research": ["research", "find", "search", "information"],
95            "analyst": ["analyze", "evaluate", "compare", "conclusions"],
96            "executor": ["execute", "do", "create", "implement"],
97            "validator": ["check", "verify", "validate", "test"]
98        }
99
100        task_lower = task.description.lower()
101
102        for subagent_type, kws in keywords.items():
103            if any(kw in task_lower for kw in kws):
104                for agent in self.subagents.values():
105                    if agent.role.value == subagent_type:
106                        return agent
107
108        # Default to the first available
109        return list(self.subagents.values())[0] if self.subagents else None
110
111    async def _synthesize_results(self, original_task: str, results: List[dict]) -> str:
112        """Synthesizes results from all subagents."""
113        results_summary = "\n\n".join([
114            f"Step {r['step']} ({r['subagent']}): {r['result']}"
115            for r in results
116        ])
117
118        synthesis_prompt = f"""Original task: {original_task}
119
120Subagent results:
121{results_summary}
122
123Create a coherent, final answer based on the results of all subagents."""
124
125        response = client.chat.completions.create(
126            model=self.model,
127            messages=[
128                {"role": "system", "content": "You are a synthesizer. You combine results into a coherent whole."},
129                {"role": "user", "content": synthesis_prompt}
130            ]
131        )
132
133        return response.choices[0].message.content

Using the Orchestrator

1import asyncio
2
3async def main():
4    # Create orchestrator
5    orchestrator = AgentOrchestrator()
6
7    # Register subagents
8    orchestrator.register_subagent(
9        ResearchSubagent("safari_researcher", SubagentRole.RESEARCHER)
10    )
11    orchestrator.register_subagent(
12        AnalysisSubagent("data_analyst", SubagentRole.ANALYST)
13    )
14    orchestrator.register_subagent(
15        ExecutorSubagent("task_executor", SubagentRole.EXECUTOR)
16    )
17    orchestrator.register_subagent(
18        ValidatorSubagent("quality_checker", SubagentRole.VALIDATOR)
19    )
20
21    # Execute a complex task
22    result = await orchestrator.execute_task(
23        "Prepare a complete 7-day Safari plan in the Serengeti for 4 people"
24    )
25
26    print("\n" + "="*50)
27    print("FINAL RESULT:")
28    print("="*50)
29    print(result)
30
31asyncio.run(main())

Multi-Level Hierarchy

1class HierarchicalOrchestrator:
2    """Orchestrator with multi-level hierarchy."""
3
4    def __init__(self, name: str, level: int = 0):
5        self.name = name
6        self.level = level
7        self.subagents: Dict[str, Subagent] = {}
8        self.sub_orchestrators: Dict[str, "HierarchicalOrchestrator"] = {}
9
10    def add_sub_orchestrator(self, orchestrator: "HierarchicalOrchestrator") -> None:
11        """Adds a sub-orchestrator (for deeper hierarchy)."""
12        orchestrator.level = self.level + 1
13        self.sub_orchestrators[orchestrator.name] = orchestrator
14
15    async def delegate_task(self, task: str) -> str:
16        """Delegates task to the appropriate hierarchy level."""
17        # Check if task requires a sub-orchestrator
18        complexity = self._assess_complexity(task)
19
20        if complexity > 3 and self.sub_orchestrators:
21            # Delegate to sub-orchestrator
22            sub_orch = self._select_sub_orchestrator(task)
23            return await sub_orch.delegate_task(task)
24        else:
25            # Execute through own subagents
26            return await self._execute_locally(task)
27
28    def _assess_complexity(self, task: str) -> int:
29        """Assesses task complexity (1-5)."""
30        # Simple heuristic
31        complexity_indicators = [
32            "complete", "detailed", "multi-step",
33            "comprehensive", "elaborate"
34        ]
35        return sum(1 for ind in complexity_indicators if ind in task.lower()) + 1
36
37    def _select_sub_orchestrator(self, task: str) -> "HierarchicalOrchestrator":
38        """Selects sub-orchestrator for the task."""
39        # Can be extended with intelligent selection
40        return list(self.sub_orchestrators.values())[0]
41
42    async def _execute_locally(self, task: str) -> str:
43        """Executes task locally."""
44        # Use own subagents
45        results = []
46        for subagent in self.subagents.values():
47            subtask = SubagentTask(
48                description=task,
49                context=f"Level {self.level}",
50                expected_output="Task result"
51            )
52            result = await subagent.execute(subtask)
53            results.append(result.output)
54
55        return "\n".join(results)
56
57
58# Hierarchy usage example
59async def hierarchical_example():
60    # Main orchestrator
61    main_orchestrator = HierarchicalOrchestrator("Main Orchestrator")
62
63    # Sub-orchestrator for research
64    research_orchestrator = HierarchicalOrchestrator("Research Team")
65    research_orchestrator.subagents["researcher"] = ResearchSubagent(
66        "researcher", SubagentRole.RESEARCHER
67    )
68
69    # Sub-orchestrator for analysis
70    analysis_orchestrator = HierarchicalOrchestrator("Analysis Team")
71    analysis_orchestrator.subagents["analyst"] = AnalysisSubagent(
72        "analyst", SubagentRole.ANALYST
73    )
74
75    # Add to hierarchy
76    main_orchestrator.add_sub_orchestrator(research_orchestrator)
77    main_orchestrator.add_sub_orchestrator(analysis_orchestrator)
78
79    # Execute complex task
80    result = await main_orchestrator.delegate_task(
81        "Conduct a complete market research on Safari in East Africa"
82    )
83    print(result)

Communication Between Subagents

1from dataclasses import dataclass, field
2from typing import Optional, List
3from datetime import datetime
4from enum import Enum
5
6class MessageType(Enum):
7    TASK = "task"
8    RESULT = "result"
9    QUERY = "query"
10    RESPONSE = "response"
11    ERROR = "error"
12
13@dataclass
14class AgentMessage:
15    """Message between agents."""
16    sender: str
17    receiver: str
18    content: str
19    message_type: MessageType
20    timestamp: datetime = field(default_factory=datetime.now)
21    correlation_id: Optional[str] = None
22    metadata: dict = field(default_factory=dict)
23
24class MessageBus:
25    """Communication bus for agents."""
26
27    def __init__(self):
28        self.messages: List[AgentMessage] = []
29        self.subscribers: Dict[str, List[callable]] = {}
30
31    def publish(self, message: AgentMessage) -> None:
32        """Publishes a message."""
33        self.messages.append(message)
34
35        # Notify subscribers
36        if message.receiver in self.subscribers:
37            for callback in self.subscribers[message.receiver]:
38                callback(message)
39
40    def subscribe(self, agent_name: str, callback: callable) -> None:
41        """Subscribes an agent to messages."""
42        if agent_name not in self.subscribers:
43            self.subscribers[agent_name] = []
44        self.subscribers[agent_name].append(callback)
45
46    def get_messages_for(self, agent_name: str) -> List[AgentMessage]:
47        """Gets messages for an agent."""
48        return [m for m in self.messages if m.receiver == agent_name]
49
50
51class CommunicatingSubagent(Subagent):
52    """Subagent with communication capabilities."""
53
54    def __init__(self, name: str, role: SubagentRole, message_bus: MessageBus):
55        super().__init__(name, role)
56        self.message_bus = message_bus
57        self.message_bus.subscribe(name, self._handle_message)
58        self.pending_responses: Dict[str, str] = {}
59
60    def _handle_message(self, message: AgentMessage) -> None:
61        """Handles incoming messages."""
62        if message.message_type == MessageType.QUERY:
63            # Reply to query
64            response = self._process_query(message.content)
65            self.send_message(
66                message.sender,
67                response,
68                MessageType.RESPONSE,
69                correlation_id=message.correlation_id
70            )
71        elif message.message_type == MessageType.RESPONSE:
72            # Save response
73            if message.correlation_id:
74                self.pending_responses[message.correlation_id] = message.content
75
76    def send_message(
77        self,
78        receiver: str,
79        content: str,
80        message_type: MessageType,
81        correlation_id: Optional[str] = None
82    ) -> None:
83        """Sends a message to another agent."""
84        import uuid
85        message = AgentMessage(
86            sender=self.name,
87            receiver=receiver,
88            content=content,
89            message_type=message_type,
90            correlation_id=correlation_id or str(uuid.uuid4())
91        )
92        self.message_bus.publish(message)
93
94    def _process_query(self, query: str) -> str:
95        """Processes a query from another agent."""
96        # Implementation depends on role
97        return f"Response to: {query}"
98
99    async def ask_other_agent(self, agent_name: str, question: str) -> str:
100        """Asks another agent and waits for a response."""
101        import uuid
102        correlation_id = str(uuid.uuid4())
103
104        self.send_message(
105            agent_name,
106            question,
107            MessageType.QUERY,
108            correlation_id
109        )
110
111        # Wait for response (with timeout)
112        import asyncio
113        for _ in range(10):
114            if correlation_id in self.pending_responses:
115                return self.pending_responses.pop(correlation_id)
116            await asyncio.sleep(0.1)
117
118        return "Timeout - no response"

Best Practices for Subagents

1"""
2Best Practices for subagent architecture:
3
41. SEPARATION OF CONCERNS
5   - Each subagent has one clearly defined responsibility
6   - Avoid "god agents" that do everything
7
82. LOOSE COUPLING
9   - Subagents communicate through message bus
10   - No direct dependencies between subagents
11
123. SINGLE SOURCE OF TRUTH
13   - The orchestrator is the sole source of truth about task state
14   - Subagents report results to the orchestrator
15
164. GRACEFUL DEGRADATION
17   - System works even when one subagent fails
18   - Implement fallback strategies
19
205. OBSERVABILITY
21   - Log all interactions between agents
22   - Implement metrics and monitoring
23
246. SCALABILITY
25   - Subagents should be stateless
26   - Easy to add more instances of the same type
27
287. TESTING
29   - Test subagents in isolation
30   - Test orchestrator integration with subagents
31"""
32
33class RobustOrchestrator(AgentOrchestrator):
34    """Orchestrator with error handling and fallback."""
35
36    async def execute_with_fallback(self, task: str) -> str:
37        """Executes task with error handling."""
38        try:
39            return await self.execute_task(task)
40        except Exception as e:
41            print(f"Main execution error: {e}")
42
43            # Fallback - simpler approach
44            return await self._simple_execution(task)
45
46    async def _simple_execution(self, task: str) -> str:
47        """Simple execution as fallback."""
48        response = client.chat.completions.create(
49            model=self.model,
50            messages=[
51                {"role": "system", "content": "You are a helpful assistant."},
52                {"role": "user", "content": task}
53            ]
54        )
55        return response.choices[0].message.content

Subagents and agent hierarchies are powerful architectural patterns that allow building complex, scalable AI systems. Congratulations - you have learned advanced AI techniques! In the last module you will create a comprehensive project combining all the skills you have acquired!