Lesson 13 of 23

Multi-Agent System Design

Agent Architecture Patterns

4 min read

AI agents are systems that use LLMs to reason, plan, and take actions. Understanding different architecture patterns is essential for designing scalable agent systems.

Single Agent Pattern

The simplest pattern—one agent handles the entire task.

class SingleAgent:
    def __init__(self, llm, tools: list):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}

    async def run(self, task: str) -> str:
        messages = [{"role": "user", "content": task}]

        while True:
            response = await self.llm.complete(
                messages=messages,
                tools=list(self.tools.values())
            )

            if response.tool_calls:
                # Execute tools
                for call in response.tool_calls:
                    result = await self.tools[call.name].execute(call.args)
                    messages.append({
                        "role": "tool",
                        "content": result,
                        "tool_call_id": call.id
                    })
            else:
                # Final response
                return response.content

Best for: Simple tasks, < 5 tools, clear single-path solutions.

Multi-Agent Patterns

1. Coordinator Pattern

A central coordinator delegates tasks to specialized agents.

┌─────────────────────────────────────────────────────────────┐
│                     Coordinator Agent                        │
│                                                              │
│  "Break down task, assign to specialists, synthesize"       │
│                                                              │
├─────────────┬─────────────┬─────────────┬──────────────────┤
│             │             │             │                   │
▼             ▼             ▼             ▼                   │
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐              │
│Research │ │  Code   │ │ Review  │ │  Test   │              │
│ Agent   │ │ Agent   │ │ Agent   │ │ Agent   │              │
└─────────┘ └─────────┘ └─────────┘ └─────────┘              │
└──────────────── Results back to Coordinator ────────────────┘
class CoordinatorAgent:
    def __init__(self, llm, specialist_agents: dict):
        self.llm = llm
        self.specialists = specialist_agents

    async def run(self, task: str) -> str:
        # Step 1: Plan the task
        plan = await self._create_plan(task)

        # Step 2: Execute each step with appropriate specialist
        results = []
        for step in plan.steps:
            agent = self.specialists[step.agent_type]
            result = await agent.run(step.instruction)
            results.append({
                "step": step.name,
                "result": result
            })

        # Step 3: Synthesize results
        return await self._synthesize(task, results)

    async def _create_plan(self, task: str) -> Plan:
        prompt = f"""Break down this task into steps.
For each step, specify which specialist should handle it.

Available specialists: {list(self.specialists.keys())}

Task: {task}

Return a JSON plan."""

        response = await self.llm.complete(prompt)
        return Plan.parse(response)

2. Pipeline Pattern

Agents process sequentially, each building on the previous output.

class PipelineAgent:
    def __init__(self, stages: list):
        self.stages = stages  # List of (agent, transform_fn)

    async def run(self, initial_input: str) -> str:
        current_output = initial_input

        for agent, transform in self.stages:
            # Transform input for this stage
            stage_input = transform(current_output)

            # Run agent
            current_output = await agent.run(stage_input)

        return current_output

# Example: Document processing pipeline
pipeline = PipelineAgent([
    (extraction_agent, lambda x: f"Extract key information: {x}"),
    (analysis_agent, lambda x: f"Analyze this information: {x}"),
    (summary_agent, lambda x: f"Summarize the analysis: {x}")
])

3. Debate/Critic Pattern

Multiple agents discuss and refine answers.

class DebateSystem:
    def __init__(self, proposer, critic, judge):
        self.proposer = proposer
        self.critic = critic
        self.judge = judge

    async def run(self, question: str, max_rounds: int = 3) -> str:
        # Initial proposal
        proposal = await self.proposer.run(question)

        for round in range(max_rounds):
            # Critic reviews
            critique = await self.critic.run(
                f"Question: {question}\nProposal: {proposal}\n"
                "Identify weaknesses and suggest improvements."
            )

            # Check if critique is satisfied
            if "no significant issues" in critique.lower():
                break

            # Proposer revises
            proposal = await self.proposer.run(
                f"Question: {question}\n"
                f"Previous answer: {proposal}\n"
                f"Critique: {critique}\n"
                "Revise your answer addressing the critique."
            )

        # Judge makes final decision
        return await self.judge.run(
            f"Question: {question}\n"
            f"Final proposal: {proposal}\n"
            "Provide the final, polished answer."
        )

Choosing the Right Pattern

Pattern Use Case Complexity Latency
Single Agent Simple tasks, few tools Low Low
Coordinator Complex tasks, multiple domains Medium Medium
Pipeline Sequential processing Low High
Debate High-stakes decisions High Very High

Interview Tip

When designing agent systems in interviews, always consider:

  1. Task decomposition - Can it be broken into subtasks?
  2. Tool selection - What capabilities does each agent need?
  3. Failure handling - What if an agent fails mid-task?
  4. Latency budget - Multi-agent adds latency

Next, we'll explore tool registry design for dynamic agent capabilities. :::

Quiz

Module 4: Multi-Agent System Design

Take Quiz