Advanced Agent Orchestration Patterns in Python
Building a single AI agent that calls an LLM and uses a few tools is straightforward. Building a system of agents that collaborate reliably, recover from failures, and know when to ask a human for help is an entirely different challenge. The difference between a demo and a production system often comes down to orchestration patterns — the architectural decisions that determine how agents communicate, make decisions, and handle the inevitable failures that come with non-deterministic systems.
I will cover the orchestration patterns I have found most effective for production multi-agent systems in Python. These patterns are framework-agnostic -- whether you are using LangGraph, CrewAI, AutoGen, or the OpenAI Agents SDK, the underlying principles apply.
The supervisor pattern
The supervisor pattern is the most common starting point for multi-agent systems. A single supervisor agent receives tasks, decides which worker agent should handle each one, and aggregates the results.
from dataclasses import dataclass, field
from typing import Protocol
from openai import OpenAI
import json
class WorkerAgent(Protocol):
name: str
def execute(self, task: str) -> str: ...
@dataclass
class Supervisor:
"""Routes tasks to specialized worker agents."""
client: OpenAI
workers: dict[str, WorkerAgent] = field(default_factory=dict)
model: str = "gpt-4o"
def register_worker(self, worker: WorkerAgent):
self.workers[worker.name] = worker
def route(self, task: str) -> str:
worker_descriptions = "\n".join(
f"- {name}: {w.__doc__ or 'No description'}"
for name, w in self.workers.items()
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
"You are a task router. Given a task, decide which "
"worker should handle it. Respond with a JSON object: "
'{"worker": "worker_name", "subtask": "refined task"}\n\n'
f"Available workers:\n{worker_descriptions}"
)},
{"role": "user", "content": task}
],
response_format={"type": "json_object"}
)
decision = json.loads(response.choices[0].message.content)
worker_name = decision["worker"]
subtask = decision["subtask"]
if worker_name not in self.workers:
return f"Error: Unknown worker '{worker_name}'"
return self.workers[worker_name].execute(subtask)
The supervisor pattern works well when tasks are independent and can be routed to a single specialist. However, it breaks down when tasks require collaboration between multiple agents or when the supervisor becomes a bottleneck.
Hierarchical agent architecture
For complex workflows, a flat supervisor is insufficient. Hierarchical architectures introduce layers of management, where sub-supervisors manage their own teams of agents.
from dataclasses import dataclass, field
from openai import OpenAI
import json
@dataclass
class AgentNode:
"""A node in the agent hierarchy."""
name: str
client: OpenAI
system_prompt: str
children: list["AgentNode"] = field(default_factory=list)
tools: list[callable] = field(default_factory=list)
model: str = "gpt-4o"
def execute(self, task: str, depth: int = 0) -> dict:
indent = " " * depth
print(f"{indent}[{self.name}] Processing: {task[:80]}...")
if not self.children:
# Leaf node: execute directly
return self._execute_direct(task)
# Manager node: decompose and delegate
subtasks = self._decompose_task(task)
results = {}
for subtask in subtasks:
target = self._select_child(subtask)
results[subtask] = target.execute(subtask, depth + 1)
# Synthesize results
return self._synthesize(task, results)
def _decompose_task(self, task: str) -> list[str]:
children_desc = ", ".join(c.name for c in self.children)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
"Decompose this task into 2-4 subtasks that can be "
"handled by these agents: " + children_desc + ". "
"Return a JSON array of subtask strings."
)},
{"role": "user", "content": task}
],
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
return result.get("subtasks", [task])
def _select_child(self, subtask: str) -> "AgentNode":
if len(self.children) == 1:
return self.children[0]
children_desc = "\n".join(
f"- {c.name}: {c.system_prompt[:100]}"
for c in self.children
)
response = self.client.chat.completions.create(
model="gpt-4o-mini", # Use cheaper model for routing
messages=[
{"role": "system", "content": (
"Select the best agent for this subtask. "
"Respond with only the agent name.\n\n"
f"Agents:\n{children_desc}"
)},
{"role": "user", "content": subtask}
]
)
name = response.choices[0].message.content.strip()
return next(
(c for c in self.children if c.name == name),
self.children[0]
)
def _execute_direct(self, task: str) -> dict:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": task}
]
)
return {
"agent": self.name,
"result": response.choices[0].message.content
}
def _synthesize(self, original_task: str, results: dict) -> dict:
results_text = "\n\n".join(
f"## {subtask}\n{result}"
for subtask, result in results.items()
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
"Synthesize these results into a coherent response "
"for the original task."
)},
{"role": "user", "content": (
f"Original task: {original_task}\n\n"
f"Results:\n{results_text}"
)}
]
)
return {
"agent": self.name,
"result": response.choices[0].message.content,
"sub_results": results
}
Usage looks like building an org chart:
client = OpenAI()
# Build the hierarchy
research_team = AgentNode(
name="ResearchManager", client=client,
system_prompt="You manage a research team.",
children=[
AgentNode(name="WebResearcher", client=client,
system_prompt="You search the web for information."),
AgentNode(name="DataAnalyst", client=client,
system_prompt="You analyze data and find patterns."),
]
)
engineering_team = AgentNode(
name="EngineeringManager", client=client,
system_prompt="You manage an engineering team.",
children=[
AgentNode(name="BackendDev", client=client,
system_prompt="You write backend code in Python."),
AgentNode(name="DevOps", client=client,
system_prompt="You handle infrastructure and deployment."),
]
)
cto = AgentNode(
name="CTO", client=client,
system_prompt="You oversee all technical operations.",
children=[research_team, engineering_team]
)
result = cto.execute("Build a recommendation system for our e-commerce platform")
Error recovery patterns
Production agent systems must handle failures gracefully. LLM calls can time out, tools can fail, and agents can produce invalid outputs. Here is a solid error recovery pattern:
from dataclasses import dataclass
from typing import Any
import time
import logging
logger = logging.getLogger(__name__)
@dataclass
class AgentResult:
success: bool
output: Any
error: str | None = None
retries: int = 0
class ResilientAgent:
"""Agent with built-in error recovery."""
def __init__(self, client, system_prompt: str, max_retries: int = 3):
self.client = client
self.system_prompt = system_prompt
self.max_retries = max_retries
self.fallback_handlers: list[callable] = []
def add_fallback(self, handler: callable):
self.fallback_handlers.append(handler)
def execute(self, task: str) -> AgentResult:
last_error = None
for attempt in range(self.max_retries):
try:
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": task}
],
timeout=30
)
output = response.choices[0].message.content
# Validate output
if not output or len(output.strip()) < 10:
raise ValueError("Agent produced empty or trivial output")
return AgentResult(
success=True, output=output, retries=attempt
)
except Exception as e:
last_error = str(e)
logger.warning(
f"Attempt {attempt + 1}/{self.max_retries} failed: {e}"
)
if attempt < self.max_retries - 1:
wait_time = 2 ** attempt
time.sleep(wait_time)
# All retries exhausted, try fallbacks
for handler in self.fallback_handlers:
try:
result = handler(task)
return AgentResult(
success=True, output=result,
retries=self.max_retries
)
except Exception as e:
logger.warning(f"Fallback failed: {e}")
return AgentResult(
success=False, output=None, error=last_error,
retries=self.max_retries
)
The key principles are: retry with exponential backoff for transient failures, validate outputs before returning them, and have fallback handlers for when all retries are exhausted. A common fallback is to use a cheaper or faster model, return a cached response, or escalate to a human.
Human-in-the-loop pattern
Not every decision should be automated. The human-in-the-loop pattern lets agents escalate to humans when confidence is low or when the stakes are high.
from dataclasses import dataclass
from enum import Enum
from typing import Callable
import json
class Confidence(Enum):
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
@dataclass
class Decision:
action: str
confidence: Confidence
reasoning: str
requires_approval: bool
class HumanInTheLoopAgent:
"""Agent that escalates to humans based on confidence."""
def __init__(
self,
client,
system_prompt: str,
approval_callback: Callable[[Decision], bool],
confidence_threshold: Confidence = Confidence.MEDIUM
):
self.client = client
self.system_prompt = system_prompt
self.approval_callback = approval_callback
self.confidence_threshold = confidence_threshold
def execute(self, task: str) -> str:
# Step 1: Agent analyzes the task and proposes an action
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": (
self.system_prompt + "\n\n"
"For every action, provide a JSON response with:\n"
'{"action": "what to do", '
'"confidence": "high|medium|low", '
'"reasoning": "why this action", '
'"requires_approval": true/false}\n\n'
"Set requires_approval to true for: irreversible "
"actions, actions involving money, actions affecting "
"production systems, or when confidence is low."
)},
{"role": "user", "content": task}
],
response_format={"type": "json_object"}
)
decision_data = json.loads(
response.choices[0].message.content
)
decision = Decision(
action=decision_data["action"],
confidence=Confidence(decision_data["confidence"]),
reasoning=decision_data["reasoning"],
requires_approval=decision_data["requires_approval"]
)
# Step 2: Check if human approval is needed
if decision.requires_approval or \
decision.confidence == Confidence.LOW:
print(f"\n--- Human Approval Required ---")
print(f"Action: {decision.action}")
print(f"Confidence: {decision.confidence.value}")
print(f"Reasoning: {decision.reasoning}")
approved = self.approval_callback(decision)
if not approved:
return "Action was rejected by human reviewer."
# Step 3: Execute the approved action
return self._execute_action(decision)
def _execute_action(self, decision: Decision) -> str:
return f"Executed: {decision.action}"
# Usage
def cli_approval(decision: Decision) -> bool:
response = input("Approve this action? (yes/no): ")
return response.lower() in ("yes", "y")
agent = HumanInTheLoopAgent(
client=OpenAI(),
system_prompt="You are a deployment agent that manages production releases.",
approval_callback=cli_approval
)
result = agent.execute("Deploy version 2.3.1 to production")
Tool-use strategy pattern
How agents select and use tools matters as much as which tools are available. The strategy pattern lets you define different tool selection behaviors depending on the context.
from abc import ABC, abstractmethod
from dataclasses import dataclass
class ToolStrategy(ABC):
"""Base class for tool selection strategies."""
@abstractmethod
def select_tools(
self, task: str, available_tools: list[dict]
) -> list[dict]:
...
class ConservativeStrategy(ToolStrategy):
"""Use minimal tools - prefer built-in knowledge."""
def select_tools(self, task, available_tools):
# Only provide tools that are explicitly relevant
return [t for t in available_tools if t["required"]]
class AggressiveStrategy(ToolStrategy):
"""Provide all potentially relevant tools."""
def select_tools(self, task, available_tools):
return available_tools
class AdaptiveStrategy(ToolStrategy):
"""Select tools based on task complexity analysis."""
def __init__(self, client):
self.client = client
def select_tools(self, task, available_tools):
tool_names = [t["name"] for t in available_tools]
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": (
"Given a task and available tools, return a JSON "
"array of tool names needed. Be selective — only "
"include tools that are clearly needed.\n\n"
f"Available tools: {tool_names}"
)},
{"role": "user", "content": task}
],
response_format={"type": "json_object"}
)
import json
selected = json.loads(
response.choices[0].message.content
).get("tools", [])
return [t for t in available_tools if t["name"] in selected]
@dataclass
class StrategicAgent:
"""Agent that uses different tool strategies."""
client: object
strategy: ToolStrategy
tools: list[dict]
def execute(self, task: str) -> str:
selected_tools = self.strategy.select_tools(task, self.tools)
print(f"Using {len(selected_tools)}/{len(self.tools)} tools")
# Execute with selected tools...
return "Result"
Putting it all together
These patterns are not mutually exclusive. A production system typically combines several of them:
- A hierarchical supervisor routes tasks to specialized teams
- Each team uses the tool strategy pattern to select appropriate tools
- Error recovery wraps every agent call with retries and fallbacks
- Human-in-the-loop gates protect high-stakes actions
Agent orchestration is really a software engineering problem. The LLM provides the reasoning capability, but reliability comes from the same patterns we use in distributed systems: retries, circuit breakers, timeouts, supervision trees, and clear escalation paths.
Start with the simplest pattern that solves your problem, usually a flat supervisor with two or three workers. Add complexity only when you hit real limitations. Every additional layer of orchestration adds latency, cost, and potential failure modes. The best agent system is the simplest one that gets the job done reliably.