Subagents and Agent Hierarchies

The expedition grows: someone scouts the route, someone counts supplies, someone checks permits, someone watches quality. When each of them talks to everyone else, after a week nobody knows who agreed on what. You need an expedition leader who hands out tasks and collects reports.

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

Subagent Architecture

The diagram shows three levels of the hierarchy, from the leader down to the tools:

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└─────────────────────────────────────────────────────────────┘

At the top the orchestrator, which coordinates the work and delegates tasks. Below it the specialist subagents, and at the bottom the tools they use. The orchestrator never reaches for tools directly.

Basic Subagent Implementation

First the data types: roles as an Enum, and the task and result as a @dataclass:

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

SubagentTask carries a description, context and expected output, everything a specialist needs so it does not have to ask the leader for details.

The base Subagent class inherits from ABC and enforces system_prompt with @abstractmethod, just like BaseAgent from the lesson on multi-agent systems:

1class Subagent(ABC):
2    """Base subagent class."""
3
4    def __init__(self, name: str, role: SubagentRole, model: str = "gpt-4o-mini"):
5        self.name = name
6        self.role = role
7        self.model = model
8        self.tools: list[dict] = []
9        self.memory: list[str] = []
10
11    @property
12    @abstractmethod
13    def system_prompt(self) -> str:
14        """Subagent's system prompt."""
15        pass
16
17    def add_tool(self, tool: dict) -> None:
18        """Adds a tool to the subagent."""
19        self.tools.append(tool)

Every subagent has its own tool list and memory. add_tool appends a tool definition in the OpenAI format, and memory collects summaries of completed tasks.

The execute method builds a prompt from the three task fields and calls the model:

1    async def execute(self, task: SubagentTask) -> SubagentResult:
2        """Executes a task."""
3        messages = [
4            {"role": "system", "content": self.system_prompt},
5            {"role": "user", "content": f"""
6Task: {task.description}
7Context: {task.context}
8Expected output: {task.expected_output}
9"""}
10        ]
11
12        response = client.chat.completions.create(
13            model=self.model,
14            messages=messages,
15            tools=self.tools if self.tools else None
16        )
17
18        output = response.choices[0].message.content
19        self.memory.append(f"Task: {task.description} -> Result: {output[:100]}...")
20
21        return SubagentResult(
22            success=True,
23            output=output,
24            metadata={"model": self.model, "role": self.role.value}
25        )

The method is async, but it uses a synchronous client, so it blocks the event loop. In production use AsyncOpenAI and await. Watch out for tools too: when the model calls a tool instead of answering with text, content is None and output[:100] raises an error.

Specialized Subagents

As before, the specialists differ only in their prompt. The researcher and the analyst:

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"""

Each prompt says what the subagent does and what it must always watch out for. The "Always" section works like camp rules for that specialist.

The executor and the validator complete the team:

1class ExecutorSubagent(Subagent):
2    """Executor subagent - carries out tasks."""
3
4    @property
5    def system_prompt(self) -> str:
6        return """You are a specialized executor agent.
7
8Your tasks:
91. Carry out planned actions
102. Use tools to complete tasks
113. Report progress
124. Handle errors
13
14Always:
15- Execute tasks step by step
16- Verify the results of each action
17- Report problems immediately"""
18
19
20class ValidatorSubagent(Subagent):
21    """Validator subagent - checks quality."""
22
23    @property
24    def system_prompt(self) -> str:
25        return """You are a specialized validation agent.
26
27Your tasks:
281. Verify result correctness
292. Check compliance with requirements
303. Identify errors and shortcomings
314. Suggest corrections
32
33Always:
34- Be critical but constructive
35- Use specific criteria
36- Propose solutions to problems"""

The validator reviews the others' results, so a mistake by one specialist has a chance of being caught.

Orchestrator - Managing Subagents

The orchestrator keeps a registry of subagents and a task history:

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        ])

get_available_subagents builds a text list of specialists, which we will show the model in a moment.

The orchestrator's steps always go in this order: plan_execution, choosing a subagent, subagent.execute and synthesize_results. Planning asks the model for a plan in JSON:

1    async def plan_execution(self, task: str) -> List[SubagentTask]:
2        """Plans task execution through appropriate subagents."""
3        planning_prompt = f"""You have a task to execute: {task}
4
5Available subagents:
6{self.get_available_subagents()}
7
8Plan the task execution. For each step specify:
91. Which subagent should execute the step
102. Task description for the subagent
113. Expected output
124. Priority (1-5)
13
14Reply in JSON format:
15{{"steps": [{{"subagent": "name", "task": "description", "expected_output": "...", "priority": 1}}, ...]}}"""
16
17        response = client.chat.completions.create(
18            model=self.model,
19            messages=[
20                {"role": "system", "content": "You are a task planner. Reply only in JSON."},
21                {"role": "user", "content": planning_prompt}
22            ],
23            response_format={"type": "json_object"}
24        )
25
26        import json
27        plan = json.loads(response.choices[0].message.content)
28
29        tasks = []
30        for step in plan.get("steps", plan):
31            tasks.append(SubagentTask(
32                description=step["task"],
33                context=task,
34                expected_output=step["expected_output"],
35                priority=step.get("priority", 1)
36            ))
37
38        return tasks

The response_format={"type": "json_object"} mode guarantees a valid JSON object, but not an array. That is why the prompt asks for an object with a steps key, and the code reads plan.get("steps", plan).

execute_task ties planning, execution and synthesis together:

1    async def execute_task(self, task: str) -> str:
2        """Executes a task using subagents."""
3        # 1. Planning
4        planned_tasks = await self.plan_execution(task)
5        print(f"Planned {len(planned_tasks)} steps")
6
7        # 2. Execution
8        results = []
9        for i, subtask in enumerate(planned_tasks):
10            # Find the appropriate subagent
11            subagent = self._select_subagent(subtask)
12
13            if subagent:
14                print(f"Step {i+1}: {subagent.name} executing: {subtask.description[:50]}...")
15                result = await subagent.execute(subtask)
16                results.append({
17                    "step": i + 1,
18                    "subagent": subagent.name,
19                    "task": subtask.description,
20                    "result": result.output
21                })
22
23        # 3. Synthesis
24        final_result = await self._synthesize_results(task, results)
25
26        return final_result

The results go into a list of dictionaries, so the synthesis knows which subagent did what.

Choosing a subagent is a simple keyword heuristic:

1    def _select_subagent(self, task: SubagentTask) -> Optional[Subagent]:
2        """Selects the best subagent for the task."""
3        # Simple heuristic - can be extended with ML
4        keywords = {
5            "researcher": ["research", "find", "search", "information"],
6            "analyst": ["analyze", "evaluate", "compare", "conclusions"],
7            "executor": ["execute", "do", "create", "implement"],
8            "validator": ["check", "verify", "validate", "test"]
9        }
10
11        task_lower = task.description.lower()
12
13        for subagent_type, kws in keywords.items():
14            if any(kw in task_lower for kw in kws):
15                for agent in self.subagents.values():
16                    if agent.role.value == subagent_type:
17                        return agent
18
19        # Default to the first available
20        return list(self.subagents.values())[0] if self.subagents else None

The dictionary keys must equal the role values, which is why the first key is "researcher" and not "research", otherwise the researcher would never be chosen. The subagent field from the plan is ignored here, which is worth improving.

Finally the synthesis merges the results into one answer:

1    async def _synthesize_results(self, original_task: str, results: List[dict]) -> str:
2        """Synthesizes results from all subagents."""
3        results_summary = "\n\n".join([
4            f"Step {r['step']} ({r['subagent']}): {r['result']}"
5            for r in results
6        ])
7
8        synthesis_prompt = f"""Original task: {original_task}
9
10Subagent results:
11{results_summary}
12
13Create a coherent, final answer based on the results of all subagents."""
14
15        response = client.chat.completions.create(
16            model=self.model,
17            messages=[
18                {"role": "system", "content": "You are a synthesizer. You combine results into a coherent whole."},
19                {"role": "user", "content": synthesis_prompt}
20            ]
21        )
22
23        return response.choices[0].message.content

This is the third model call in a single task, next to planning and the subagents' work.

Using the Orchestrator

We register four specialists and ask for a Safari plan:

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())

asyncio.run starts the main coroutine, and the orchestrator plans, delegates and assembles the result by itself. Which specialist gets a step depends on the words in the descriptions generated by the planner, not in the task itself.

Multi-Level Hierarchy

A large expedition has team leaders. An orchestrator can have other orchestrators under it:

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)

delegate_task passes the task down when the complexity score exceeds 3.

Helper methods assess complexity and execute the task locally:

1    def _assess_complexity(self, task: str) -> int:
2        """Assesses task complexity (1-5)."""
3        # Simple heuristic
4        complexity_indicators = [
5            "complete", "detailed", "multi-step",
6            "comprehensive", "elaborate"
7        ]
8        return sum(1 for ind in complexity_indicators if ind in task.lower()) + 1
9
10    def _select_sub_orchestrator(self, task: str) -> "HierarchicalOrchestrator":
11        """Selects sub-orchestrator for the task."""
12        # Can be extended with intelligent selection
13        return list(self.sub_orchestrators.values())[0]
14
15    async def _execute_locally(self, task: str) -> str:
16        """Executes task locally."""
17        # Use own subagents
18        results = []
19        for subagent in self.subagents.values():
20            subtask = SubagentTask(
21                description=task,
22                context=f"Level {self.level}",
23                expected_output="Task result"
24            )
25            result = await subagent.execute(subtask)
26            results.append(result.output)
27
28        return "\n".join(results)

Complexity is the number of indicator words plus one, and _select_sub_orchestrator always takes the first team for now.

The example builds two teams under the main orchestrator:

1# Hierarchy usage example
2async def hierarchical_example():
3    # Main orchestrator
4    main_orchestrator = HierarchicalOrchestrator("Main Orchestrator")
5
6    # Sub-orchestrator for research
7    research_orchestrator = HierarchicalOrchestrator("Research Team")
8    research_orchestrator.subagents["researcher"] = ResearchSubagent(
9        "researcher", SubagentRole.RESEARCHER
10    )
11
12    # Sub-orchestrator for analysis
13    analysis_orchestrator = HierarchicalOrchestrator("Analysis Team")
14    analysis_orchestrator.subagents["analyst"] = AnalysisSubagent(
15        "analyst", SubagentRole.ANALYST
16    )
17
18    # Add to hierarchy
19    main_orchestrator.add_sub_orchestrator(research_orchestrator)
20    main_orchestrator.add_sub_orchestrator(analysis_orchestrator)
21
22    # Execute complex task
23    result = await main_orchestrator.delegate_task(
24        "Conduct a complete market research on Safari in East Africa"
25    )
26    print(result)

Spot the trap: the task contains too few indicator words, so it stays with the main orchestrator, which has no subagents of its own and returns empty text. Lower the threshold or give it a subagent.

Communication Between Subagents

Sometimes a specialist needs to ask a colleague. First the message types:

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)

correlation_id links a question with its answer, and field(default_factory=...) gives every message its own timestamp.

A Message Bus is the channel through which agents talk without direct dependencies:

1class MessageBus:
2    """Communication bus for agents."""
3
4    def __init__(self):
5        self.messages: List[AgentMessage] = []
6        self.subscribers: Dict[str, List[callable]] = {}
7
8    def publish(self, message: AgentMessage) -> None:
9        """Publishes a message."""
10        self.messages.append(message)
11
12        # Notify subscribers
13        if message.receiver in self.subscribers:
14            for callback in self.subscribers[message.receiver]:
15                callback(message)
16
17    def subscribe(self, agent_name: str, callback: callable) -> None:
18        """Subscribes an agent to messages."""
19        if agent_name not in self.subscribers:
20            self.subscribers[agent_name] = []
21        self.subscribers[agent_name].append(callback)
22
23    def get_messages_for(self, agent_name: str) -> List[AgentMessage]:
24        """Gets messages for an agent."""
25        return [m for m in self.messages if m.receiver == agent_name]

An agent subscribes to its own name, and publish calls the receiver's callbacks. The sender does not need the receiver's object, only its name, and that is exactly what loose coupling means.

A communicating subagent answers queries and sends messages:

1class CommunicatingSubagent(Subagent):
2    """Subagent with communication capabilities."""
3
4    def __init__(self, name: str, role: SubagentRole, message_bus: MessageBus):
5        super().__init__(name, role)
6        self.message_bus = message_bus
7        self.message_bus.subscribe(name, self._handle_message)
8        self.pending_responses: Dict[str, str] = {}
9
10    def _handle_message(self, message: AgentMessage) -> None:
11        """Handles incoming messages."""
12        if message.message_type == MessageType.QUERY:
13            # Reply to query
14            response = self._process_query(message.content)
15            self.send_message(
16                message.sender,
17                response,
18                MessageType.RESPONSE,
19                correlation_id=message.correlation_id
20            )
21        elif message.message_type == MessageType.RESPONSE:
22            # Save response
23            if message.correlation_id:
24                self.pending_responses[message.correlation_id] = message.content
25
26    def send_message(
27        self,
28        receiver: str,
29        content: str,
30        message_type: MessageType,
31        correlation_id: Optional[str] = None
32    ) -> None:
33        """Sends a message to another agent."""
34        import uuid
35        message = AgentMessage(
36            sender=self.name,
37            receiver=receiver,
38            content=content,
39            message_type=message_type,
40            correlation_id=correlation_id or str(uuid.uuid4())
41        )
42        self.message_bus.publish(message)
43
44    def _process_query(self, query: str) -> str:
45        """Processes a query from another agent."""
46        # Implementation depends on role
47        return f"Response to: {query}"

The answer comes back with the same correlation_id, so the asker knows what it answers.

Asking a question and waiting for the answer:

1    async def ask_other_agent(self, agent_name: str, question: str) -> str:
2        """Asks another agent and waits for a response."""
3        import uuid
4        correlation_id = str(uuid.uuid4())
5
6        self.send_message(
7            agent_name,
8            question,
9            MessageType.QUERY,
10            correlation_id
11        )
12
13        # Wait for response (with timeout)
14        import asyncio
15        for _ in range(10):
16            if correlation_id in self.pending_responses:
17                return self.pending_responses.pop(correlation_id)
18            await asyncio.sleep(0.1)
19
20        return "Timeout - no response"

The bus calls callbacks synchronously, so the answer is usually waiting at the first check, and the limit is one second.

Best Practices for Subagents

The principles of subagent architecture gathered in one place:

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"""

Pay attention to the point about statelessness: our subagents have memory, so when scaling, move it to external storage.

The last element is an error-tolerant orchestrator:

1class RobustOrchestrator(AgentOrchestrator):
2    """Orchestrator with error handling and fallback."""
3
4    async def execute_with_fallback(self, task: str) -> str:
5        """Executes task with error handling."""
6        try:
7            return await self.execute_task(task)
8        except Exception as e:
9            print(f"Main execution error: {e}")
10
11            # Fallback - simpler approach
12            return await self._simple_execution(task)
13
14    async def _simple_execution(self, task: str) -> str:
15        """Simple execution as fallback."""
16        response = client.chat.completions.create(
17            model=self.model,
18            messages=[
19                {"role": "system", "content": "You are a helpful assistant."},
20                {"role": "user", "content": task}
21            ]
22        )
23        return response.choices[0].message.content

When the full pipeline fails, the fallback asks the model directly. I recommend designing the hierarchy flat from the start and adding more levels only when there is a real need.

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

Remember: a good orchestrator is an expedition leader who hands out tasks, collects reports and does not carry every backpack alone.