Agent Architectures Compared: Single Agent, Pipeline, Router, and Swarm

Choosing the Right Agent Architecture

Not every problem needs the same agent architecture. Using a swarm of agents for a simple lookup task is overengineering. Using a single agent for a complex multi-domain workflow leads to poor results. Understanding the four fundamental architectures — and when to use each — is the key architectural decision in any agent project.

Architecture 1: Single Agent

A single agent with a set of tools handles the entire task from start to finish. This is the simplest architecture and the right starting point for most projects.

flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus<br/>classify"]
    PLAN["Plan and tool<br/>selection"]
    AGENT["Agent loop<br/>LLM plus tools"]
    GUARD{"Guardrails<br/>and policy"}
    EXEC["Execute and<br/>verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus<br/>next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff

User → [Single Agent + Tools] → Response

from openai import OpenAI
import json

client = OpenAI()

def single_agent(user_input: str, tools: list, tool_executor) -> str:
    messages = [
        {"role": "system", "content": "You are a helpful agent with access to tools."},
        {"role": "user", "content": user_input},
    ]

    for _ in range(15):
        response = client.chat.completions.create(
            model="gpt-4o", messages=messages, tools=tools,
        )
        msg = response.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            return msg.content

        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            result = tool_executor(tc.function.name, args)
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)})

    return "Max iterations reached."

Pros: Simple to build, debug, and maintain. Low latency. One set of instructions to manage.

Cons: Degrades with many tools (>15). Cannot specialize in multiple domains. Single point of failure.

Use when: The task involves a single domain, requires fewer than 15 tools, and can be completed in under 10 steps.

Architecture 2: Pipeline (Sequential)

Multiple agents execute in sequence, each handling one stage of a workflow. The output of one agent feeds into the next.

User → [Agent A: Research] → [Agent B: Analyze] → [Agent C: Write] → Response

from dataclasses import dataclass

@dataclass
class PipelineStage:
    name: str
    instructions: str
    tools: list

def pipeline_agent(user_input: str, stages: list[PipelineStage], tool_executor) -> str:
    """Execute agents sequentially, passing context forward."""
    accumulated_context = user_input

    for stage in stages:
        messages = [
            {"role": "system", "content": stage.instructions},
            {"role": "user", "content": accumulated_context},
        ]

        # Run this stage's agent loop
        for _ in range(10):
            response = client.chat.completions.create(
                model="gpt-4o", messages=messages,
                tools=stage.tools if stage.tools else None,
            )
            msg = response.choices[0].message
            messages.append(msg)

            if not msg.tool_calls:
                accumulated_context = (
                    f"Previous stage ({stage.name}) output:\n{msg.content}\n\n"
                    f"Original request: {user_input}"
                )
                break

            for tc in msg.tool_calls:
                args = json.loads(tc.function.arguments)
                result = tool_executor(tc.function.name, args)
                messages.append({
                    "role": "tool", "tool_call_id": tc.id,
                    "content": json.dumps(result),
                })

    return accumulated_context

# Define the pipeline
stages = [
    PipelineStage(
        name="researcher",
        instructions="Research the topic thoroughly using available tools. Produce raw findings.",
        tools=research_tools,
    ),
    PipelineStage(
        name="analyst",
        instructions="Analyze the research findings. Identify key patterns and insights.",
        tools=analysis_tools,
    ),
    PipelineStage(
        name="writer",
        instructions="Write a polished report based on the analysis. Use clear structure.",
        tools=[],  # Writer uses no tools, just generates text
    ),
]

Pros: Clear separation of concerns. Each agent is specialized and focused. Easy to test stages independently.

Cons: Rigid — cannot skip stages or go back. Information loss between stages. Total latency is the sum of all stages.

Use when: The workflow has clear, sequential phases that naturally build on each other (research, analyze, write; extract, transform, load).

Architecture 3: Router (Dynamic Dispatch)

A routing agent examines the input and dispatches to the appropriate specialist agent. This is a fan-out pattern where different types of requests go to different handlers.

              ┌→ [Billing Agent]
User → [Router] → [Technical Agent]
              └→ [Account Agent]

def router_agent(user_input: str, specialist_agents: dict) -> str:
    """Route to the appropriate specialist based on input classification."""

    # Step 1: Classify the input
    classification = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": (
                "Classify the user's request into one of these categories: "
                + ", ".join(specialist_agents.keys()) +
                ". Return JSON with a 'category' field."
            )},
            {"role": "user", "content": user_input},
        ],
        response_format={"type": "json_object"},
    )

    category = json.loads(classification.choices[0].message.content).get("category")

    if category not in specialist_agents:
        return f"I could not determine the right specialist for your request."

    # Step 2: Dispatch to the specialist
    specialist = specialist_agents[category]
    return single_agent(user_input, specialist["tools"], specialist["executor"])

# Define specialists
specialists = {
    "billing": {
        "tools": billing_tools,
        "executor": billing_tool_executor,
    },
    "technical": {
        "tools": technical_tools,
        "executor": technical_tool_executor,
    },
    "account": {
        "tools": account_tools,
        "executor": account_tool_executor,
    },
}

Pros: Each specialist has a focused tool set (better accuracy). Scales to many domains by adding specialists. Clean separation of concerns.

Cons: Classification errors route to the wrong specialist. Does not handle cross-domain requests well. Extra latency from the routing step.

Use when: Requests fall into distinct categories that need different tools and expertise (customer support with billing, technical, and account departments).

Architecture 4: Swarm (Collaborative Multi-Agent)

Multiple agents work together dynamically, handing off to each other based on the evolving state of the conversation. Any agent can transfer control to any other agent.

User → [Triage Agent] ⇄ [Specialist A] ⇄ [Specialist B] → Response

@dataclass
class SwarmAgent:
    name: str
    instructions: str
    tools: list
    handoff_targets: list[str]  # Names of agents this agent can hand off to

def swarm_run(user_input: str, agents: dict[str, SwarmAgent], start_agent: str) -> str:
    """Run a swarm of agents that can hand off to each other."""
    current_agent_name = start_agent
    messages = [{"role": "user", "content": user_input}]

    for _ in range(30):  # Global step limit
        agent = agents[current_agent_name]
        agent_messages = [
            {"role": "system", "content": (
                f"{agent.instructions}\n\n"
                f"You can hand off to: {', '.join(agent.handoff_targets)}. "
                f"To hand off, call the transfer_to_agent tool."
            )},
        ] + messages

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=agent_messages,
            tools=agent.tools + [transfer_tool],
        )
        msg = response.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            return msg.content

        for tc in msg.tool_calls:
            if tc.function.name == "transfer_to_agent":
                target = json.loads(tc.function.arguments)["agent_name"]
                if target in agents:
                    current_agent_name = target
                    messages.append({
                        "role": "tool", "tool_call_id": tc.id,
                        "content": f"Transferred to {target}",
                    })
            else:
                args = json.loads(tc.function.arguments)
                result = execute_tool(tc.function.name, args)
                messages.append({
                    "role": "tool", "tool_call_id": tc.id,
                    "content": json.dumps(result),
                })

    return "Swarm reached maximum iterations."

Pros: Highly flexible — agents collaborate dynamically. Handles cross-domain requests naturally. Mirrors real team structures.

Cons: Most complex to build and debug. Handoff loops are hard to prevent. Conversation context grows rapidly.

Use when: Tasks require expertise from multiple domains within a single conversation, and the path between domains cannot be predicted in advance.

Quick Comparison

Architecture	Complexity	Best For	Latency	Debuggability
Single Agent	Low	Simple, single-domain tasks	Low	High
Pipeline	Medium	Sequential workflows	Medium-High	High
Router	Medium	Multi-category classification	Medium	Medium
Swarm	High	Dynamic multi-domain collaboration	Variable	Low

FAQ

Should I start with a swarm architecture to be future-proof?

No. Start with the simplest architecture that solves your problem — usually a single agent. Graduate to more complex architectures only when you hit concrete limitations. Over-architecting with a swarm when a single agent suffices adds debugging complexity, increases costs, and slows development with no measurable benefit.

Can I combine these architectures?

Absolutely. Real production systems often combine patterns. A router that dispatches to specialist pipelines is common. A swarm where individual agents use the plan-and-execute pattern internally is another. Think of these as composable building blocks, not mutually exclusive choices.

How do I debug issues in multi-agent systems?

Implement comprehensive logging at three levels: agent-level (which agent is active, what instructions it has), message-level (every message in the conversation), and tool-level (every tool call with inputs and outputs). OpenAI's Agents SDK provides built-in tracing that captures all of this automatically, which is one of its biggest advantages over hand-rolled multi-agent systems.

---

#AgentArchitecture #MultiAgent #DesignPatterns #AIAgents #Python #AgenticAI #LearnAI #AIEngineering