AI Agent Orchestration Design Patterns — Task Decomposition, Handoffs, and Loop Control

When you start building AI agents, the first wall you hit isn't prompt quality — it's the design question of how much to delegate to a single agent.

My first agent crammed three responsibilities into one: web research, summarization, and report generation. It worked, technically. But midway through, the context would fill up and the response quality would gradually degrade. Retries reproduced the same pattern. That experience pushed me to take orchestration design seriously.

How to Decide Task Decomposition Granularity

The decision to split an agent into sub-agents is easiest to reason about across two axes: context window consumption and processing independence.

Separate context-heavy tasks. Reading large document sets for summarization, scanning an entire codebase for bugs — running these alongside other work contaminates the context and introduces drift in later decisions. Ask yourself: "What does the context look like after this task finishes?" That mental model often makes the split obvious.

Parallelize independent tasks. If three API calls don't reference each other's results, there's no reason to run them sequentially. Anthropic's Claude Agent SDK lets you run sub-agents in parallel with asyncio.gather():

import asyncio
from anthropic import Anthropic
 
client = Anthropic()
 
async def run_agent(task: str, context: str) -> str:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        system="You are a specialist agent. Complete the given task.",
        messages=[{"role": "user", "content": f"Context: {context}
 
Task: {task}"}]
    )
    return response.content[0].text
 
async def parallel_research(topics: list[str], shared_context: str) -> list[str]:
    tasks = [run_agent(f"Research and summarize: {topic}", shared_context) for topic in topics]
    return await asyncio.gather(*tasks)

Something the official docs don't cover: parallel execution hits API rate limits quickly in practice. Limiting concurrency to 3–5 with asyncio.Semaphore is the pragmatic sweet spot.

Structuring Sub-Agent Handoffs

What you pass to a sub-agent — and what you withhold — determines quality more than the agent's prompt.

Pass: The goal (Why), current known state (What is known), completion criteria (Definition of Done), and failure policy.

Don't pass: The orchestrator's internal reasoning, instructions for other sub-agents (causes interference), full failure logs from past attempts (summarize the key lesson only).

def create_handoff_context(
    goal: str,
    current_state: dict,
    success_criteria: str,
    failure_policy: str = "report and stop"
) -> str:
    return f"""
## Your Mission
{goal}
 
## What Is Currently Known
{json.dumps(current_state, indent=2)}
 
## Completion Criteria
{success_criteria}
 
## On Failure
{failure_policy}
"""

After adopting this structure, sub-agent responses shifted from "interpreting what I should do" to "doing the work." The difference is noticeable immediately.

Loop Control and Termination Design

The most dangerous failure mode in agent loops isn't infinite looping — it's terminating before the task is actually complete.

Never delegate termination decisions entirely to the agent. Three-layer control works well:

class AgentLoop:
    def __init__(self, max_iterations: int = 10, timeout_seconds: int = 300):
        self.max_iterations = max_iterations
        self.timeout_seconds = timeout_seconds
        self.iteration = 0
        self.start_time = None
 
    def should_continue(self, agent_says_done: bool, result: dict) -> bool:
        # Layer 1: Hard limits (highest priority)
        if self.iteration >= self.max_iterations:
            return False
        import time
        if time.time() - self.start_time > self.timeout_seconds:
            return False
        # Layer 2: Agent self-report
        if agent_says_done:
            # Layer 3: Independent result validation
            return not self._validate_completion(result)
        return True
 
    def _validate_completion(self, result: dict) -> bool:
        required_keys = ["summary", "confidence", "next_actions"]
        return all(k in result and result[k] for k in required_keys)

In practice, max_iterations=10 is sufficient for nearly all tasks. Tasks that need more usually indicate incomplete decomposition — the loop budget is a useful forcing function.

Error Handling Patterns

Agent errors fall into three categories.

Transient errors (API timeout, rate limit): Retry with exponential backoff, max 3 times, then escalate.

Tool execution errors (file not found, permission denied): Change tool invocation and retry. Escalate if the same error occurs twice.

Logic errors (task cannot satisfy completion criteria): Retrying is futile. Return structured error information the orchestrator can act on.

from enum import Enum
 
class ErrorType(Enum):
    TRANSIENT = "transient"
    TOOL = "tool"
    LOGIC = "logic"
 
def classify_error(error: Exception) -> ErrorType:
    s = str(error).lower()
    if "timeout" in s or "rate limit" in s:
        return ErrorType.TRANSIENT
    if "not found" in s or "permission" in s:
        return ErrorType.TOOL
    return ErrorType.LOGIC

A Real Configuration

Here's a simplified version of the content generation agent I use personally:

Orchestrator → Research Agents (3 parallel) → Structure Agent → Writing Agent → Quality Check Agent

The critical design choice at each handoff: each agent must return output in a shape the next agent can consume without interpretation. I enforce this by specifying JSON output schemas in each agent's system prompt.

If you're just starting out: begin with a two-stage setup — a researcher and a writer. Even this simple split produces noticeably better output than a single do-everything agent. Build intuition there before adding complexity. It looks slower, but it's faster in practice.