Systèmes multi-agents IA

1 agent généraliste
└─ Recherche + Analyse + Rédaction
⏱️ Durée : 5 minutes
⭐ Qualité : 6/10

3 agents spécialisés
├─ Researcher : Collecte des données
├─ Analyst : Analyse et insights
└─ Writer : Rédaction du rapport

⏱️ Durée : 3 minutes (parallèle)
⭐ Qualité : 9/10

from typing import List, Dict
import openai

class Agent:
    """Agent de base"""

    def __init__(self, name: str, role: str, tools: List = None):
        self.name = name
        self.role = role
        self.tools = tools or []

    def run(self, task: str) -> str:
        """Exécute une tâche"""
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"Tu es un {self.role}."},
                {"role": "user", "content": task}
            ]
        )
        return response.choices[0].message.content


class SupervisorAgent:
    """Agent superviseur qui coordonne les workers"""

    def __init__(self, workers: List[Agent]):
        self.workers = {w.name: w for w in workers}

    def run(self, task: str) -> str:
        """Décompose la tâche et délègue aux workers"""

        # Étape 1 : Planifier
        plan = self._create_plan(task)
        print(f"📋 Plan créé : {len(plan)} étapes")

        results = []

        # Étape 2 : Exécuter chaque sous-tâche
        for step in plan:
            agent_name = step["agent"]
            subtask = step["task"]

            print(f"\n🤖 {agent_name} : {subtask}")

            if agent_name in self.workers:
                result = self.workers[agent_name].run(subtask)
                results.append({
                    "agent": agent_name,
                    "task": subtask,
                    "result": result
                })
                print(f"✅ Résultat : {result[:100]}...")
            else:
                results.append({
                    "agent": agent_name,
                    "task": subtask,
                    "result": f"Erreur : agent {agent_name} introuvable"
                })

        # Étape 3 : Synthétiser
        final_report = self._synthesize(task, results)
        return final_report

    def _create_plan(self, task: str) -> List[Dict]:
        """Crée un plan de tâches"""
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{
                "role": "user",
                "content": f"""Tu es un superviseur. Décompose cette tâche pour les agents disponibles :

Agents disponibles : {', '.join(self.workers.keys())}

Tâche : {task}

Réponds en JSON : [{{"agent": "nom", "task": "description"}}, ...]"""
            }],
            response_format={"type": "json_object"}
        )

        import json
        plan = json.loads(response.choices[0].message.content)
        return plan.get("steps", plan.get("plan", []))

    def _synthesize(self, original_task: str, results: List[Dict]) -> str:
        """Synthétise les résultats"""
        results_text = "\n\n".join([
            f"**{r['agent']}** ({r['task']}):\n{r['result']}"
            for r in results
        ])

        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{
                "role": "user",
                "content": f"""Synthétise ces résultats en un rapport final cohérent.

Tâche originale : {original_task}

Résultats des agents :
{results_text}

Rapport final :"""
            }]
        )

        return response.choices[0].message.content


# ========== EXEMPLE D'UTILISATION ==========

# Créer les agents workers
researcher = Agent(
    name="Researcher",
    role="chercheur expert en collecte de données"
)

analyst = Agent(
    name="Analyst",
    role="analyste de données qui extrait des insights"
)

writer = Agent(
    name="Writer",
    role="rédacteur qui crée des rapports clairs et structurés"
)

# Créer le superviseur
supervisor = SupervisorAgent(workers=[researcher, analyst, writer])

# Exécuter une tâche complexe
result = supervisor.run("""
Analyse le marché des voitures électriques en 2024 :
- Principaux acteurs
- Parts de marché
- Tendances et prévisions
Crée un rapport de 2 pages.
""")

print(f"\n\n📄 RAPPORT FINAL :\n{result}")

📋 Plan créé : 3 étapes

🤖 Researcher : Recherche les principaux acteurs et parts de marché des voitures électriques en 2024
✅ Résultat : Tesla (19%), BYD (17%), Volkswagen (12%)...

🤖 Analyst : Analyse les tendances et prévisions du marché des voitures électriques
✅ Résultat : Croissance de 35% en 2024, baisse des prix des batteries...

🤖 Writer : Rédige un rapport de 2 pages basé sur les données et analyses
✅ Résultat : [Rapport structuré]...

📄 RAPPORT FINAL :
[Rapport final synthétisé par le superviseur]

class CollaborativeAgent(Agent):
    """Agent capable de collaborer avec d'autres agents"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.peers = []

    def add_peer(self, peer: 'CollaborativeAgent'):
        """Ajoute un agent pair"""
        self.peers.append(peer)

    def ask_peer(self, peer_name: str, question: str) -> str:
        """Pose une question à un pair"""
        peer = next((p for p in self.peers if p.name == peer_name), None)
        if peer:
            return peer.answer_peer(question, self.name)
        return f"Pair {peer_name} introuvable"

    def answer_peer(self, question: str, from_agent: str) -> str:
        """Répond à une question d'un pair"""
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"Tu es {self.name}, {self.role}. Un collègue te pose une question."},
                {"role": "user", "content": f"{from_agent} demande : {question}"}
            ]
        )
        return response.choices[0].message.content

    def run_collaborative(self, task: str, max_iterations: int = 5) -> str:
        """Exécute une tâche en collaborant avec les pairs"""

        context = f"Tâche : {task}\n\n"

        for iteration in range(max_iterations):
            print(f"\n🔄 {self.name} - Itération {iteration + 1}")

            # Penser à la prochaine action
            response = openai.chat.completions.create(
                model="gpt-4",
                messages=[{
                    "role": "user",
                    "content": f"""{context}

En tant que {self.name} ({self.role}), que fais-tu maintenant ?

Options :
1. ask_peer(nom, question) : Demander de l'aide à un collègue
2. finish(réponse) : Terminer avec une réponse

Réponds en JSON : {{"action": "ask_peer" ou "finish", "params": ...}}"""
                }],
                response_format={"type": "json_object"}
            )

            import json
            decision = json.loads(response.choices[0].message.content)

            action = decision.get("action")

            if action == "finish":
                print(f"✅ {self.name} termine")
                return decision.get("params", {}).get("answer", "Terminé")

            elif action == "ask_peer":
                peer_name = decision["params"]["peer"]
                question = decision["params"]["question"]

                print(f"❓ {self.name} demande à {peer_name} : {question}")

                answer = self.ask_peer(peer_name, question)
                print(f"💬 {peer_name} répond : {answer[:100]}...")

                context += f"\n{self.name} → {peer_name} : {question}\n{peer_name} → {self.name} : {answer}\n"

        return "Limite d'itérations atteinte"


# ========== EXEMPLE ==========

# Créer les agents
coder = CollaborativeAgent(name="Coder", role="développeur Python expert")
reviewer = CollaborativeAgent(name="Reviewer", role="code reviewer qui vérifie la qualité")
tester = CollaborativeAgent(name="Tester", role="testeur QA qui écrit les tests")

# Les connecter (pair-à-pair)
coder.add_peer(reviewer)
coder.add_peer(tester)
reviewer.add_peer(coder)
reviewer.add_peer(tester)
tester.add_peer(coder)
tester.add_peer(reviewer)

# Exécuter une tâche collaborative
result = coder.run_collaborative("Crée une fonction qui valide une adresse email")

class SequentialMultiAgent:
    """Système multi-agents séquentiel"""

    def __init__(self, agents: List[Agent]):
        self.agents = agents

    def run(self, initial_input: str) -> str:
        """Exécute la chaîne d'agents"""

        current_output = initial_input

        for agent in self.agents:
            print(f"\n🔧 {agent.name} en action...")

            # Chaque agent enrichit la sortie du précédent
            current_output = agent.run(current_output)

            print(f"✅ Résultat : {current_output[:150]}...")

        return current_output


# ========== EXEMPLE : PIPELINE DE CONTENU ==========

# Agents de la chaîne
outline_creator = Agent(
    name="OutlineCreator",
    role="créateur de plan qui structure les idées"
)

content_writer = Agent(
    name="ContentWriter",
    role="rédacteur qui développe le contenu à partir du plan"
)

editor = Agent(
    name="Editor",
    role="éditeur qui améliore la clarté et la grammaire"
)

seo_optimizer = Agent(
    name="SEOOptimizer",
    role="expert SEO qui optimise pour les moteurs de recherche"
)

# Créer le pipeline
pipeline = SequentialMultiAgent([
    outline_creator,
    content_writer,
    editor,
    seo_optimizer
])

# Exécuter
final_article = pipeline.run("Sujet : Les avantages de l'IA dans l'éducation")

pip install crewai crewai-tools

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

# ========== 1. DÉFINIR LES OUTILS ==========

search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()

# ========== 2. CRÉER LES AGENTS ==========

researcher = Agent(
    role="Market Research Analyst",
    goal="Collecter des données précises sur le marché",
    backstory="""Tu es un chercheur expérimenté avec 10 ans d'expérience
    en analyse de marché. Tu es méticuleux et sais trouver les bonnes sources.""",
    tools=[search_tool, scrape_tool],
    verbose=True,
    allow_delegation=False
)

analyst = Agent(
    role="Data Analyst",
    goal="Analyser les données et extraire des insights",
    backstory="""Tu es un analyste de données expert en statistiques
    et visualisation. Tu transformes les données brutes en insights actionnables.""",
    verbose=True,
    allow_delegation=False
)

writer = Agent(
    role="Content Writer",
    goal="Créer un rapport clair et professionnel",
    backstory="""Tu es un rédacteur senior spécialisé en rapports d'analyse.
    Tu sais vulgariser des concepts complexes.""",
    verbose=True,
    allow_delegation=False
)

# ========== 3. DÉFINIR LES TÂCHES ==========

research_task = Task(
    description="""Recherche sur le marché des voitures électriques en 2024.
    Collecte :
    - Les 5 principaux acteurs
    - Leurs parts de marché
    - Les dernières innovations

    Fournis un rapport structuré avec sources.""",
    agent=researcher,
    expected_output="Rapport de recherche avec données et sources"
)

analysis_task = Task(
    description="""Analyse les données de recherche.
    Identifie :
    - Les tendances principales
    - Les opportunités
    - Les risques

    Fournis une analyse SWOT.""",
    agent=analyst,
    expected_output="Analyse SWOT détaillée",
    context=[research_task]  # Dépend de research_task
)

writing_task = Task(
    description="""Rédige un rapport exécutif de 2 pages.
    Structure :
    - Résumé exécutif
    - Principaux acteurs
    - Tendances et insights
    - Recommandations

    Ton professionnel et concis.""",
    agent=writer,
    expected_output="Rapport exécutif de 2 pages",
    context=[research_task, analysis_task]  # Dépend des deux
)

# ========== 4. CRÉER L'ÉQUIPE (CREW) ==========

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,  # Ou Process.hierarchical
    verbose=2
)

# ========== 5. EXÉCUTER ==========

result = crew.kickoff()

print(f"\n\n{'='*60}")
print("📄 RAPPORT FINAL")
print(f"{'='*60}\n")
print(result)

[Researcher] Starting task: Recherche sur le marché...
[Researcher] Using tool: search_tool
[Researcher] Task completed ✅

[Analyst] Starting task: Analyse les données...
[Analyst] Analyzing data from Researcher...
[Analyst] Task completed ✅

[Writer] Starting task: Rédige un rapport...
[Writer] Synthesizing from Researcher and Analyst...
[Writer] Task completed ✅

============================================================
📄 RAPPORT FINAL
============================================================

# Marché des Voitures Électriques 2024

## Résumé Exécutif {#résumé-exécutif}
Le marché des VE connaît une croissance de 35% en 2024...

[Rapport complet]

manager = Agent(
    role="Project Manager",
    goal="Coordonner l'équipe",
    allow_delegation=True  # Peut déléguer aux autres agents
)

crew = Crew(
    agents=[manager, researcher, analyst],
    tasks=[main_task],
    process=Process.hierarchical,
    manager_llm="gpt-4"  # LLM pour le manager
)

def task_callback(output):
    print(f"✅ Tâche terminée : {output.description}")

task = Task(
    description="...",
    agent=researcher,
    callback=task_callback
)

pip install langgraph

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# ========== 1. DÉFINIR L'ÉTAT ==========

class AgentState(TypedDict):
    """État partagé entre agents"""
    messages: Annotated[list, operator.add]
    next_agent: str
    final_answer: str

# ========== 2. CRÉER LES AGENTS ==========

def researcher_agent(state: AgentState) -> AgentState:
    """Agent de recherche"""
    print("🔍 Researcher en action...")

    # Simuler une recherche
    result = "Données collectées : Tesla 19%, BYD 17%..."

    return {
        "messages": [{"role": "researcher", "content": result}],
        "next_agent": "analyst"
    }

def analyst_agent(state: AgentState) -> AgentState:
    """Agent d'analyse"""
    print("📊 Analyst en action...")

    # Analyser les données du researcher
    previous_data = state["messages"][-1]["content"]
    analysis = f"Analyse : {previous_data} → Tendance haussière"

    return {
        "messages": [{"role": "analyst", "content": analysis}],
        "next_agent": "decision"
    }

def decision_node(state: AgentState) -> str:
    """Décide du prochain agent"""
    # Logique de décision
    if len(state["messages"]) < 5:
        return "researcher"  # Continuer la recherche
    else:
        return "writer"  # Passer à la rédaction

def writer_agent(state: AgentState) -> AgentState:
    """Agent rédacteur"""
    print("✍️ Writer en action...")

    # Synthétiser tous les messages
    all_data = "\n".join([m["content"] for m in state["messages"]])
    report = f"Rapport final basé sur :\n{all_data}"

    return {
        "messages": [{"role": "writer", "content": report}],
        "final_answer": report,
        "next_agent": END
    }

# ========== 3. CONSTRUIRE LE GRAPHE ==========

workflow = StateGraph(AgentState)

# Ajouter les nœuds (agents)
workflow.add_node("researcher", researcher_agent)
workflow.add_node("analyst", analyst_agent)
workflow.add_node("writer", writer_agent)

# Définir les transitions
workflow.add_edge("researcher", "analyst")
workflow.add_conditional_edges(
    "analyst",
    decision_node,  # Fonction qui décide du prochain nœud
    {
        "researcher": "researcher",
        "writer": "writer"
    }
)

# Point d'entrée
workflow.set_entry_point("researcher")

# Compiler
app = workflow.compile()

# ========== 4. EXÉCUTER ==========

initial_state = {
    "messages": [],
    "next_agent": "",
    "final_answer": ""
}

result = app.invoke(initial_state)

print(f"\n\n📄 Résultat final :\n{result['final_answer']}")

class MessageBus:
    """Bus de messages pour communication inter-agents"""

    def __init__(self):
        self.messages = []
        self.subscribers = {}

    def subscribe(self, agent_name: str, topics: List[str]):
        """Un agent s'abonne à des topics"""
        for topic in topics:
            if topic not in self.subscribers:
                self.subscribers[topic] = []
            self.subscribers[topic].append(agent_name)

    def publish(self, from_agent: str, topic: str, message: str):
        """Publier un message sur un topic"""
        self.messages.append({
            "from": from_agent,
            "topic": topic,
            "message": message,
            "timestamp": datetime.now()
        })

        # Notifier les abonnés
        if topic in self.subscribers:
            print(f"📨 {from_agent} publie sur '{topic}' → {len(self.subscribers[topic])} abonnés")

    def get_messages(self, for_agent: str, topic: str = None) -> List[dict]:
        """Récupérer les messages pour un agent"""
        relevant_topics = [t for t, subs in self.subscribers.items() if for_agent in subs]

        return [
            msg for msg in self.messages
            if (topic is None or msg["topic"] == topic)
            and msg["topic"] in relevant_topics
        ]

class SharedMemory:
    """Mémoire partagée entre agents"""

    def __init__(self):
        self.data = {}
        self.locks = {}

    def write(self, key: str, value: Any, agent_name: str):
        """Écrire dans la mémoire partagée"""
        self.data[key] = {
            "value": value,
            "written_by": agent_name,
            "timestamp": datetime.now()
        }
        print(f"💾 {agent_name} écrit '{key}'")

    def read(self, key: str, agent_name: str) -> Any:
        """Lire depuis la mémoire partagée"""
        if key in self.data:
            print(f"📖 {agent_name} lit '{key}'")
            return self.data[key]["value"]
        return None

    def list_keys(self) -> List[str]:
        """Liste toutes les clés disponibles"""
        return list(self.data.keys())

def critique_refinement_loop(
    generator_agent: Agent,
    critic_agent: Agent,
    task: str,
    max_rounds: int = 3
) -> str:
    """Boucle de critique et amélioration"""

    current_output = generator_agent.run(task)
    print(f"📝 Génération initiale : {current_output[:100]}...")

    for round in range(max_rounds):
        print(f"\n🔄 Round {round + 1} de critique")

        # Critique
        critique = critic_agent.run(f"Critique ce travail :\n\n{current_output}")
        print(f"💬 Critique : {critique[:100]}...")

        # Vérifier si satisfaisant
        if "satisfaisant" in critique.lower() or "excellent" in critique.lower():
            print("✅ Travail approuvé !")
            break

        # Améliorer
        improved = generator_agent.run(
            f"Améliore ce travail basé sur cette critique :\n\nTravail :\n{current_output}\n\nCritique :\n{critique}"
        )

        current_output = improved
        print(f"✨ Amélioration : {improved[:100]}...")

    return current_output

# Équipe de support client
support_team = Crew(
    agents=[
        Agent(role="Triage Agent", goal="Classifier les tickets"),
        Agent(role="Technical Support", goal="Résoudre problèmes techniques"),
        Agent(role="Billing Support", goal="Gérer questions de facturation"),
        Agent(role="Escalation Manager", goal="Escalader cas complexes")
    ],
    process=Process.hierarchical
)

content_crew = Crew(
    agents=[
        Agent(role="SEO Researcher", goal="Trouver mots-clés et tendances"),
        Agent(role="Content Writer", goal="Rédiger article optimisé SEO"),
        Agent(role="Fact Checker", goal="Vérifier exactitude"),
        Agent(role="Editor", goal="Améliorer clarté et style"),
        Agent(role="Social Media Manager", goal="Créer posts réseaux sociaux")
    ],
    process=Process.sequential
)

code_team = Crew(
    agents=[
        Agent(role="Code Generator", goal="Écrire code fonctionnel"),
        Agent(role="Security Reviewer", goal="Vérifier vulnérabilités"),
        Agent(role="Performance Reviewer", goal="Optimiser performance"),
        Agent(role="Test Writer", goal="Créer tests unitaires"),
        Agent(role="Documentation Writer", goal="Documenter le code")
    ],
    process=Process.sequential
)