Multi-Agent Orchestration Patterns: Supervisor vs Swarm vs Hierarchical

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langchain_openai import ChatOpenAI

# Define shared state
class AgentState(TypedDict):
    messages: list
    next_agent: str
    research_output: str
    draft_output: str
    final_output: str

# Initialize model
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Supervisor node: decides routing
def supervisor_node(state: AgentState) -> Command:
    system_prompt = """You are a Supervisor coordinating these agents:
    - researcher: information gathering and data collection
    - writer: content creation
    - critic: review and improvement
    
    Based on current progress, decide which agent should act next,
    or return FINISH if the task is complete."""
    
    response = llm.invoke([
        {"role": "system", "content": system_prompt},
        *state["messages"]
    ])
    
    next_agent = parse_routing_decision(response.content)
    
    if next_agent == "FINISH":
        return Command(goto=END, update={"final_output": state["draft_output"]})
    
    return Command(goto=next_agent, update={"next_agent": next_agent})

# Researcher Agent node
def researcher_node(state: AgentState) -> Command:
    response = llm.invoke([
        {"role": "system", "content": "You are a research expert. Gather and organize information."},
        {"role": "user", "content": state["messages"][-1]["content"]}
    ])
    return Command(
        goto="supervisor",
        update={
            "research_output": response.content,
            "messages": state["messages"] + [
                {"role": "assistant", "content": response.content}
            ]
        }
    )

# Writer Agent node
def writer_node(state: AgentState) -> Command:
    response = llm.invoke([
        {"role": "system", "content": "You are a writing expert. Produce high-quality content from research."},
        {"role": "user", "content": f"Write based on this research:\n{state['research_output']}"}
    ])
    return Command(
        goto="supervisor",
        update={
            "draft_output": response.content,
            "messages": state["messages"] + [
                {"role": "assistant", "content": response.content}
            ]
        }
    )

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("critic", critic_node)

workflow.add_edge(START, "supervisor")

app = workflow.compile()

# Execute
result = app.invoke({
    "messages": [{"role": "user", "content": "Write a technical analysis of AI agent orchestration patterns"}],
    "next_agent": "",
    "research_output": "",
    "draft_output": "",
    "final_output": ""
})

import { StateGraph, START, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { BaseMessage } from "@langchain/core/messages";

// Define state type
interface AgentState {
  messages: BaseMessage[];
  nextAgent: string;
  researchOutput: string;
  draftOutput: string;
}

const llm = new ChatOpenAI({ model: "gpt-4o", temperature: 0 });

// Supervisor node
async function supervisorNode(state: AgentState) {
  const response = await llm.invoke([
    { role: "system", content: "Route to: researcher | writer | critic | FINISH" },
    ...state.messages,
  ]);

  const nextAgent = parseRoute(response.content as string);
  return { nextAgent };
}

// Conditional routing
function routeFromSupervisor(state: AgentState): string {
  if (state.nextAgent === "FINISH") return END;
  return state.nextAgent;
}

// Build graph
const workflow = new StateGraph<AgentState>({
  channels: {
    messages: { default: () => [] },
    nextAgent: { default: () => "" },
    researchOutput: { default: () => "" },
    draftOutput: { default: () => "" },
  },
});

workflow.addNode("supervisor", supervisorNode);
workflow.addNode("researcher", researcherNode);
workflow.addNode("writer", writerNode);

workflow.addEdge(START, "supervisor");
workflow.addConditionalEdges("supervisor", routeFromSupervisor);

const app = workflow.compile();

from swarm import Swarm, Agent

client = Swarm()

# Define handoff functions
def transfer_to_tech_support():
    """Transfer user to technical support agent"""
    return tech_agent

def transfer_to_order_agent():
    """Transfer user to order processing agent"""
    return order_agent

def escalate_to_human():
    """Escalate to human support"""
    return "ESCALATE: Transferred to human agent, ticket #" + generate_ticket_id()

# Triage Agent - entry point
triage_agent = Agent(
    name="Triage Agent",
    instructions="""You are a customer service triage agent. Route by issue type:
    - Technical issues (installation, config, bugs) -> transfer to tech support
    - Order issues (refunds, shipping, payment) -> transfer to order agent
    - Unclassifiable -> escalate to human""",
    functions=[transfer_to_tech_support, transfer_to_order_agent, escalate_to_human],
)

# Tech Support Agent
tech_agent = Agent(
    name="Tech Support",
    instructions="""You are a technical support specialist. Resolve technical issues.
    If the issue involves refunds or orders, call transfer_to_order_agent.
    If the issue exceeds your capabilities, call escalate_to_human.""",
    functions=[transfer_to_order_agent, escalate_to_human],
    model="gpt-4o",
)

# Order Agent
order_agent = Agent(
    name="Order Agent",
    instructions="""You are an order specialist. Handle refunds, shipping, payment.
    If troubleshooting is needed, call transfer_to_tech_support.""",
    functions=[transfer_to_tech_support, escalate_to_human,
              process_refund, check_order_status],
)

# Run conversation
response = client.run(
    agent=triage_agent,
    messages=[{"role": "user", "content": "My order shows shipped but hasn't arrived in 3 days, and the app keeps crashing"}],
)

print(response.messages[-1]["content"])
# Agents automatically flow between triage -> order -> tech as needed

interface SwarmAgent {
  name: string;
  instructions: string;
  functions: AgentFunction[];
  model?: string;
}

interface AgentFunction {
  name: string;
  description: string;
  handler: (args: any) => SwarmAgent | string;
}

class SwarmOrchestrator {
  private currentAgent: SwarmAgent;
  private conversationHistory: Message[] = [];
  private maxHandoffs = 10; // Prevent infinite loops

  constructor(entryAgent: SwarmAgent) {
    this.currentAgent = entryAgent;
  }

  async run(userMessage: string): Promise<string> {
    this.conversationHistory.push({ role: "user", content: userMessage });
    let handoffCount = 0;

    while (handoffCount < this.maxHandoffs) {
      const response = await this.callLLM(this.currentAgent);

      // Check for handoff
      if (response.functionCall) {
        const fn = this.currentAgent.functions.find(
          f => f.name === response.functionCall!.name
        );
        const result = fn!.handler(response.functionCall!.arguments);

        if (typeof result === "object" && "name" in result) {
          // Handoff to new agent
          console.log(`[Handoff] ${this.currentAgent.name} -> ${result.name}`);
          this.currentAgent = result;
          handoffCount++;
          continue;
        }
        // Regular function call result
        this.conversationHistory.push({ role: "function", content: result });
        continue;
      }

      // No handoff—return final answer
      return response.content;
    }

    throw new Error("Exceeded maximum handoff limit");
  }
}

from crewai import Agent, Task, Crew, Process

# Top-level Manager (automatically managed by CrewAI hierarchical process)
manager_llm = "gpt-4o"

# Mid-level Team Lead Agents
research_lead = Agent(
    role="Research Team Lead",
    goal="Coordinate the research team to collect and verify all relevant data",
    backstory="You are a senior research director skilled at distributing search tasks and cross-validating information.",
    llm="gpt-4o",
    allow_delegation=True,  # Can delegate downward
)

engineering_lead = Agent(
    role="Engineering Team Lead",
    goal="Coordinate the engineering team to ensure code quality and architecture soundness",
    backstory="You are a technical director responsible for coordinating frontend, backend, and DevOps.",
    llm="gpt-4o",
    allow_delegation=True,
)

# Bottom-level Worker Agents
web_researcher = Agent(
    role="Web Researcher",
    goal="Search and extract the latest technical literature from the web",
    backstory="You specialize in web information retrieval and can quickly locate high-quality sources.",
    llm="gpt-4o-mini",  # Workers use smaller models to reduce cost
    allow_delegation=False,
)

data_analyst = Agent(
    role="Data Analyst",
    goal="Analyze data and generate actionable insights",
    backstory="You are a data analysis expert skilled in statistical analysis and trend identification.",
    llm="gpt-4o-mini",
    allow_delegation=False,
)

backend_dev = Agent(
    role="Backend Developer",
    goal="Implement backend APIs and business logic",
    backstory="You are a senior backend engineer proficient in Python and distributed systems.",
    llm="gpt-4o-mini",
    allow_delegation=False,
)

# Define tasks
research_task = Task(
    description="Research latest advances in multi-agent orchestration patterns including papers, open-source projects, and enterprise practices",
    expected_output="Structured research report with at least 10 key findings",
    agent=research_lead,
)

implementation_task = Task(
    description="Based on research results, design and implement a multi-pattern orchestration engine prototype",
    expected_output="Runnable prototype code and architecture documentation",
    agent=engineering_lead,
    context=[research_task],  # Depends on research task output
)

# Build hierarchical Crew
crew = Crew(
    agents=[research_lead, engineering_lead, web_researcher, 
            data_analyst, backend_dev],
    tasks=[research_task, implementation_task],
    process=Process.hierarchical,  # Key: enable hierarchical mode
    manager_llm=manager_llm,
    verbose=True,
)

# Execute
result = crew.kickoff()
print(result)

import { AutoGenGroupChat, Agent, UserProxy } from "autogen";

// Define hierarchy
const projectManager = new Agent({
  name: "ProjectManager",
  systemMessage: `You are the project manager. Decompose complex tasks into subtasks,
    assign them to appropriate Team Leads. Monitor progress and handle cross-team dependencies.`,
  model: "gpt-4o",
});

const researchLead = new Agent({
  name: "ResearchLead",
  systemMessage: `You are the research lead. Receive research tasks from PM,
    break them into specific retrieval tasks for Workers, aggregate results and report up.`,
  model: "gpt-4o",
});

const webResearcher = new Agent({
  name: "WebResearcher",
  systemMessage: "You are a web researcher. Execute specific search and extraction tasks.",
  model: "gpt-4o-mini",
});

// Multi-level GroupChat configuration
const researchTeam = new AutoGenGroupChat({
  agents: [researchLead, webResearcher],
  maxRound: 5,
  speakerSelectionMethod: "round_robin",
});

const topLevelChat = new AutoGenGroupChat({
  agents: [projectManager, researchLead],
  maxRound: 10,
  speakerSelectionMethod: "auto",
  nestedChats: {
    ResearchLead: researchTeam, // Nested sub-team
  },
});

await topLevelChat.initiate(
  "Design an agent framework that supports hot-switching between three orchestration patterns"
);

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
async def call_agent_with_timeout(agent_fn, state, timeout=30):
    """Agent call with timeout and exponential backoff retry"""
    try:
        result = await asyncio.wait_for(agent_fn(state), timeout=timeout)
        return result
    except asyncio.TimeoutError:
        logger.warning(f"Agent {agent_fn.__name__} timed out after {timeout}s")
        raise
    except Exception as e:
        logger.error(f"Agent {agent_fn.__name__} failed: {e}")
        raise

from opentelemetry import trace
from opentelemetry.trace import StatusCode

tracer = trace.get_tracer("multi-agent-orchestrator")

def traced_agent_call(agent_name: str):
    """Create a Span for each agent invocation"""
    def decorator(fn):
        async def wrapper(state):
            with tracer.start_as_current_span(f"agent.{agent_name}") as span:
                span.set_attribute("agent.name", agent_name)
                span.set_attribute("agent.input_tokens", count_tokens(state))
                try:
                    result = await fn(state)
                    span.set_attribute("agent.output_tokens", count_tokens(result))
                    span.set_status(StatusCode.OK)
                    return result
                except Exception as e:
                    span.set_status(StatusCode.ERROR, str(e))
                    span.record_exception(e)
                    raise
        return wrapper
    return decorator