Multi-Agent Systems

Multi-agent systems are an architecture where multiple specialized AI agents collaborate to solve complex tasks. They are like a herd of animals, where each individual has its role - hunters, observers, defenders!

Multi-Agent Basics

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"
21
22class BaseAgent(ABC):
23    """Base agent class."""
24
25    def __init__(self, name: str, role: AgentRole, model: str = "gpt-4o-mini"):
26        self.name = name
27        self.role = role
28        self.model = model
29        self.client = OpenAI()
30        self.memory: list[AgentMessage] = []
31
32    @property
33    @abstractmethod
34    def system_prompt(self) -> str:
35        """Agent's system prompt."""
36        pass
37
38    def process(self, message: AgentMessage) -> str:
39        """Processes a message and returns a response."""
40        self.memory.append(message)
41
42        response = self.client.chat.completions.create(
43            model=self.model,
44            messages=[
45                {"role": "system", "content": self.system_prompt},
46                {"role": "user", "content": message.content}
47            ]
48        )
49
50        return response.choices[0].message.content

Implementing Specialist Agents

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."""
26
27class WriterAgent(BaseAgent):
28    """Writer agent - creates content."""
29
30    @property
31    def system_prompt(self) -> str:
32        return """You are a writer agent. Your role is to:
331. Create engaging content
342. Adapt style to the audience
353. Structure the text
364. Ensure clarity of message
37
38Write concisely and stay on topic."""
39
40class CriticAgent(BaseAgent):
41    """Critic agent - reviews and improves."""
42
43    @property
44    def system_prompt(self) -> str:
45        return """You are a critic agent. Your role is to:
461. Evaluate content quality
472. Identify errors and shortcomings
483. Suggest improvements
494. Verify facts
50
51Be constructive but honest."""

Agent Orchestration

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
45
46    def run_pipeline(self, task: str) -> dict[str, str]:
47        """Runs the processing pipeline."""
48        results = {}
49
50        # 1. Researcher gathers information
51        research = self.send_message("user", "researcher",
52            f"Gather information on: {task}")
53        results["research"] = research
54
55        # 2. Analyzer analyzes
56        analysis = self.send_message("researcher", "analyzer",
57            f"Analyze this information:\n{research}")
58        results["analysis"] = analysis
59
60        # 3. Writer creates content
61        draft = self.send_message("analyzer", "writer",
62            f"Based on the analysis, write an article:\n{analysis}")
63        results["draft"] = draft
64
65        # 4. Critic reviews
66        review = self.send_message("writer", "critic",
67            f"Review this article:\n{draft}")
68        results["review"] = review
69
70        return results
71
72# Usage
73orchestrator = AgentOrchestrator()
74orchestrator.register_agent(ResearcherAgent("researcher", AgentRole.RESEARCHER))
75orchestrator.register_agent(AnalyzerAgent("analyzer", AgentRole.ANALYZER))
76orchestrator.register_agent(WriterAgent("writer", AgentRole.WRITER))
77orchestrator.register_agent(CriticAgent("critic", AgentRole.CRITIC))
78
79results = orchestrator.run_pipeline("The future of artificial intelligence in 2025")

Asynchronous Communication

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()
26
27    async def run_parallel_tasks(self, tasks: list[tuple[str, str]]) -> dict:
28        """Runs tasks in parallel."""
29        # Add tasks to the queue
30        for agent_name, task in tasks:
31            await self.message_queue.put(
32                AgentMessage("system", agent_name, task)
33            )
34
35        # Add end signal
36        await self.message_queue.put(
37            AgentMessage("system", "STOP", "STOP")
38        )
39
40        # Start processing
41        await self.process_messages()
42
43        return self.results
44
45# Parallel processing example
46async def main():
47    system = AsyncAgentSystem()
48    # ... register agents ...
49
50    tasks = [
51        ("researcher", "Research topic X"),
52        ("analyzer", "Analyze data Y"),
53    ]
54
55    results = await system.run_parallel_tasks(tasks)
56    print(results)
57
58asyncio.run(main())

Agent Collaboration Patterns

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
14
15    @staticmethod
16    def debate(agent1: BaseAgent, agent2: BaseAgent, topic: str, rounds: int = 3) -> list[str]:
17        """Debate - agents discuss a topic."""
18        conversation = []
19
20        current = f"Discussion on: {topic}. Present your position."
21
22        for i in range(rounds):
23            # Agent 1
24            response1 = agent1.process(
25                AgentMessage(agent2.name, agent1.name, current)
26            )
27            conversation.append(f"{agent1.name}: {response1}")
28
29            # Agent 2 responds
30            response2 = agent2.process(
31                AgentMessage(agent1.name, agent2.name, response1)
32            )
33            conversation.append(f"{agent2.name}: {response2}")
34            current = response2
35
36        return conversation
37
38    @staticmethod
39    def consensus(agents: list[BaseAgent], question: str) -> str:
40        """Consensus - agents reach a common conclusion."""
41        opinions = []
42
43        # Collect opinions
44        for agent in agents:
45            opinion = agent.process(
46                AgentMessage("system", agent.name, question)
47            )
48            opinions.append(f"{agent.name}: {opinion}")
49
50        # Synthesis (use one of the agents)
51        synthesis_prompt = f"""Question: {question}
52
53Agent opinions:
54{chr(10).join(opinions)}
55
56Formulate a common consensus based on these opinions."""
57
58        return agents[0].process(
59            AgentMessage("system", agents[0].name, synthesis_prompt)
60        )

Multi-agent systems open the door to complex AI applications. In the next lesson you will learn about CrewAI - a framework for creating agent teams!