Hierarchical Agent System

Overview

Multi-level agent hierarchy with delegation and reporting. This guide covers architecture, implementation, and production deployment of AI agents.

Agent Architecture

User Input
    ↓
Agent Orchestrator
    ↓
┌───────────────┐
│ Planning      │ ← LLM decides what to do
│ Tool Selection│ ← Choose appropriate tool
│ Execution     │ ← Run the tool
│ Observation   │ ← Process tool result
│ Reflection    │ ← Evaluate and iterate
└───────────────┘
    ↓
Final Response

Implementation

python
from openai import OpenAI
from typing import Callable, Any
import json
client = OpenAI()
Define agent tools
def example_tool(param: str) -> str:
    """Example tool for hierarchy."""
    # Implement actual tool logic
    return f"Tool result for: {param}"
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "example_tool",
            "description": "Performs hierarchy operations",
            "parameters": {
                "type": "object",
                "properties": {
                    "param": {
                        "type": "string",
                        "description": "Input parameter"
                    }
                },
                "required": ["param"]
            }
        }
    }
]
TOOL_FNS = {"example_tool": example_tool}
class HierarchicalAgentSystem:
    """Hierarchical Agent System
    
    Multi-level agent hierarchy with delegation and reporting
    """
    
    def __init__(self, model: str = "gpt-4o", max_iterations: int = 10):
        self.client = OpenAI()
        self.model = model
        self.max_iterations = max_iterations
        self.system = f"""You are an autonomous AI agent specialized in {specialty}.
        
You have access to tools to help complete tasks.
Always:
Plan your approach before executing
Use tools systematically
Verify results before proceeding
Handle errors gracefully
Report progress clearly"""
    
    def run(self, task: str, context: dict = None) -> dict:
        """Execute agent task with tool calling loop."""
        
        messages = [{"role": "system", "content": self.system}]
        
        if context:
            messages.append({
                "role": "user",
                "content": f"Context:\n{json.dumps(context, indent=2)}"
            })
        
        messages.append({"role": "user", "content": f"Task: {task}"})
        
        iterations = 0
        tool_calls_made = []
        
        while iterations < self.max_iterations:
            iterations += 1
            
            # Get agent response
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=TOOLS,
                tool_choice="auto"
            )
            
            msg = response.choices[0].message
            messages.append(msg.model_dump())
            
            # Check if done
            if response.choices[0].finish_reason == "stop":
                return {
                    "result": msg.content,
                    "iterations": iterations,
                    "tool_calls": tool_calls_made,
                    "success": True
                }
            
            # Execute tool calls
            if msg.tool_calls:
                for tc in msg.tool_calls:
                    tool_name = tc.function.name
                    tool_args = json.loads(tc.function.arguments)
                    
                    # Execute tool
                    if tool_name in TOOL_FNS:
                        tool_result = TOOL_FNStool_name
                    else:
                        tool_result = f"Error: Unknown tool {tool_name}"
                    
                    tool_calls_made.append({
                        "tool": tool_name,
                        "args": tool_args,
                        "result": str(tool_result)[:500]
                    })
                    
                    # Add tool result to messages
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": str(tool_result)
                    })
        
        return {
            "result": "Max iterations reached",
            "iterations": iterations,
            "tool_calls": tool_calls_made,
            "success": False
        }
Usage
agent = HierarchicalAgentSystem()
result = agent.run("Complete a hierarchy task: [describe your specific task]")
print(f"Result: {result['result']}")
print(f"Iterations: {result['iterations']}")
print(f"Tools used: {len(result['tool_calls'])}")

Adding Custom Tools

python
def create_tool_spec(name: str, description: str, params: dict) -> dict:
    """Helper to create OpenAI tool spec."""
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": params,
                "required": list(params.keys())
            }
        }
    }
Example: Adding a search tool
def web_search(query: str) -> str:
    """Search the web for information."""
    # Implement with your preferred search API
    return f"Search results for: {query}"
Add to TOOLS and TOOL_FNS
TOOLS.append(create_tool_spec(
    "web_search",
    "Search the web for current information",
    {"query": {"type": "string", "description": "Search query"}}
))
TOOL_FNS["web_search"] = web_search

Error Handling and Resilience

python
import time
class ResilientAgent(HierarchicalAgentSystem):
    """Agent with error recovery."""
    
    def safe_run(self, task: str, retries: int = 3) -> dict:
        """Run with automatic retry on failure."""
        for attempt in range(retries):
            try:
                result = self.run(task)
                if result["success"]:
                    return result
            except Exception as e:
                if attempt == retries - 1:
                    return {"result": f"Failed: {e}", "success": False}
                wait = 2 ** attempt
                print(f"Attempt {attempt+1} failed, retrying in {wait}s...")
                time.sleep(wait)
        return {"result": "Max retries exceeded", "success": False}

Best Practices

Limit iterations — prevent infinite loops with max_iterations

Log everything — agent actions need full audit trail

Sandbox tools — never give agents unrestricted system access

Human-in-the-loop — add approval steps for high-stakes actions

Test adversarially — verify agent handles unexpected inputs

Resources

OpenAI Function Calling: https://platform.openai.com/docs/guides/function-calling

LangGraph docs: https://langchain-ai.github.io/langgraph

Agent design patterns: https://www.anthropic.com/research/building-effective-agents