Systèmes multi-agents

Demande à un seul modèle de rassembler des informations, de les analyser, d'écrire un article et en plus de le critiquer. Tu obtiendras un texte où tout est un peu là et rien n'est bien fait, parce qu'un seul prompt tire le modèle dans quatre directions à la fois. Dans la savane non plus, personne ne fait tout tout seul.

Les systèmes multi-agents sont une architecture dans laquelle plusieurs agents IA spécialisés collaborent pour résoudre des tâches complexes. C'est comme un troupeau où chaque individu a son rôle - chasseurs, guetteurs, défenseurs. Chaque agent reçoit son propre prompt système et fait une seule chose, mais bien.

Les bases du multi-agent

Mettons-nous d'abord d'accord sur le vocabulaire. On stocke les rôles dans un Enum, un ensemble fermé de constantes nommées, et un message entre agents dans une @dataclass :

1from dataclasses import dataclass
2from enum import Enum
3from abc import ABC, abstractmethod
4from openai import OpenAI
5
6class AgentRole(Enum):
7    """Rôles des agents dans le système."""
8    RESEARCHER = "researcher"      # Rassemble des informations
9    ANALYZER = "analyzer"          # Analyse les données
10    WRITER = "writer"              # Rédige du contenu
11    CRITIC = "critic"              # Relit et améliore
12    COORDINATOR = "coordinator"    # Coordonne le travail
13
14@dataclass
15class AgentMessage:
16    """Message entre agents."""
17    sender: str
18    receiver: str
19    content: str
20    message_type: str = "task"

AgentMessage retient l'expéditeur, le destinataire, le contenu et le type du message. Le type par défaut est "task", donc pour une mission ordinaire, tu n'as pas besoin de le préciser.

Voici maintenant la classe de base de l'agent. Elle hérite de ABC, et la propriété system_prompt est marquée @abstractmethod, donc chaque sous-classe doit la définir :

1class BaseAgent(ABC):
2    """Classe de base des agents."""
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        """System prompt de l'agent."""
15        pass
16
17    def process(self, message: AgentMessage) -> str:
18        """Traite un message et renvoie une réponse."""
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

La méthode process enregistre le message dans la mémoire de l'agent et envoie deux messages au modèle : le prompt système avec le rôle et le contenu de la tâche. Impossible de créer un objet BaseAgent lui-même, Python lève une TypeError car l'implémentation de system_prompt manque.

Implémentation des agents spécialisés

Les agents spécialisés ne diffèrent que par leur prompt. La première paire, c'est le chercheur et l'analyste :

1class ResearcherAgent(BaseAgent):
2    """Agent de recherche - rassemble des informations."""
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    """Agent analytique - analyse les données."""
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."""

Chaque classe redéfinit seulement system_prompt, et toute la logique d'appel du modèle reste dans BaseAgent. Voilà l'héritage en pratique : le code commun une fois, les différences dans les sous-classes.

La deuxième paire, c'est le rédacteur et le critique :

1class WriterAgent(BaseAgent):
2    """Agent rédacteur - crée du contenu."""
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    """Agent critique - relit et améliore."""
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."""

Le Researcher rassemble les informations, l'Analyzer les analyse, le Writer crée le contenu, et le Critic relit et améliore. Ce sont des rôles typiques d'un système multi-agents. Remarque qu'un rôle de "Destroyer qui supprime des fichiers" n'apparaît jamais ici : un agent est là pour aider, pas pour détruire.

Orchestration des agents

Quelqu'un doit faire passer les messages entre les membres du troupeau. C'est le rôle de l'orchestrateur : il enregistre les agents et sert d'intermédiaire à chaque conversation :

1from typing import Optional
2
3class AgentOrchestrator:
4    """Coordinateur du travail des agents."""
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        """Enregistre un agent dans le système."""
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        """Envoie un message entre 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        # Enregistre la réponse
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 vérifie que le destinataire existe, enregistre le message dans l'historique, appelle process et enregistre aussi la réponse. Toute la conversation reste ainsi au même endroit.

La méthode run_pipeline dispose les agents en chaîne, où la sortie de l'un devient l'entrée du suivant :

1    def run_pipeline(self, task: str) -> dict[str, str]:
2        """Lance le pipeline de traitement."""
3        results = {}
4
5        # 1. Le chercheur rassemble les informations
6        research = self.send_message("user", "researcher",
7            f"Gather information on: {task}")
8        results["research"] = research
9
10        # 2. L'analyste analyse
11        analysis = self.send_message("researcher", "analyzer",
12            f"Analyze this information:\n{research}")
13        results["analysis"] = analysis
14
15        # 3. Le rédacteur crée le contenu
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. Le critique relit
21        review = self.send_message("writer", "critic",
22            f"Review this article:\n{draft}")
23        results["review"] = review
24
25        return results

L'ordre est fixe : le Researcher rassemble les informations, l'Analyzer analyse les données, le Writer crée le contenu, le Critic relit. Le dictionnaire results conserve le résultat de chaque étape, pas seulement le dernier.

Il ne reste plus qu'à enregistrer les quatre agents et à lancer le pipeline :

1# Utilisation
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")

Chaque agent est enregistré sous son nom, donc send_message("user", "researcher", ...) arrive exactement là où il faut.

Communication asynchrone

Les agents ne sont pas obligés de parler strictement à tour de rôle. Tu peux leur laisser des messages dans une asyncio.Queue, comme des notes près du point d'eau :

1import asyncio
2from typing import Callable
3
4class AsyncAgentSystem:
5    """Système d'agents asynchrone."""
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        """Traite les messages de la file."""
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()

La boucle process_messages attend un message avec await, le traite, et s'arrête quand elle reçoit le signal "STOP".

La méthode run_parallel_tasks place les tâches dans la file et lance le traitement :

1    async def run_parallel_tasks(self, tasks: list[tuple[str, str]]) -> dict:
2        """Lance les tâches en parallèle."""
3        # Ajoute les tâches à la file
4        for agent_name, task in tasks:
5            await self.message_queue.put(
6                AgentMessage("system", agent_name, task)
7            )
8
9        # Ajoute le signal de fin
10        await self.message_queue.put(
11            AgentMessage("system", "STOP", "STOP")
12        )
13
14        # Démarre le traitement
15        await self.process_messages()
16
17        return self.results
18
19# Exemple de traitement parallèle
20async def main():
21    system = AsyncAgentSystem()
22    # ... enregistre les 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())

Soyons honnêtes : malgré leur nom, ces tâches ne s'exécutent pas en parallèle. Il n'y a qu'un seul consommateur de la file, et agent.process est un appel bloquant ordinaire. Un vrai parallélisme viendrait du client asynchrone AsyncOpenAI avec asyncio.gather, ou de asyncio.to_thread. La file reste utile pour découpler les expéditeurs des destinataires.

Les outils de l'agent

Un agent qui sait seulement écrire, c'est comme un pisteur sans jumelles. Les outils lui permettent d'agir : fouiller une base, récupérer la météo, calculer un coût. Dans LangChain, on transforme une fonction en outil avec le décorateur @tool :

1from langchain_core.tools import tool
2
3@tool
4def search_db(query: str) -> list:
5    """Recherche dans la base d'observations d'animaux."""
6    return [f"Observation matching: {query}"]
7
8print(search_db.name)              # search_db
9print(search_db.invoke("lions"))   # ['Observation matching: lions']

La docstring n'est pas décorative ici : le modèle la lit comme description de l'outil, et sans elle le décorateur lève une erreur. Les types des paramètres indiquent au modèle quels arguments fournir. Un agent avec des outils travaille en boucle : il planifie une action, agit, observe le résultat et réfléchit pour corriger sa trajectoire.

Modèles de collaboration entre agents

Pour finir, trois modèles classiques. Chain est une chaîne où chaque agent transmet sa sortie au suivant :

1class CollaborationPatterns:
2    """Patterns de collaboration entre agents."""
3
4    @staticmethod
5    def chain(agents: list[BaseAgent], initial_input: str) -> str:
6        """Chaîne - chaque agent transmet sa sortie au suivant."""
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

C'est une généralisation de notre run_pipeline : la liste d'agents peut avoir n'importe quelle longueur.

Debate est une discussion entre deux agents sur plusieurs tours :

1    @staticmethod
2    def debate(agent1: BaseAgent, agent2: BaseAgent, topic: str, rounds: int = 3) -> list[str]:
3        """Débat - les agents discutent d'un sujet."""
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            # L'agent 2 répond
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

Chaque réponse d'un agent devient l'entrée de l'autre, et la liste conversation enregistre tout l'échange.

Consensus rassemble les avis de tous et demande à un agent d'en faire la synthèse :

1    @staticmethod
2    def consensus(agents: list[BaseAgent], question: str) -> str:
3        """Consensus - les agents parviennent à une conclusion commune."""
4        opinions = []
5
6        # Collecte les 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        # Synthèse (utilise l'un des 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) est le caractère de nouvelle ligne, utilisé à la place de "\n" dans la f-string. Je te conseille de commencer par une simple chaîne et de n'ajouter le débat et le consensus que là où une seule perspective ne suffit vraiment pas, car chaque tour signifie des appels payants supplémentaires au modèle.

Les systèmes multi-agents ouvrent la porte à des applications d'IA complexes. Dans la prochaine leçon, tu vas découvrir CrewAI - un framework pour créer des équipes d'agents qui te donne ces modèles tout prêts !

Retiens : un bon système d'agents fonctionne comme un troupeau où chacun connaît son rôle et où le guide garde le cap.