Multi-Agent Orchestration Architectures

When Single Agents Are Not Enough

A single LLM agent with tools can handle straightforward tasks: answer a question, search a database, call an API. But complex workflows strain the single-agent model. A research assistant that needs to search the web, analyze data, write a report, and create visualizations is doing four different jobs. Each job benefits from different system prompts, tools, and even different models.

Multi-agent systems split a complex task across specialized agents that communicate and coordinate. Each agent has a focused role, a tailored prompt, and access to only the tools it needs. This separation makes systems easier to build, debug, and improve.

Pattern 1: Supervisor Architecture

A supervisor agent receives the user’s request, breaks it into subtasks, delegates to worker agents, and synthesizes results. It is the most common pattern and the easiest to reason about.

from dataclasses import dataclass, field
from openai import OpenAI

client = OpenAI()

@dataclass
class Agent:
    name: str
    system_prompt: str
    model: str = "gpt-4o"
    tools: list = field(default_factory=list)

    def run(self, messages: list[dict]) -> str:
        response = client.chat.completions.create(
            model=self.model,
            messages=[{"role": "system", "content": self.system_prompt}] + messages,
            tools=self.tools or None,
        )
        return response.choices[0].message.content

# Define specialized agents
researcher = Agent(
    name="researcher",
    system_prompt="You are a research assistant. Find relevant information and cite sources. Be thorough but concise.",
)

analyst = Agent(
    name="analyst",
    system_prompt="You are a data analyst. Analyze information, identify patterns, and provide quantitative insights.",
)

writer = Agent(
    name="writer",
    system_prompt="You are a technical writer. Synthesize research and analysis into clear, well-structured reports.",
)

# Supervisor agent
supervisor = Agent(
    name="supervisor",
    system_prompt="""You are a project supervisor. Given a user request:
1. Break it into subtasks
2. Decide which specialist should handle each subtask
3. Synthesize the results into a final response.

Available specialists: researcher, analyst, writer.
Respond with a JSON plan: {"steps": [{"agent": "name", "task": "description"}]}""",
)

Implementing the Supervisor Loop

import json

AGENTS = {
    "researcher": researcher,
    "analyst": analyst,
    "writer": writer,
}

def run_supervisor(user_request: str) -> str:
    # Step 1: Supervisor creates a plan
    plan_response = supervisor.run([
        {"role": "user", "content": user_request}
    ])
    plan = json.loads(plan_response)

    # Step 2: Execute each step
    results = {}
    context = f"Original request: {user_request}\n\n"

    for step in plan["steps"]:
        agent_name = step["agent"]
        task = step["task"]
        agent = AGENTS[agent_name]

        # Include previous results as context
        prev_context = context
        for prev_agent, prev_result in results.items():
            prev_context += f"\n{prev_agent} found:\n{prev_result}\n"

        result = agent.run([
            {"role": "user", "content": f"{prev_context}\nYour task: {task}"}
        ])
        results[agent_name] = result
        print(f"[{agent_name}] completed: {task[:50]}...")

    # Step 3: Final synthesis
    synthesis_prompt = f"Original request: {user_request}\n\nResults:\n"
    for agent_name, result in results.items():
        synthesis_prompt += f"\n--- {agent_name} ---\n{result}\n"
    synthesis_prompt += "\nSynthesize these into a final, cohesive response."

    final = writer.run([{"role": "user", "content": synthesis_prompt}])
    return final

answer = run_supervisor("Analyze the pros and cons of serverless vs containers for a startup")
print(answer)

Pattern 2: Sequential Pipeline

In a sequential pipeline, agents execute in a fixed order. Each agent transforms the output and passes it to the next. This works when the workflow is predictable and linear.

@dataclass
class PipelineStep:
    agent: Agent
    instruction: str

def run_pipeline(user_input: str, steps: list[PipelineStep]) -> str:
    current_input = user_input

    for i, step in enumerate(steps):
        prompt = f"{step.instruction}\n\nInput:\n{current_input}"
        current_input = step.agent.run([{"role": "user", "content": prompt}])
        print(f"Step {i + 1}/{len(steps)} [{step.agent.name}] complete")

    return current_input

# Define a content pipeline
pipeline = [
    PipelineStep(
        agent=researcher,
        instruction="Research this topic and gather key facts, statistics, and expert opinions.",
    ),
    PipelineStep(
        agent=analyst,
        instruction="Analyze this research. Identify the three strongest arguments for and against. Rate the quality of evidence.",
    ),
    PipelineStep(
        agent=writer,
        instruction="Write a balanced 500-word article based on this analysis. Include an introduction, body, and conclusion.",
    ),
]

article = run_pipeline("The impact of remote work on software team productivity", pipeline)

Sequential pipelines are simple and predictable, but they cannot handle branches or loops. If the analyst determines the research is insufficient, there is no way to send the task back to the researcher.

Pattern 3: Graph-Based Orchestration

Graph-based orchestration models the workflow as a directed graph where nodes are agents and edges are conditional transitions. This handles branching, loops, and dynamic routing.

from enum import Enum

class NodeResult(Enum):
    SUCCESS = "success"
    NEEDS_MORE_INFO = "needs_more_info"
    FAILED = "failed"

@dataclass
class GraphNode:
    agent: Agent
    instruction: str
    transitions: dict[NodeResult, str]  # result -> next node name

@dataclass
class AgentGraph:
    nodes: dict[str, GraphNode]
    start_node: str

    def run(self, user_input: str, max_iterations: int = 10) -> str:
        current_node = self.start_node
        state = {"user_input": user_input, "history": []}
        iterations = 0

        while iterations < max_iterations:
            iterations += 1
            node = self.nodes[current_node]

            # Build context from state
            context = f"User request: {state['user_input']}\n\n"
            for entry in state["history"]:
                context += f"[{entry['agent']}]: {entry['output'][:200]}...\n\n"

            context += f"Your task: {node.instruction}\n"
            context += "End your response with STATUS: success, needs_more_info, or failed."

            result = node.agent.run([{"role": "user", "content": context}])

            # Parse status
            status = NodeResult.SUCCESS
            if "STATUS: needs_more_info" in result:
                status = NodeResult.NEEDS_MORE_INFO
            elif "STATUS: failed" in result:
                status = NodeResult.FAILED

            state["history"].append({
                "agent": node.agent.name,
                "node": current_node,
                "output": result,
                "status": status.value,
            })

            # Determine next node
            next_node = node.transitions.get(status)
            if next_node is None or next_node == "END":
                return result

            current_node = next_node
            print(f"  -> Transitioning to {current_node} ({status.value})")

        return state["history"][-1]["output"]

# Build a research graph with feedback loops
graph = AgentGraph(
    start_node="research",
    nodes={
        "research": GraphNode(
            agent=researcher,
            instruction="Research this topic. If you have enough information, STATUS: success. Otherwise, STATUS: needs_more_info.",
            transitions={
                NodeResult.SUCCESS: "analyze",
                NodeResult.NEEDS_MORE_INFO: "deep_research",
                NodeResult.FAILED: "END",
            },
        ),
        "deep_research": GraphNode(
            agent=researcher,
            instruction="Dig deeper into the gaps identified. Focus on finding primary sources and data.",
            transitions={
                NodeResult.SUCCESS: "analyze",
                NodeResult.NEEDS_MORE_INFO: "analyze",  # Proceed anyway after second attempt
                NodeResult.FAILED: "END",
            },
        ),
        "analyze": GraphNode(
            agent=analyst,
            instruction="Analyze the research findings. Identify key insights and check for contradictions.",
            transitions={
                NodeResult.SUCCESS: "write",
                NodeResult.NEEDS_MORE_INFO: "research",  # Loop back
                NodeResult.FAILED: "END",
            },
        ),
        "write": GraphNode(
            agent=writer,
            instruction="Write the final report based on the research and analysis.",
            transitions={
                NodeResult.SUCCESS: "END",
                NodeResult.FAILED: "END",
            },
        ),
    },
)

result = graph.run("What are the best practices for LLM caching in production?")

State Management

As agents pass data between each other, state management becomes critical. Use a shared state object that each agent can read from and write to.

@dataclass
class WorkflowState:
    user_input: str
    research_findings: list[str] = field(default_factory=list)
    analysis_results: dict = field(default_factory=dict)
    draft: str = ""
    errors: list[str] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)

    def add_finding(self, finding: str, source: str):
        self.research_findings.append(f"{finding} [source: {source}]")

    def to_context(self) -> str:
        """Convert state to a context string for agents."""
        parts = [f"User request: {self.user_input}"]
        if self.research_findings:
            parts.append(f"Research ({len(self.research_findings)} findings):")
            for f in self.research_findings:
                parts.append(f"  - {f}")
        if self.analysis_results:
            parts.append(f"Analysis: {json.dumps(self.analysis_results, indent=2)}")
        if self.errors:
            parts.append(f"Previous errors: {'; '.join(self.errors)}")
        return "\n".join(parts)

Error Handling and Retries

Individual agent failures should not crash the entire workflow. Implement retries at the agent level and fallback strategies at the orchestrator level.

import time

def run_agent_with_retry(agent: Agent, messages: list[dict], max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            result = agent.run(messages)
            if result and len(result.strip()) > 0:
                return result
            raise ValueError("Empty response from agent")
        except Exception as e:
            if attempt == max_retries - 1:
                return f"[Agent {agent.name} failed after {max_retries} attempts: {str(e)}]"
            time.sleep(2 ** attempt)  # Exponential backoff

Observability

Multi-agent systems are hard to debug without logging. Track every agent invocation, its inputs, outputs, latency, and token usage.

import time
import logging

logger = logging.getLogger("multi_agent")

def traced_run(agent: Agent, messages: list[dict], trace_id: str) -> str:
    start = time.time()
    logger.info(f"[{trace_id}] {agent.name} started | input_length={len(str(messages))}")

    try:
        result = agent.run(messages)
        elapsed = time.time() - start
        logger.info(f"[{trace_id}] {agent.name} completed | {elapsed:.2f}s | output_length={len(result)}")
        return result
    except Exception as e:
        elapsed = time.time() - start
        logger.error(f"[{trace_id}] {agent.name} failed | {elapsed:.2f}s | error={str(e)}")
        raise

Choosing the Right Pattern

Use a supervisor when tasks are independent and can be parallelized, or when the decomposition itself requires intelligence.

Use a sequential pipeline when the workflow is linear and predictable, like extract-transform-load or draft-review-publish.

Use a graph when the workflow has conditional branches, feedback loops, or dynamic routing based on intermediate results.

Start with the simplest pattern that handles your use case. A single agent with good tools often outperforms a poorly designed multi-agent system. Add agents only when a single agent’s context window or instruction set becomes unmanageable.

Summary

Multi-agent orchestration splits complex LLM tasks across specialized agents. The supervisor pattern delegates subtasks and synthesizes results. Sequential pipelines pass data through a fixed chain of agents. Graph-based orchestration handles branches, loops, and conditional routing. Use shared state objects for inter-agent communication, implement retries at the agent level, and log every invocation for debugging. Start simple and add agents only when the complexity demands it.