Multi-Agent Systems

Ask one model to gather information, analyze it, write an article and then criticise it too. You will get a text where everything is a little bit there and nothing is done properly, because a single prompt pulls the model in four directions at once. On the savanna nobody does everything alone either.

Multi-agent systems are an architecture where multiple specialized AI agents collaborate to solve complex tasks. It is like a herd of animals where each individual has its role - hunters, lookouts, defenders. Each agent gets its own system prompt and does one thing well.

Multi-Agent Basics

First let's agree on a vocabulary. We store roles as an Enum, a closed set of named constants, and a message between agents as a @dataclass:

1from dataclasses import dataclass
2from enum import Enum
3from abc import ABC, abstractmethod
4from openai import OpenAI
5
6class AgentRole(Enum):
7    """Agent roles in the system."""
8    RESEARCHER = "researcher"      # Gathers information
9    ANALYZER = "analyzer"          # Analyzes data
10    WRITER = "writer"              # Writes content
11    CRITIC = "critic"              # Reviews and improves
12    COORDINATOR = "coordinator"    # Coordinates work
13
14@dataclass
15class AgentMessage:
16    """Message between agents."""
17    sender: str
18    receiver: str
19    content: str
20    message_type: str = "task"

AgentMessage remembers the sender, receiver, content and message type. The default type is "task", so for an ordinary assignment you do not have to pass it.

Now the base agent class. It inherits from ABC, and the system_prompt property is marked with @abstractmethod, so every subclass has to define it:

1class BaseAgent(ABC):
2    """Base agent class."""
3
4    def __init__(self, name: str, role: AgentRole, model: str = "gpt-4o-mini"):
5        self.name = name
6        self.role = role
7        self.model = model
8        self.client = OpenAI()
9        self.memory: list[AgentMessage] = []
10
11    @property
12    @abstractmethod
13    def system_prompt(self) -> str:
14        """Agent's system prompt."""
15        pass
16
17    def process(self, message: AgentMessage) -> str:
18        """Processes a message and returns a response."""
19        self.memory.append(message)
20
21        response = self.client.chat.completions.create(
22            model=self.model,
23            messages=[
24                {"role": "system", "content": self.system_prompt},
25                {"role": "user", "content": message.content}
26            ]
27        )
28
29        return response.choices[0].message.content

The process method stores the message in the agent's memory and sends two messages to the model: the system prompt with the role and the task content. You cannot create an object of BaseAgent itself, Python raises a TypeError because the system_prompt implementation is missing.

Implementing Specialized Agents

Specialized agents differ only in their prompt. The first pair is the researcher and the analyzer:

1class ResearcherAgent(BaseAgent):
2    """Research agent - gathers information."""
3
4    @property
5    def system_prompt(self) -> str:
6        return """You are a research agent. Your role is to:
71. Gather information on the given topic
82. Identify key facts
93. Verify sources
104. Structure knowledge
11
12Always provide sources and categorize information."""
13
14class AnalyzerAgent(BaseAgent):
15    """Analytical agent - analyzes data."""
16
17    @property
18    def system_prompt(self) -> str:
19        return """You are an analytical agent. Your role is to:
201. Analyze provided data
212. Identify patterns and trends
223. Formulate conclusions
234. Assess risks and opportunities
24
25Be precise and provide specific numbers."""

Each class overrides only system_prompt, and the whole model-calling logic stays in BaseAgent. That is inheritance in practice: shared code once, differences in subclasses.

The second pair is the writer and the critic:

1class WriterAgent(BaseAgent):
2    """Writer agent - creates content."""
3
4    @property
5    def system_prompt(self) -> str:
6        return """You are a writer agent. Your role is to:
71. Create engaging content
82. Adapt style to the audience
93. Structure the text
104. Ensure clarity of message
11
12Write concisely and stay on topic."""
13
14class CriticAgent(BaseAgent):
15    """Critic agent - reviews and improves."""
16
17    @property
18    def system_prompt(self) -> str:
19        return """You are a critic agent. Your role is to:
201. Evaluate content quality
212. Identify errors and shortcomings
223. Suggest improvements
234. Verify facts
24
25Be constructive but honest."""

The Researcher gathers information, the Analyzer analyzes it, the Writer creates content, and the Critic reviews and improves. These are typical roles in a multi-agent system. Notice that a "Destroyer that deletes files" role never appears here: an agent is there to help, not to destroy.

Agent Orchestration

Someone has to pass messages between members of the herd. That is the orchestrator's job: it registers agents and mediates every conversation:

1from typing import Optional
2
3class AgentOrchestrator:
4    """Coordinator of agent work."""
5
6    def __init__(self):
7        self.agents: dict[str, BaseAgent] = {}
8        self.conversation_history: list[AgentMessage] = []
9
10    def register_agent(self, agent: BaseAgent) -> None:
11        """Registers an agent in the system."""
12        self.agents[agent.name] = agent
13        print(f"Registered agent: {agent.name} ({agent.role.value})")
14
15    def send_message(
16        self,
17        sender: str,
18        receiver: str,
19        content: str
20    ) -> str:
21        """Sends a message between agents."""
22        if receiver not in self.agents:
23            raise ValueError(f"Agent {receiver} does not exist!")
24
25        message = AgentMessage(
26            sender=sender,
27            receiver=receiver,
28            content=content
29        )
30
31        self.conversation_history.append(message)
32
33        response = self.agents[receiver].process(message)
34
35        # Save response
36        response_message = AgentMessage(
37            sender=receiver,
38            receiver=sender,
39            content=response,
40            message_type="response"
41        )
42        self.conversation_history.append(response_message)
43
44        return response

send_message checks that the receiver exists, stores the message in the history, calls process and stores the response as well. This keeps the whole conversation in one place.

The run_pipeline method arranges the agents in a chain where the output of one becomes the input of the next:

1    def run_pipeline(self, task: str) -> dict[str, str]:
2        """Runs the processing pipeline."""
3        results = {}
4
5        # 1. Researcher gathers information
6        research = self.send_message("user", "researcher",
7            f"Gather information on: {task}")
8        results["research"] = research
9
10        # 2. Analyzer analyzes
11        analysis = self.send_message("researcher", "analyzer",
12            f"Analyze this information:\n{research}")
13        results["analysis"] = analysis
14
15        # 3. Writer creates content
16        draft = self.send_message("analyzer", "writer",
17            f"Based on the analysis, write an article:\n{analysis}")
18        results["draft"] = draft
19
20        # 4. Critic reviews
21        review = self.send_message("writer", "critic",
22            f"Review this article:\n{draft}")
23        results["review"] = review
24
25        return results

The order is fixed: the Researcher gathers information, the Analyzer analyzes the data, the Writer creates content, the Critic reviews. The results dictionary keeps the output of every stage, not only the last one.

All that is left is to register the four agents and run the pipeline:

1# Usage
2orchestrator = AgentOrchestrator()
3orchestrator.register_agent(ResearcherAgent("researcher", AgentRole.RESEARCHER))
4orchestrator.register_agent(AnalyzerAgent("analyzer", AgentRole.ANALYZER))
5orchestrator.register_agent(WriterAgent("writer", AgentRole.WRITER))
6orchestrator.register_agent(CriticAgent("critic", AgentRole.CRITIC))
7
8results = orchestrator.run_pipeline("The future of artificial intelligence in 2025")

Each agent is stored under its name, so send_message("user", "researcher", ...) lands exactly where it should.

Asynchronous Communication

Agents do not have to talk strictly in turn. You can leave them messages in an asyncio.Queue, like notes at the watering hole:

1import asyncio
2from typing import Callable
3
4class AsyncAgentSystem:
5    """Asynchronous agent system."""
6
7    def __init__(self):
8        self.agents: dict[str, BaseAgent] = {}
9        self.message_queue: asyncio.Queue = asyncio.Queue()
10        self.results: dict[str, str] = {}
11
12    async def process_messages(self) -> None:
13        """Processes messages from the queue."""
14        while True:
15            message = await self.message_queue.get()
16
17            if message.content == "STOP":
18                break
19
20            agent = self.agents.get(message.receiver)
21            if agent:
22                response = agent.process(message)
23                self.results[message.receiver] = response
24
25            self.message_queue.task_done()

The process_messages loop waits for a message with await, handles it, and stops when it receives the "STOP" signal.

The run_parallel_tasks method puts tasks into the queue and starts processing:

1    async def run_parallel_tasks(self, tasks: list[tuple[str, str]]) -> dict:
2        """Runs tasks in parallel."""
3        # Add tasks to the queue
4        for agent_name, task in tasks:
5            await self.message_queue.put(
6                AgentMessage("system", agent_name, task)
7            )
8
9        # Add end signal
10        await self.message_queue.put(
11            AgentMessage("system", "STOP", "STOP")
12        )
13
14        # Start processing
15        await self.process_messages()
16
17        return self.results
18
19# Parallel processing example
20async def main():
21    system = AsyncAgentSystem()
22    # ... register agents ...
23
24    tasks = [
25        ("researcher", "Research topic X"),
26        ("analyzer", "Analyze data Y"),
27    ]
28
29    results = await system.run_parallel_tasks(tasks)
30    print(results)
31
32asyncio.run(main())

To be honest: despite the name, these tasks do not run in parallel. There is a single queue consumer, and agent.process is an ordinary blocking call. Real parallelism would come from the asynchronous AsyncOpenAI client with asyncio.gather, or from asyncio.to_thread. The queue is still useful for decoupling senders from receivers.

Agent Tools

An agent that can only write is like a tracker without binoculars. Tools let it act: search a database, fetch the weather, calculate a cost. In LangChain you turn a function into a tool with the @tool decorator:

1from langchain_core.tools import tool
2
3@tool
4def search_db(query: str) -> list:
5    """Searches the animal observation database."""
6    return [f"Observation matching: {query}"]
7
8print(search_db.name)              # search_db
9print(search_db.invoke("lions"))   # ['Observation matching: lions']

The docstring is not decoration here: the model reads it as the tool description, and without it the decorator raises an error. Parameter types tell the model which arguments to pass. An agent with tools works in a loop: it plans an action, acts, observes the result and reflects on whether to correct its course.

Agent Collaboration Patterns

Finally, three classic patterns. Chain is a sequence in which every agent passes its output to the next one:

1class CollaborationPatterns:
2    """Agent collaboration patterns."""
3
4    @staticmethod
5    def chain(agents: list[BaseAgent], initial_input: str) -> str:
6        """Chain - each agent passes output to the next."""
7        current_output = initial_input
8
9        for agent in agents:
10            message = AgentMessage("system", agent.name, current_output)
11            current_output = agent.process(message)
12
13        return current_output

It generalizes our run_pipeline: the list of agents can have any length.

Debate is a discussion between two agents over several rounds:

1    @staticmethod
2    def debate(agent1: BaseAgent, agent2: BaseAgent, topic: str, rounds: int = 3) -> list[str]:
3        """Debate - agents discuss a topic."""
4        conversation = []
5
6        current = f"Discussion on: {topic}. Present your position."
7
8        for i in range(rounds):
9            # Agent 1
10            response1 = agent1.process(
11                AgentMessage(agent2.name, agent1.name, current)
12            )
13            conversation.append(f"{agent1.name}: {response1}")
14
15            # Agent 2 responds
16            response2 = agent2.process(
17                AgentMessage(agent1.name, agent2.name, response1)
18            )
19            conversation.append(f"{agent2.name}: {response2}")
20            current = response2
21
22        return conversation

Each answer from one agent becomes the input of the other, and the conversation list records the whole exchange.

Consensus gathers everyone's opinions and asks one agent for a synthesis:

1    @staticmethod
2    def consensus(agents: list[BaseAgent], question: str) -> str:
3        """Consensus - agents reach a common conclusion."""
4        opinions = []
5
6        # Collect opinions
7        for agent in agents:
8            opinion = agent.process(
9                AgentMessage("system", agent.name, question)
10            )
11            opinions.append(f"{agent.name}: {opinion}")
12
13        # Synthesis (use one of the agents)
14        synthesis_prompt = f"""Question: {question}
15
16Agent opinions:
17{chr(10).join(opinions)}
18
19Formulate a common consensus based on these opinions."""
20
21        return agents[0].process(
22            AgentMessage("system", agents[0].name, synthesis_prompt)
23        )

chr(10) is the newline character, used instead of "\n" inside the f-string. I recommend starting with a simple chain and adding debate and consensus only where one perspective really is not enough, because every round means more paid model calls.

Multi-agent systems open the door to complex AI applications. In the next lesson you will learn CrewAI - a framework for building agent teams that gives you these patterns ready-made!

Remember: a good agent system works like a herd in which everyone knows their role and the guide keeps the direction.