How to Orchestrate Autonomous Sub-Agents Without Blowing Your LLM Context Window
Source: Dev.to
We have all hit the “monolithic LLM wall.”
You design an incredibly capable AI agent, arm it with a suite of tools, and give it a complex, multi-step task—like writing a comprehensive technical paper complete with data analysis, web research, and code verification. At first, it works beautifully. But as the steps accumulate, the context window fills up. The agent begins to experience “attention drift.” It forgets its original instructions, hallucinates tool outputs, and eventually spins out of control, burning through millions of tokens and your API budget.
The problem isn’t the LLM’s reasoning capacity; it’s the architecture. Trying to solve a complex, multi-domain problem within a single agent’s context window is the modern software equivalent of writing an entire enterprise application inside a single, monolithic main() function.
To build AI systems that can scale to handle real-world complexity, we must transition from monolithic agents to hierarchical multi-agent orchestration.
By decomposing complex goals into isolated, specialized sub-agents—each operating within its own bounded context and resource budget—we can build resilient, self-improving AI systems that scale indefinitely.
In this post, we will dive deep into the architectural patterns of multi-agent orchestration, explore how to manage agent lifecycles, and write production-grade Python code to spawn and supervise sub-agents.
(The concepts and code demonstrated here are drawn from my ebook Hermes Agent, The Self-Evolving AI Workforce)
- The Core Concept: Hierarchical Decomposition and Supervisory Control
Multi-agent orchestration is not just a design convenience; it is an architectural necessity. The theoretical foundation of this approach rests on two pillars: task decomposition and supervisory control. Together, they transform a monolithic agent into a scalable, resilient hierarchy of specialized workers.
The Master Carpenter Analogy
Think of a master carpenter building a custom cabinet. The master does not personally cut every dovetail, sand every surface, or install every hinge. Instead, she decomposes the project into distinct sub-tasks: joinery, finishing, and hardware installation.
For each sub-task, she assigns an apprentice with the right tools and expertise. She monitors their progress, checks their quality, and integrates their individual outputs into the final product. If an apprentice hits a snag, she intervenes, provides guidance, or reassigns resources.
In this scenario, the parent agent is the master carpenter, and the sub-agents are the apprentices. Each apprentice operates with their own focused toolset and an independent iteration budget.
+------------------+
| Parent Agent | self.limit:
raise TimeoutError("Iteration budget exceeded!")
class AIAgent:
def __init__(self, **kwargs):
self.config = kwargs
self.session_id = kwargs.get("session_id")
self.budget = IterationBudget(kwargs.get("max_iterations", 50))
async def run_conversation(self, prompt: str) -> Dict[str, Any]:
# Simulate agent execution and tool calling
await asyncio.sleep(1)
self.budget.consume(5) # Simulate consuming 5 iterations of reasoning
return {
"status": "success",
"output": f"Processed prompt: '{prompt}' using model {self.config.get('model')}",
"iterations_used": self.budget.used
}
class SessionDB:
def __init__(self, db_path: Path):
self.db_path = db_path
self.db_path.mkdir(parents=True, exist_ok=True)
self.sessions_file = self.db_path / "sessions.json"
if not self.sessions_file.exists():
self.sessions_file.write_text("{}")
def ensure_tables(self):
# In a real SQL database, this would execute CREATE TABLE statements
pass
def upsert_session(self, session_id: str, metadata: Dict[str, Any]):
data = json.loads(self.sessions_file.read_text())
data[session_id] = metadata
self.sessions_file.write_text(json.dumps(data, indent=4))
print(f"💾 Session '{session_id}' persisted to database.")
def get_hermes_home() -> Path:
home = Path.home() / ".hermes"
home.mkdir(exist_ok=True)
return home
# Setup Logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("MultiAgentOrchestrator")
# ---------------------------------------------------------------------------
# Step 1: Parent Agent Supervisor Configuration
# ---------------------------------------------------------------------------
parent_config = {
"base_url": "https://api.openai.com/v1",
"api_key": "sk-mock-key",
"model": "gpt-4o",
"provider": "openai",
"api_mode": "chat",
"max_iterations": 90, # Parent gets a generous budget
"tool_delay": 1.0, # Rate-limiting safety delay
"enabled_toolsets": ["filesystem", "web", "terminal", "code_execution"],
"save_trajectories": True,
"session_id": "supervisor_session_101",
}
# Initialize Parent Agent
parent_agent = AIAgent(
base_url=parent_config["base_url"],
api_key=parent_config["api_key"],
model=parent_config["model"],
provider=parent_config["provider"],
api_mode=parent_config["api_mode"],
max_iterations=parent_config["max_iterations"],
tool_delay=parent_config["tool_delay"],
enabled_toolsets=parent_config["enabled_toolsets"],
save_trajectories=parent_config["save_trajectories"],
session_id=parent_config["session_id"],
)
logger.info(f"Supervisor Agent Initialized. Model: {parent_config['model']} | Session: {parent_config['session_id']}")
# ---------------------------------------------------------------------------
# Step 2: Initialize Persistent Session Storage
# ---------------------------------------------------------------------------
hermes_home = get_hermes_home()
session_db = SessionDB(db_path=hermes_home / "sessions")
session_db.ensure_tables()
# Register parent session in DB
session_db.upsert_session(
session_id=parent_config["session_id"],
metadata={
"role": "supervisor",
"model": parent_config["model"],
"max_iterations": parent_config["max_iterations"],
"status": "active"
}
)
# ---------------------------------------------------------------------------
# Step 3: Sub-Agent Spawner Configuration & Lifecycle Management
# ---------------------------------------------------------------------------
SUB_AGENT_MODEL = "gpt-4-mini" # Using a faster, cheaper model for sub-agents
SUB_AGENT_MAX_ITERATIONS = 50 # Capped iteration budget for safety
def build_sub_agent_config(task_slug: str, specialized_tools: List[str]) -> dict:
"""
Generates a tailored configuration for a specialized sub-agent.
"""
sub_session_id = f"{parent_config['session_id']}_sub_{task_slug}"
return {
"base_url": parent_config["base_url"],
"api_key": parent_config["api_key"],
"model": SUB_AGENT_MODEL,
"provider": parent_config["provider"],
"api_mode": "chat",
"max_iterations": SUB_AGENT_MAX_ITERATIONS,
"tool_delay": 0.5,
"enabled_toolsets": specialized_tools, # Restrict tools to only what is needed!
"save_trajectories": True,
"session_id": sub_session_id,
}
async def orchestrate_sub_task(task_name: str, prompt: str, tools: List[str]) -> Dict[str, Any]:
"""
Spawns, executes, tracks, and terminates a sub-agent.
"""
logger.info(f"🚀 Spawning sub-agent for task: [{task_name}]")
# Generate configuration
sub_config = build_sub_agent_config(task_name, tools)
# Persist sub-agent creation to database
session_db.upsert_session(
session_id=sub_config["session_id"],
metadata={
"role": f"worker_{task_name}",
"parent_session_id": parent_config["session_id"],
"model": sub_config["model"],
"max_iterations": sub_config["max_iterations"],
"status": "spawned"
}
)
# Instantiate Sub-Agent
sub_agent = AIAgent(**sub_config)
try:
# Execute Task (Delegation Phase)
logger.info(f"Delegating task to sub-agent [{sub_config['session_id']}]...")
result = await sub_agent.run_conversation(prompt)
# Update Status to Success
session_db.upsert_session(
session_id=sub_config["session_id"],
metadata={"status": "completed", "iterations_used": result["iterations_used"]}
)
logger.info(f"✅ Sub-agent [{task_name}] completed successfully.")
return result
except Exception as e:
logger.error(f"❌ Sub-agent [{task_name}] failed: {str(e)}")
session_db.upsert_session(
session_id=sub_config["session_id"],
metadata={"status": "failed", "error": str(e)}
)
raise e
finally:
# Resource Cleanup Phase
logger.info(f"🧹 Terminating sub-agent [{sub_config['session_id']}] and cleaning up resources.")
# In a production system, you would call:
# sub_agent.cleanup_browser()
# sub_agent.cleanup_vm()
# ---------------------------------------------------------------------------
# Step 4: Run Orchestration Loop
# ---------------------------------------------------------------------------
async def main():
print("\n--- Starting Multi-Agent Orchestration Demo ---\n")
# Define specialized sub-tasks
tasks = [
{
"name": "research",
"prompt": "Search the web for the latest advancements in solid-state batteries.",
"tools": ["web"]
},
{
"name": "analysis",
"prompt": "Analyze the research data and generate a Python script to model efficiency curves.",
"tools": ["filesystem", "code_execution"]
}
]
# Execute sub-agents sequentially (can be parallelized using asyncio.gather)
for task in tasks:
try:
result = await orchestrate_sub_task(
task_name=task["name"],
prompt=task["prompt"],
tools=task["tools"]
)
print(f"Result Output: {result['output']}\n")
except Exception:
print(f"Skipping downstream tasks due to failure in task: {task['name']}")
if __name__ == "__main__":
asyncio.run(main())
Enter fullscreen mode
Exit fullscreen mode7. Key Architectural Takeaways
If you are designing a multi-agent system, keep these core architectural principles in mind:
Strict Tool Isolation: Never give a sub-agent more tools than it needs. A web-searching agent does not need write access to your terminal; a code-execution agent does not need access to your browser. Limiting tools dramatically reduces security risks and prompt confusion.
Independent Budgets: Always cap your sub-agents’ iteration budgets below the parent’s budget. If a parent has 90 iterations, its sub-agents should be capped at 30 or 50. This ensures the parent always retains enough budget to handle failures and synthesize the final results.
Persistent State vs. Ephemeral Context: Keep your LLM context windows ephemeral. Use a persistent, file-based database or shared folder to write intermediate data, and only pass highly compressed summaries back into the active context.
Let’s Discuss
How do you handle error recovery in your multi-agent systems? If a critical sub-agent fails or runs out of budget, do you prefer to have the parent agent retry with a modified prompt, or do you escalate the failure directly to the human-in-the-loop?
What are your thoughts on budget refunds for programmatic tools? Do you agree that pure code execution shouldn’t count against an agent’s reasoning budget, or does that open the door to unmonitored resource consumption?
Leave a comment below with your experiences, and let’s build more resilient AI systems together!
The concepts and code demonstrated here are drawn directly from the comprehensive roadmap laid out in the ebook Hermes Agent, The Self-Evolving AI Workforce: details link, you can find also my programming ebooks with AI here: Programming & AI eBooks.