How to Orchestrate Autonomous Sub-Agents Without Blowing Your LLM Context Window

                   +------------------+
                   |   Parent Agent   |  <-- Master Carpenter (Supervisor)
                   +--------+---------+
                            |
         +------------------+------------------+
         |                  |                  |
+--------v-------+ +--------v-------+ +--------v-------+
|  Sub-Agent A   | |  Sub-Agent B   | |  Sub-Agent C   |  <-- Apprentices (Workers)
| (Web Searcher) | | (Code Builder) | | (Doc Writer)  |
+----------------+ +----------------+ +----------------+

class IterationBudget:
    """Thread-safe iteration counter for an agent.

    Each agent (parent or subagent) gets its own IterationBudget.
    The parent's budget is capped at max_iterations (default 90).
    Each subagent gets an independent budget capped at
    delegation.max_iterations (default 50) — this means total
    iterations across parent + subagents can exceed the parent's cap.
    """

+-------------------------------------------------------+
|                  PERSISTENT BLACKBOARD                |
|               (Shared File-Based Memory)              |
+---------------------------^---------------------------+
                            |
         +------------------+------------------+
         |                  |                  |
+--------v-------+ +--------v-------+ +--------v-------+
|  Sub-Agent A   | |  Sub-Agent B   | |  Sub-Agent C   |
| Writes Search  | | Reads Search   | | Reads Code     |
| Results        | | Writes Code    | | Writes Final   |
|                | | Artifacts      | | Report         |
+----------------+ +----------------+ +----------------+

#!/usr/bin/env python3
"""
Production-Grade Parent-Agent Supervisor and Sub-Agent Spawner.
"""
import logging
import asyncio
import json
from typing import Dict, List, Any, Optional
from pathlib import Path

# Mocking the imports from the Hermes framework for demonstration
# In a real environment, these are imported from your agent library
class IterationBudget:
    def __init__(self, limit: int):
        self.limit = limit
        self.used = 0

    def consume(self, amount: int = 1):
        self.used += amount
        if self.used > 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())