Agent Orchestration

Agent orchestration is the control plane that drives the think → act → observe cycle. The orchestrator sits on CPU, calls the LLM (on GPU) for decisions, dispatches tool calls to sandboxes, captures output, and feeds it back. Getting this loop right — with timeout handling, parallel execution, recovery, and human approval gates — is what separates toy demos from production agents.

---

1. The Execution Loop

User message
  │
  ├─→ LLM: Think (what should I do?)
  │     │
  │     ├─→ Tool call: execute_code("npm test")
  │     │     │
  │     │     └─→ Sandbox: run command, capture stdout/stderr
  │     │           │
  │     │           └─→ Tool result → back to LLM
  │     │
  │     ├─→ Tool call: read_file("src/App.tsx")
  │     │     └─→ ...same flow...
  │     │
  │     └─→ No more tool calls → final response to user
  │
  └─→ Stream response tokens to user

The LLM drives the loop — the orchestrator just follows instructions. But the orchestrator enforces boundaries: timeouts, resource limits, approval gates, and error recovery.

---

2. Core Orchestrator Implementation

class AgentOrchestrator:
    def __init__(self, llm, sandbox, tools, config):
        self.llm = llm
        self.sandbox = sandbox
        self.tools = tools              # name → ToolHandler
        self.config = config
        self.messages = []              # Conversation history
        self.tool_call_count = 0
        self.total_tokens = 0
    
    async def run(self, user_message: str) -> AsyncIterator[str]:
        """Execute the agent loop until completion or limit."""
        self.messages.append({"role": "user", "content": user_message})
        
        while self.tool_call_count < self.config.max_tool_calls:
            # Step 1: LLM decides what to do
            response = await self.llm.chat(
                messages=self.messages,
                tools=self._tool_schemas(),
                stream=True,
            )
            
            # Step 2: Collect response (may include tool calls)
            assistant_message = await self._collect_response(response)
            self.messages.append(assistant_message)
            
            # Step 3: If no tool calls, we're done — yield final text
            if not assistant_message.get("tool_calls"):
                if assistant_message.get("content"):
                    yield assistant_message["content"]
                return
            
            # Step 4: Execute tool calls
            tool_results = await self._execute_tools(assistant_message["tool_calls"])
            
            # Step 5: Add results to conversation
            for result in tool_results:
                self.messages.append(result)
                self.tool_call_count += 1
        
        # Hit tool call limit
        yield "I've reached the maximum number of steps. Here's what I've done so far..."
    
    async def _execute_tools(self, tool_calls: list) -> list:
        """Execute tool calls, potentially in parallel."""
        if self.config.parallel_tools and len(tool_calls) > 1:
            return await self._execute_parallel(tool_calls)
        else:
            return await self._execute_sequential(tool_calls)
    
    async def _execute_sequential(self, tool_calls):
        results = []
        for call in tool_calls:
            result = await self._execute_single_tool(call)
            results.append(result)
        return results
    
    async def _execute_single_tool(self, call) -> dict:
        """Execute one tool call with timeout and error handling."""
        tool_name = call["function"]["name"]
        tool_args = json.loads(call["function"]["arguments"])
        
        handler = self.tools.get(tool_name)
        if not handler:
            return self._tool_error(call["id"], f"Unknown tool: {tool_name}")
        
        # Human approval gate for dangerous operations
        if handler.requires_approval:
            approved = await self._request_approval(tool_name, tool_args)
            if not approved:
                return self._tool_error(call["id"], "User declined this action")
        
        try:
            result = await asyncio.wait_for(
                handler.execute(self.sandbox, **tool_args),
                timeout=self.config.tool_timeout,
            )
            # Truncate large outputs
            output = truncate_output(result, max_tokens=self.config.max_tool_output_tokens)
            return {
                "role": "tool",
                "tool_call_id": call["id"],
                "content": output,
            }
        except asyncio.TimeoutError:
            return self._tool_error(call["id"], f"Tool timed out after {self.config.tool_timeout}s")
        except Exception as e:
            return self._tool_error(call["id"], f"Tool error: {str(e)}")

---

3. Parallel Tool Execution

When the LLM requests multiple independent tool calls, execute them concurrently:

async def _execute_parallel(self, tool_calls: list) -> list:
    """Execute independent tool calls concurrently."""
    tasks = []
    for call in tool_calls:
        task = asyncio.create_task(self._execute_single_tool(call))
        tasks.append((call["id"], task))
    
    results = []
    for call_id, task in tasks:
        try:
            result = await asyncio.wait_for(task, timeout=self.config.tool_timeout)
            results.append(result)
        except asyncio.TimeoutError:
            results.append(self._tool_error(call_id, "Timed out"))
        except Exception as e:
            results.append(self._tool_error(call_id, f"Error: {e}"))
    
    return results

When to parallelize:

• ✅ Reading multiple files simultaneously
• ✅ Running independent API calls
• ❌ Sequential operations (write file → run tests on that file)
• ❌ Operations that modify shared state

Most LLM APIs (OpenAI, Anthropic) support emitting multiple tool_calls in a single response when the model determines they’re independent.

---

4. Tool Registry and Schemas

class ToolRegistry:
    """Register tools with execution handlers and metadata."""
    
    def __init__(self):
        self.tools = {}
    
    def register(self, name, handler, schema, requires_approval=False):
        self.tools[name] = Tool(
            name=name,
            handler=handler,
            schema=schema,
            requires_approval=requires_approval,
        )
    
    def schemas(self) -> list[dict]:
        """Return OpenAI-compatible tool schemas."""
        return [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.schema["description"],
                    "parameters": t.schema["parameters"],
                }
            }
            for t in self.tools.values()
        ]

# Register tools
registry = ToolRegistry()

registry.register("execute_code", execute_code_handler, {
    "description": "Execute code in the sandbox. Returns stdout and stderr.",
    "parameters": {
        "type": "object",
        "properties": {
            "language": {"type": "string", "enum": ["python", "bash", "javascript"]},
            "code": {"type": "string", "description": "Code to execute"},
        },
        "required": ["language", "code"],
    }
})

registry.register("read_file", read_file_handler, {
    "description": "Read the contents of a file.",
    "parameters": {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "Path relative to workspace root"},
        },
        "required": ["path"],
    }
})

registry.register("delete_file", delete_file_handler, {
    "description": "Delete a file or directory.",
    "parameters": {
        "type": "object",
        "properties": {
            "path": {"type": "string"},
        },
        "required": ["path"],
    }
}, requires_approval=True)  # Destructive action — requires human approval

---

5. Preemption and Recovery

Long-running agent tasks need timeout and recovery mechanisms:

Timeout Hierarchy

TIMEOUT_CONFIG = {
    "single_tool_call": 120,       # 2 min per tool execution
    "single_turn": 600,            # 10 min per LLM turn (think + all tools)
    "full_session": 3600,          # 1 hour total session
    "idle_timeout": 1800,          # 30 min since last user message
    "max_tool_calls": 100,         # Max tool calls per session
    "max_tokens": 1_000_000,       # Max total tokens (input + output)
}

Recovery via Command Log Replay

If a sandbox crashes or is preempted, reconstruct state by replaying the tool execution log:

class CommandLog:
    """Ordered log of all tool executions for replay-based recovery."""
    
    def __init__(self, storage):
        self.storage = storage
        self.log = []
    
    async def record(self, tool_call: dict, result: str):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "tool": tool_call["function"]["name"],
            "args": tool_call["function"]["arguments"],
            "result_hash": hashlib.md5(result.encode()).hexdigest(),
        }
        self.log.append(entry)
        await self.storage.append(entry)
    
    async def replay(self, sandbox, up_to: int = None):
        """Replay tool calls to reconstruct sandbox state."""
        entries = self.log[:up_to] if up_to else self.log
        
        for entry in entries:
            # Only replay state-modifying operations
            if entry["tool"] in ["execute_code", "write_file", "delete_file"]:
                handler = get_tool_handler(entry["tool"])
                args = json.loads(entry["args"])
                try:
                    await handler.execute(sandbox, **args)
                except Exception as e:
                    logger.warning(f"Replay failed at {entry['timestamp']}: {e}")
                    break
        
        return len(entries)

Limitation: Replay works for filesystem state but not for running processes. If the agent started a server, replay will restart it, but network connections and in-memory state are lost.

---

6. Human-in-the-Loop

For high-stakes actions, pause and ask for human approval:

class ApprovalGate:
    """Pause agent execution for human approval on dangerous actions."""
    
    DANGEROUS_ACTIONS = {
        "delete_file": "Permanently deletes a file",
        "execute_code": lambda args: "sudo" in args.get("code", "") or "rm -rf" in args.get("code", ""),
        "write_file": lambda args: args.get("path", "").startswith("/etc/"),
    }
    
    async def check(self, tool_name: str, args: dict, user_channel) -> bool:
        """Returns True if approved, False if rejected."""
        check = self.DANGEROUS_ACTIONS.get(tool_name)
        
        if check is None:
            return True  # Not dangerous, auto-approve
        
        if callable(check) and not check(args):
            return True  # Condition not met, auto-approve
        
        # Request human approval
        approval_request = {
            "type": "approval_required",
            "tool": tool_name,
            "args": args,
            "reason": check if isinstance(check, str) else "Potentially dangerous action",
        }
        
        await user_channel.send(approval_request)
        
        # Wait for user response (with timeout)
        try:
            response = await asyncio.wait_for(
                user_channel.receive(),
                timeout=300,  # 5 minutes to decide
            )
            return response.get("approved", False)
        except asyncio.TimeoutError:
            return False  # Default deny on timeout

---

7. Observability

Track the execution loop’s health:

# Per-session metrics
session_metrics = {
    "session_id": "sess_abc",
    "total_turns": 8,
    "total_tool_calls": 23,
    "total_tokens": 45_000,
    "total_duration_sec": 180,
    
    # Per-turn breakdown
    "avg_llm_latency_ms": 2500,      # LLM response time
    "avg_tool_latency_ms": 800,       # Tool execution time
    "parallel_tool_pct": 0.35,        # % of turns with parallel tools
    
    # Error tracking
    "tool_errors": 2,                  # Failed tool calls
    "tool_timeouts": 1,                # Timed-out tool calls
    "retries": 1,                      # Automatic retries
    
    # Cost
    "llm_cost_usd": 0.12,
    "compute_cost_usd": 0.03,
}

# Alert on anomalies
if session_metrics["tool_errors"] / session_metrics["total_tool_calls"] > 0.20:
    alert("Session has >20% tool error rate — possible model issue")

---

Self-Check

• Implement the core ReAct agent loop with tool dispatch and output capture
• Design a tool registry with OpenAI-compatible schemas
• Execute parallel tool calls when independent, sequential when dependent
• Implement command log replay for state recovery after sandbox crash
• Add human-in-the-loop approval gates for destructive operations
• Set up multi-level timeouts (per-tool, per-turn, per-session)

---

References

• ReAct: Synergizing Reasoning and Acting in Language Models (Yao et al., 2022)
• Toolformer: Language Models Can Teach Themselves to Use Tools (Schick et al., 2023)
• OpenAI Function Calling Guide
• For sandbox infrastructure, see Sandbox Types
• For scaling agent sessions, see Agent Scaling