A multi-agent system uses multiple LLM-powered agents, each with specialized roles, that collaborate to solve complex tasks that a single agent cannot handle as effectively.
Single LLM = generalist. Multi-agent = specialists. Break complex tasks into smaller, manageable pieces handled by experts.
| Architecture | Description | Best for |
|---|---|---|
| Chain (Sequential) | Agent A → Agent B → Agent C | Linear workflows (research → write → edit) |
| Parallel (Broadcast) | All agents work simultaneously | Independent tasks, diversity of outputs |
| Hierarchical | Supervisor → Worker agents | Complex tasks needing coordination |
| Peer-to-Peer | Agents negotiate and collaborate | Markets, simulations, debates |
class Agent:
def __init__(self, name, role, instructions, tools=None):
self.name = name
self.role = role
self.instructions = instructions
self.tools = tools or []
self.memory = []
def execute(self, task, context=None):
prompt = f"""
You are {self.role}. {self.instructions}
Context: {context}
Task: {task}
"""
response = llm.invoke(prompt)
self.memory.append({"task": task, "response": response})
return responsedef sequential_workflow(topic):
# Step 1: Research
researcher = Agent("Researcher", "research specialist",
"Gather key information about the topic")
research = researcher.execute(topic)
# Step 2: Write
writer = Agent("Writer", "content writer",
"Write a clear, engaging article based on research")
draft = writer.execute(topic, context=f"Research: {research}")
# Step 3: Edit
editor = Agent("Editor", "professional editor",
"Improve clarity, grammar, and flow")
final = editor.execute(draft)
return final
def parallel_brainstorm(topic, num_agents=3):
agents = [
Agent(f"Agent_{i}", "creative thinker",
"Generate unique ideas for the given topic")
for i in range(num_agents)
]
# All agents work simultaneously
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor() as executor:
results = list(executor.map(lambda a: a.execute(topic), agents))
# Combine and deduplicate
return combine_results(results)
def debate(refine_prompt, rounds=3):
advocate = Agent("Advocate", "supporter",
"Defend the position and strengthen it")
critic = Agent("Critic", "devil's advocate",
"Find flaws and weaknesses in the argument")
position = refine_prompt
for round in range(rounds):
# Advocate strengthens
position = advocate.execute(position)
# Critic finds flaws
flaws = critic.execute(position)
# Address flaws in next round
position = f"{position}\n\nAddressing criticisms: {flaws}"
return position
A central supervisor agent decomposes tasks, assigns them to specialized workers, and synthesizes results.
class Supervisor:
def __init__(self, workers):
self.workers = workers
def plan(self, objective):
# Supervisor breaks down the task
prompt = f"""
Break down this objective into 3-5 subtasks:
Objective: {objective}
For each subtask, specify:
- Which worker should handle it
- What information they need
Return as a list of (worker_name, subtask, context)
"""
plan = llm.invoke(prompt)
return parse_plan(plan)
def execute_plan(self, plan):
results = {}
for worker_name, subtask, context in plan:
worker = self.workers[worker_name]
result = worker.execute(subtask, context)
results[worker_name] = result
# Synthesize final answer
synthesis_prompt = f"""
Based on these worker results, produce a final answer:
{results}
"""
return llm.invoke(synthesis_prompt)
Supervisor: "Customer wants to return a defective product"
↓
Worker 1 (Policy Expert): "30-day return window, free shipping"
Worker 2 (Sentiment Analyzer): "Customer is frustrated"
Worker 3 (Action Taker): "Generate return label"
Worker 4 (Writer): "Apologize and explain next steps"def ensemble_vote(task, agents):
# Each agent contributes an answer
answers = []
for agent in agents:
answer = agent.execute(task)
answers.append(answer)
# Voting prompt to select best
vote_prompt = f"""
Task: {task}
Answers from multiple experts:
{answers}
Which answer is most accurate and helpful? Explain your choice.
"""
return llm.invoke(vote_prompt)
from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
# Create specialized tools
research_tool = Tool(name="Research", func=research_agent,
description="For gathering information")
writing_tool = Tool(name="Writer", func=writing_agent,
description="For creating content")
# Agents can call these tools
agent = initialize_agent(
[research_tool, writing_tool],
llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
Design a 3-agent system for writing blog posts: Planner, Writer, and Editor/SEO. Describe each agent's role and how they interact.
Write a function that sets up a debate between a "pro" and "con" agent on any topic, then determines the winner.
def debate(topic):
pro = Agent("Pro", "argue for the proposition")
con = Agent("Con", "argue against the proposition")
pro_argument = pro.execute(topic)
con_rebuttal = con.execute(pro_argument)
pro_rebuttal = pro.execute(con_rebuttal)
judge = Agent("Judge", "determine which argument is stronger")
winner = judge.execute(f"Pro: {pro_argument}\nCon: {con_rebuttal}")
return winner
Create a multi-agent system where 3 agents review a piece of code and vote on its quality. What criteria would they evaluate?