Agent Orchestration: Best Practices for Multi-Agent Systems

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

TL;DR — Agent Orchestration in 90 Seconds

What is Agent Orchestration?

Agent orchestration is the coordination of multiple AI agents across:

• Task division: who does what? which expertise does each part need?
• Communication: how do agents share information? via state, messages, or events?
• Control: who decides the next step? a central orchestrator or conditional routing?
• Error handling: what happens when things fail? retry, fallback, human-in-the-loop?
• Resource management: how do you optimize LLM calls? caching, model selection, token limits?
• Observability: how do you track what’s happening? logging, tracing, metrics?

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

Single agent is enough for: linear tasks, a single domain of expertise, no iteration needed, task solvable in one LLM call.

Multi-agent is necessary for: diverse expertise (research + writing + review), iteration (writer ↔ reviewer), parallelism, tasks exceeding context limits, multiple tools, human-in-the-loop workflows.

Orchestration Patterns — In Detail

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, simple error handling, fully traceable. Weaknesses: single point of failure, orchestrator becomes a bottleneck, extra LLM costs.

def orchestrator(state):
    task = state["task"]
    # LLM decides which agent is needed
    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 output to the next. CrewAI’s Process.sequential uses this pattern by default.

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

Strengths: simple, predictable, each agent receives all prior results. Weaknesses: no parallelism, one failing agent blocks everything.

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

3. DAG (Directed Acyclic Graph)

Agents work with dependencies, running in parallel where possible. A DAG lets you execute independent tasks concurrently and coordinate dependent ones.

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

Strengths: parallelism—independent tasks run simultaneously. efficiency—total time equals the longest path. Weaknesses: complex to debug, dependencies must be explicit, race conditions possible 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 info 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 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 (e.g., 3 research streams that merge later).

4. Event-Driven (Reactive)

Agents respond to events instead of direct calls. An event bus distributes events to interested agents.

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

Strengths: loose coupling, scalable, agents can be added dynamically. Weaknesses: hard to trace, race conditions possible, debugging is 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 is not predetermined.

Pattern Comparison

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

Best Practices — In Depth

1. Define Clear Roles (Single Responsibility Principle)

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

Bad:

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 state objects instead of implicit communication. This makes your workflow traceable and fault-resistant.

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 happen. 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 are expensive. With five agents, each running three revisions, you’re looking at 15+ LLM calls. On GPT-4o, that can easily be $5–20 per run.

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: Exit 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 won’t know which agent takes how long 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

Bonus: LangSmith for detailed tracing — shows every LLM call, state transition, and token calculation visually:

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

6. Human-in-the-Loop

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

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

# First run: executes until before 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, financial impact, low agent confidence.

Anti-Patterns — What to Avoid

1. God Agent

A single agent doing everything. The prompt balloons → the LLM loses focus → quality suffers. Hard to debug and expensive to operate. Solution: Split into three to five specialized agents.

2. Unbounded Loops

Agents calling each other with no exit condition. Solution: Set max iterations and implement a circuit breaker:

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"  # Circuit breaker
    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: three to five agents. If you need more, split into multiple independent crews.

5. No Cost Control

Without cost tracking, multi-agent systems can quickly run hundreds of dollars per run. Solution: Use a 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)

Key Exam Points

• Orchestration: Coordinating multiple agents through task division, communication, control, 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 controls
• Framework Selection: LangGraph for complex graphs, CrewAI for rapid prototyping, custom solutions for Event-Driven architectures

FAQ

When do I need Multi-Agent instead of Single-Agent? When your task requires different expertise areas, iteration is necessary (Writer ↔ Reviewer), parallelism is possible, or the task is too complex for a single prompt.

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

Which orchestration pattern works best? Pipeline for straightforward linear tasks, Hub-and-Spoke when you need dynamic routing, DAG for parallel work, Event-Driven for loosely coupled systems. Most production systems combine several patterns.

How do I prevent infinite loops? Always set a maximum iteration count. Implement a circuit breaker that escalates to Human-in-the-Loop or END after too many iterations.

How do I manage costs in Multi-Agent systems?

1. Use CostTracker with budget limits. 2) Route simple tasks to cheaper models. 3) Cache identical prompts. 4) Enable early stopping. 5) Run local models during development.

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

Recommended Reading