Agent Orchestration: Best Practices for Multi-Agent Systems

Agent orchestration is the art of coordinating multiple AI agents so they solve complex tasks together—reliably, efficiently, and transparently. Individual agents are rarely the bottleneck. The challenge is coordination: who does what, in what order, and what happens when things break? How do you keep costs under control?

TL;DR — Agent Orchestration in 90 Seconds

What Is Agent Orchestration?

Agent orchestration coordinates multiple AI agents across six dimensions:

• Task delegation: Who handles what? Which parts require specific expertise?
• Communication: How do agents exchange information? Through shared state, messages, or events?
• Control flow: Who decides the next step? A central orchestrator or conditional routing logic?
• Error handling: What happens when things fail? Retry, fallback, or human intervention?
• Resource optimization: How do you minimize LLM calls? Caching, model selection, token budgets?
• Observability: How do you track what’s happening? Logs, traces, and metrics?

Single-Agent vs. Multi-Agent — When Do You Need Orchestration?

Single-agent works for: linear tasks, a single domain of expertise, one-shot solutions that fit within a single LLM call.

Multi-agent becomes necessary when: you need multiple specialties (research plus writing plus review), iterative feedback loops (writer ↔ reviewer), parallel execution, tasks exceeding context windows, varied tooling, or human-in-the-loop approval.

Orchestration Patterns — Deep Dive

1. Hub-and-Spoke (Central Orchestrator)

A central orchestrator agent decides which agents run and when.

         [Orchestrator]
        /     |      \
[Agent A] [Agent B] [Agent C]

Strengths: Clear control flow, straightforward error handling, fully auditable. Weaknesses: Single point of failure, orchestrator becomes a bottleneck, extra LLM cost per decision.

def orchestrator(state):
    task = state["task"]
    # LLM decides which agent to use
    response = llm.invoke(f"Which agent for: {task}? (research/code/write)")
    task_type = response.content.strip().lower()
    return {"task_type": task_type}

def route_to_agent(state):
    return state["task_type"]

# In LangGraph:
workflow.add_conditional_edges("orchestrator", route_to_agent, {
    "research": "research_agent",
    "code": "code_agent",
    "write": "writer_agent"
})

2. Pipeline (Sequential)

Agents work one after another, each passing results to the next. CrewAI’s Process.sequential follows this pattern by default.

[Agent A] → [Agent B] → [Agent C] → [Output]

Strengths: Simple, predictable, each agent receives all prior results. Weaknesses: No parallelism, one failure blocks everything downstream.

When to use: Linear workflows without branching (research → write → review).

3. DAG (Directed Acyclic Graph)

Agents execute according to dependencies, with parallelism wherever possible. A DAG lets independent tasks run simultaneously while coordinating dependent ones.

[Agent A] ──→ [Agent C] ──→ [Agent E] (Merge)
     └──→ [Agent B] ──→ [Agent D] ──┘

Strengths: Parallelism—independent tasks run concurrently. Efficient—total time equals the longest path. Weaknesses: Complex to debug, dependencies must be explicit, potential race conditions with shared state.

Example with LangGraph (parallel researchers):

class ResearchState(TypedDict):
    topic: str
    general_research: str
    technical_research: str
    combined_report: str

def general_researcher(state: ResearchState) -> dict:
    response = llm.invoke(f"General information on: {state['topic']}")
    return {"general_research": response.content}

def technical_researcher(state: ResearchState) -> dict:
    response = llm.invoke(f"Technical details on: {state['topic']}")
    return {"technical_research": response.content}

def merge_results(state: ResearchState) -> dict:
    combined = f"General:\n{state['general_research']}\n\nTechnical:\n{state['technical_research']}"
    return {"combined_report": combined}

# Graph with parallel nodes
workflow = StateGraph(ResearchState)
workflow.add_node("general", general_researcher)
workflow.add_node("technical", technical_researcher)
workflow.add_node("merge", merge_results)

workflow.set_entry_point("general")
workflow.add_edge("general", "technical")  # Both run in parallel
workflow.add_edge("general", "merge")
workflow.add_edge("technical", "merge")
workflow.add_edge("merge", END)

When to use: When independent subtasks can run in parallel (for example, three separate research streams that merge afterward).

4. Event-Driven (Reactive)

Agents respond to events rather than direct calls. An event bus distributes events to all interested subscribers.

[Event Bus]
  ├── [Agent A] (listens for "research.done")
  ├── [Agent B] (listens for "code.reviewed")
  └── [Agent C] (listens for "test.failed")

Strengths: Loose coupling, scales horizontally, agents can be added dynamically. Weaknesses: Hard to trace, race conditions possible, debugging becomes complex.

from collections import defaultdict

class EventBus:
    def __init__(self):
        self.subscribers = defaultdict(list)
    
    def subscribe(self, event_type: str, handler):
        self.subscribers[event_type].append(handler)
    
    def publish(self, event_type: str, data: dict):
        results = []
        for handler in self.subscribers[event_type]:
            results.append(handler(data))
        return results

bus = EventBus()
bus.subscribe("research.done", writer_agent)
bus.subscribe("research.done", fact_checker_agent)
bus.subscribe("code.reviewed", refactor_agent)
bus.publish("research.done", {"topic": "AI 2026", "result": "..."})

When to use: Loosely coupled systems where agents are added dynamically and execution order isn’t predetermined.

Pattern Comparison

Pattern	Complexity	Parallelism	Flexibility	Debugging	Best For
Pipeline	Low	❌	Low	Easy	Linear workflows
Hub-and-Spoke	Medium	❌	Medium	Moderate	Dynamic routing
DAG	High	✅	High	Difficult	Parallel tasks
Event-Driven	High	✅	Very high	Very difficult	Loosely coupled systems

Best Practices — In Detail

1. Define Clear Roles (Single Responsibility Principle)

Each agent should have exactly one responsibility. An agent that simultaneously researches, writes, and tests is difficult to debug, expensive to run, and produces worse results than three specialized agents.

Poor:

agent = Agent(
    role="Researcher, Writer and Tester",
    goal="Research, write and test everything"
)

Good:

researcher = Agent(role="Researcher", goal="Find information")
writer = Agent(role="Writer", goal="Write content based on research")
tester = Agent(role="Tester", goal="Test functionality and report bugs")

2. Make State Explicit

Use typed states instead of implicit communication. This makes the workflow traceable and resilient to errors.

from typing import TypedDict, Optional, List, Annotated
from operator import add

class WorkflowState(TypedDict):
    input: str
    task_type: str
    research_result: Optional[str]
    draft: Optional[str]
    review_feedback: Optional[str]
    final_output: Optional[str]
    error: Optional[str]
    retry_count: int
    messages: Annotated[List[str], add]  # Appended, not overwritten

3. Error Handling and Retry Logic

LLMs are unreliable — rate limits, timeouts, hallucinations. Every agent needs error handling:

import logging

logger = logging.getLogger("agent_orchestrator")

def agent_with_retry(agent_func, state, max_retries=3, fallback=None):
    for attempt in range(max_retries):
        try:
            logger.info(f"Agent {agent_func.__name__} - Attempt {attempt + 1}")
            result = agent_func(state)
            if validate_result(result):
                return result
            else:
                logger.warning(f"Agent {agent_func.__name__} - invalid result")
                state["error"] = "Invalid result"
        except Exception as e:
            logger.error(f"Agent {agent_func.__name__} - Error: {str(e)}")
            state["error"] = str(e)
            state["retry_count"] = attempt + 1
    
    if fallback:
        return fallback(state)
    return {**state, "error": f"Agent failed after {max_retries} retries"}

def validate_result(result):
    if not result:
        return False
    if "error" in result and result["error"]:
        return False
    return True

4. Cost Control

LLM calls add up fast. With 5 agents each making 3 revisions, that’s 15+ LLM calls. Using GPT-4o, a single run can cost $5–20.

class CostTracker:
    def __init__(self, max_budget=10.0):
        self.costs = {}
        self.max_budget = max_budget
    
    def track(self, agent_name, tokens, model="gpt-4o"):
        cost_per_1k = {
            "gpt-4o": 0.005,
            "gpt-4o-mini": 0.0003,
            "claude-3.5-sonnet": 0.003,
        }
        rate = cost_per_1k.get(model, 0.005)
        cost = (tokens / 1000) * rate
        self.costs[agent_name] = self.costs.get(agent_name, 0) + cost
        
        if self.total() > self.max_budget:
            raise Exception(f"Budget exceeded: ${self.total():.2f}")
    
    def total(self):
        return sum(self.costs.values())
    
    def report(self):
        lines = [f"Cost Report (Total: ${self.total():.4f})"]
        for agent, cost in sorted(self.costs.items()):
            lines.append(f"  {agent}: ${cost:.4f}")
        return "\n".join(lines)

Cost Optimization Strategies:

• Caching: Don’t run the same prompt twice
• Model Selection: Use GPT-4o-mini ($0.30/M) for simple tasks, GPT-4o ($5.00/M) for complex ones
• Early Stopping: Stop when the result is “good enough”
• Token Limits: Set max_tokens per agent
• Local Models: Use Ollama for development ($0)

5. Observability and Logging

Without observability, you’re flying blind — you don’t know how long each agent takes or where failures occur.

import logging
import time
from functools import wraps

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agent_orchestrator")

def observed_agent(agent_func):
    @wraps(agent_func)
    def wrapper(state):
        name = agent_func.__name__
        start = time.time()
        logger.info(f"[{name}] START - Input keys: {list(state.keys())}")
        try:
            result = agent_func(state)
            duration = time.time() - start
            logger.info(f"[{name}] DONE - {duration:.2f}s")
            return result
        except Exception as e:
            duration = time.time() - start
            logger.error(f"[{name}] ERROR - {duration:.2f}s - {str(e)}")
            raise
    return wrapper

Additionally: Use LangSmith for detailed tracing — visualizes every LLM call, state transition, and token calculation:

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."

6. Human-in-the-Loop

For critical decisions, a human should approve. LangGraph supports this natively with interrupt_before:

app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["publisher"]  # Pauses before publisher
)

# First run: executes up to publisher
result = app.invoke(initial_state, config={"configurable": {"thread_id": "1"}})

# Human reviews and approves
approval = input("Approve? (y/n): ")
result = app.invoke(
    {"approved": approval == "y"},
    config={"configurable": {"thread_id": "1"}}
)

When to use Human-in-the-Loop: Publishing content, production code changes, decisions with financial impact, low agent confidence.

Anti-Patterns — What to Avoid

1. God Agent

One agent doing everything. The prompt becomes massive → LLM loses focus → poor quality. Hard to debug and expensive to operate. Solution: Split into 3–5 specialized agents.

2. Infinite Loops

Agents calling each other without a stopping condition. Solution: Set maximum iterations and implement an emergency brake:

MAX_ITERATIONS = 5

def route_after_review(state):
    if state.get("approved"):
        return "publish"
    if state.get("iteration_count", 0) >= MAX_ITERATIONS:
        return "human_review"  # Emergency brake
    return "writer"

3. No Validation

Blindly trusting agent output. LLMs hallucinate and invent sources. Solution: Validate every output:

def validate_research(result: str) -> bool:
    if not result or len(result) < 100:
        return False
    if "http" not in result:  # At least one source
        return False
    return True

4. Too Many Agents

More agents means more coordination overhead, higher costs, and more failure points. Rule of thumb: 3–5 agents. If you need more, split into multiple independent crews.

5. No Cost Control

Without cost tracking, multi-agent systems can easily cost hundreds of dollars per run. Solution: Use CostTracker with a budget limit (see above).

Framework Comparison for Orchestration

Feature	LangGraph	CrewAI	Custom
Pipeline	✅	✅ (sequential)	✅
Hub-and-Spoke	✅ (conditional edges)	✅ (hierarchical)	✅
DAG/Parallel	✅ (native)	❌	✅ (asyncio)
Event-Driven	❌ (not native)	❌	✅ (custom bus)
Cycles	✅ (conditional edges)	Limited	✅
Human-in-the-Loop	✅ (interrupt)	Manual	✅
Checkpointing	✅ (native)	❌	✅ (custom)

Exam Essentials

• Orchestration: Coordinating multiple agents through task delegation, communication, control flow, error handling, and resource management
• 4 Patterns: Pipeline (sequential), Hub-and-Spoke (centralized), DAG (parallel), Event-Driven (reactive)
• Best Practices: Clear roles (SRP), explicit typed states, retry logic with max_retries, cost control via CostTracker, observability through logging and LangSmith, Human-in-the-Loop for critical decisions
• Anti-Patterns: God Agent, unbounded loops, missing validation, too many agents, no cost control
• Framework Selection: LangGraph for complex graphs, CrewAI for rapid prototyping, custom solutions for Event-Driven patterns

FAQ

When do I need multi-agent over single-agent? When the task demands multiple specialized skills, iteration is required (Writer ↔ Reviewer), parallelism is possible, or the problem is too complex for a single prompt to handle.

What’s the optimal number of agents? Three to five agents covers most use cases. Beyond eight, coordination becomes unwieldy and costs spike. For complex requirements, split the work into multiple independent crews or graphs instead.

Which orchestration pattern should I use? Pipeline for straightforward linear tasks, Hub-and-Spoke for dynamic routing, DAG for parallel work, Event-Driven for loosely coupled systems. Most production systems blend multiple patterns.

How do I prevent infinite loops? Always set a maximum iteration count. Implement an emergency stop that triggers Human-in-the-Loop or END status when iterations exceed the threshold.

How do I control costs in multi-agent systems?

1. CostTracker with a budget limit. 2) Cheaper models for simple tasks. 3) Caching for repeated prompts. 4) Early stopping. 5) Local models during development.

Do I need LangSmith in production? Highly recommended. Without tracing, you’re flying blind—you won’t know which agent takes how long or where failures occur. LangSmith offers a free tier for small projects.

Recommended Reading