CrewAI - framework multi-agents

Dans la leçon précédente, on a écrit nous-mêmes la classe d'agent, l'orchestrateur et la chaîne de messages. Ça fonctionnait, mais chaque nouvelle fonctionnalité, comme les outils, la mémoire ou la délégation de tâches, devrait être ajoutée à la main. Quand l'expédition grandit, mieux vaut utiliser un règlement de campement tout prêt.

CrewAI est un framework Python pour créer des équipes d'agents IA qui collaborent pour résoudre des tâches complexes. C'est comme former l'équipage d'une expédition, où chacun a son rôle. Tout le framework repose sur trois notions : Agent (qui), Task (quoi) et Crew (comment ensemble).

Installation et bases

On installe deux paquets et on importe les classes de base. Le modèle de langage se configure avec la classe LLM de CrewAI lui-même :

1# pip install crewai crewai-tools
2
3from crewai import Agent, Task, Crew, Process, LLM
4
5# Modèle LLM
6llm = LLM(model="gpt-4o-mini", temperature=0.7)

Attention à un changement dans la bibliothèque : d'anciens exemples en ligne utilisent ici ChatOpenAI de LangChain, mais les versions actuelles de CrewAI ne dépendent plus de LangChain et refusent un tel objet lors de la validation de l'agent. LLM(model="gpt-4o-mini") utilise en coulisses la clé de la variable OPENAI_API_KEY.

Créer des agents

Un agent dans CrewAI, ce sont trois phrases sur lui-même : un rôle (role), un objectif (goal) et une histoire (backstory). Le modèle les reçoit comme la description du personnage qu'il doit incarner. Voici le chercheur :

1# Agent chercheur
2researcher = Agent(
3    role="Senior Research Analyst",
4    goal="Conduct thorough research and provide reliable information",
5    backstory="""You are an experienced analyst with 15 years of experience
6    in researching technology trends. You specialize in AI and ML analysis.
7    You always verify facts and look for reliable sources.""",
8    verbose=True,
9    allow_delegation=False,
10    llm=llm
11)

verbose=True affiche le raisonnement de l'agent dans la console, et allow_delegation=False lui interdit de confier son travail à d'autres. Dans les versions actuelles, False est de toute façon la valeur par défaut.

Le rédacteur et l'éditeur se ressemblent, seule la description du personnage change :

1# Agent rédacteur
2writer = Agent(
3    role="Content Writer",
4    goal="Create engaging and educational content about technology",
5    backstory="""You are a talented technical writer with a passion for
6    translating complex concepts into simple language. Your articles
7    are always well-organized and enjoyable to read.""",
8    verbose=True,
9    allow_delegation=True,
10    llm=llm
11)
12
13# Agent éditeur
14editor = Agent(
15    role="Senior Editor",
16    goal="Ensure the highest quality of published content",
17    backstory="""You are an experienced editor with an eye for detail.
18    You care about style consistency, factual accuracy, and clarity of message.
19    You are not afraid to suggest changes that will improve the text.""",
20    verbose=True,
21    allow_delegation=False,
22    llm=llm
23)

Le rédacteur a allow_delegation=True, donc quand il ne sait pas quelque chose, il peut demander à un membre de l'équipage. C'est la seule différence technique, le reste, c'est le caractère écrit dans backstory.

Définir des tâches

Une Task décrit le travail à faire. Chaque tâche a une description, un résultat attendu (expected_output) et un agent assigné :

1# Tâche de recherche
2research_task = Task(
3    description="""Conduct thorough research on the topic: {topic}
4
5    Your task:
6    1. Find the latest information and trends
7    2. Identify key players and technologies
8    3. Analyze potential applications
9    4. Assess future development directions
10
11    Deliver a detailed research report.""",
12    expected_output="Detailed research report with citations and sources",
13    agent=researcher
14)

La notation {topic} dans la description est un emplacement pour une variable. CrewAI y met la valeur du paramètre inputs au lancement de l'équipage.

La tâche de rédaction utilise le résultat de la recherche. C'est à ça que sert le paramètre context :

1# Tâche de rédaction
2writing_task = Task(
3    description="""Based on the research report, write a blog article.
4
5    Requirements:
6    1. Engaging introduction
7    2. Clear structure with headings
8    3. Practical examples
9    4. Summary with key takeaways
10    5. Length: 1000-1500 words
11
12    Use accessible language for technical readers.""",
13    expected_output="Complete blog article ready for publication",
14    agent=writer,
15    context=[research_task]  # Utilise le résultat de la tâche précédente
16)

context=[research_task] transmet au rédacteur le rapport du chercheur, donc on n'a pas besoin de coller les textes à la main comme le faisait notre orchestrateur.

L'éditeur reçoit l'article du rédacteur comme contexte :

1# Tâche d'édition
2editing_task = Task(
3    description="""Review and edit the article.
4
5    Check:
6    1. Grammar and style correctness
7    2. Argument consistency
8    3. Message clarity
9    4. Engaging tone
10
11    Make necessary corrections.""",
12    expected_output="Edited, final article",
13    agent=editor,
14    context=[writing_task]
15)

Description, résultat attendu, agent et contexte : ce sont les quatre éléments avec lesquels tu construis chaque tâche.

Créer et lancer le Crew

Un Crew relie agents et tâches. Le paramètre process décide de la façon dont l'équipage travaille :

1# Crée l'équipe
2content_crew = Crew(
3    agents=[researcher, writer, editor],
4    tasks=[research_task, writing_task, editing_task],
5    process=Process.sequential,  # ou Process.hierarchical
6    verbose=True
7)
8
9# Lance avec des inputs
10result = content_crew.kickoff(inputs={"topic": "RAG in enterprise AI"})
11print(result)

Process.sequential exécute les tâches l'une après l'autre, dans l'ordre de la liste. CrewAI prend en charge deux types de processus : sequential et hierarchical. kickoff renvoie un objet CrewOutput dont le texte se trouve dans le champ raw, et print l'affiche directement.

Outils pour les agents

Un agent sans outils ne connaît que ce que le modèle a appris. Le paquet crewai-tools fournit des outils tout prêts pour la recherche web, les sites et les fichiers :

1from crewai_tools import (
2    SerperDevTool,      # Recherche Google
3    WebsiteSearchTool,  # Recherche sur un site web
4    FileReadTool,       # Lecture de fichier
5    PDFSearchTool,      # Recherche dans un PDF
6)
7
8# Configure les outils
9search_tool = SerperDevTool()
10web_tool = WebsiteSearchTool()
11file_tool = FileReadTool()
12
13# Agent avec outils
14researcher_with_tools = Agent(
15    role="Research Analyst",
16    goal="Gather and analyze information from various sources",
17    backstory="Expert in information retrieval...",
18    tools=[search_tool, web_tool, file_tool],
19    verbose=True,
20    llm=llm
21)

SerperDevTool a besoin d'une clé dans la variable SERPER_API_KEY. L'agent décide lui-même quand utiliser quel outil, d'après leurs descriptions.

Un outil personnalisé est une classe qui hérite de BaseTool. Aujourd'hui, on l'importe depuis crewai.tools, et non depuis crewai_tools comme dans les anciens exemples :

1# Outil personnalisé
2from crewai.tools import BaseTool
3from pydantic import BaseModel, Field
4
5class DatabaseQueryInput(BaseModel):
6    query: str = Field(description="SQL query to execute")
7
8class DatabaseQueryTool(BaseTool):
9    name: str = "Database Query"
10    description: str = "Executes SQL queries on the database"
11    args_schema: type[BaseModel] = DatabaseQueryInput
12
13    def _run(self, query: str) -> str:
14        # Implémentation de la requête
15        return f"Query result: {query}"

args_schema est un modèle Pydantic qui décrit les arguments, et la méthode _run contient la logique proprement dite. Le modèle de langage lit name et description, alors écris-les clairement.

Processus hiérarchique

Dans un processus hiérarchique, c'est un manager qui répartit les tâches. On le crée comme un agent ordinaire :

1# Manager coordinateur
2manager = Agent(
3    role="Project Manager",
4    goal="Coordinate teamwork and ensure timely delivery",
5    backstory="Experienced PM with AI team management skills",
6    verbose=True,
7    llm=llm
8)
9
10# Crew avec hiérarchie
11hierarchical_crew = Crew(
12    agents=[researcher, writer, editor],
13    tasks=[research_task, writing_task, editing_task],
14    process=Process.hierarchical,
15    manager_agent=manager,  # ou manager_llm=llm
16    verbose=True
17)

Le manager ne va pas dans la liste agents, mais dans manager_agent. À la place de ton propre manager, tu peux passer manager_llm, et CrewAI en crée un lui-même. L'un des deux est obligatoire dans un processus hiérarchique.

Mémoire et contexte

L'équipage peut se souvenir de ce qui a été établi. On active la mémoire avec le drapeau memory=True, et embedder indique le modèle d'embeddings que CrewAI utilise pour enregistrer les souvenirs :

1# Crew avec mémoire
2crew_with_memory = Crew(
3    agents=[researcher, writer],
4    tasks=[research_task, writing_task],
5    process=Process.sequential,
6    memory=True,  # Active la mémoire
7    embedder={
8        "provider": "openai",
9        "config": {"model": "text-embedding-3-small"}
10    },
11    verbose=True
12)
13
14# Agent avec mémoire
15agent_with_memory = Agent(
16    role="Assistant",
17    goal="Help the user",
18    backstory="...",
19    memory=True,
20    llm=llm
21)

Ce sont les mêmes embeddings qu'il y a deux leçons, simplement utilisés pour stocker les souvenirs de l'équipage plutôt que des documents.

Callbacks et monitoring

Avec des tâches longues, tu veux savoir ce qui se passe. Crew accepte deux fonctions de rappel : task_callback après chaque tâche et step_callback après chaque étape d'un agent :

1from crewai.tasks.task_output import TaskOutput
2
3def on_task_end(output: TaskOutput) -> None:
4    """Appelée à la fin de chaque tâche."""
5    print(f"Completed: {output.description[:50]}")
6    print(f"   Output: {output.raw[:100]}...")
7
8def on_step(step) -> None:
9    """Appelée après chaque étape d'un agent."""
10    print(f"Agent step: {str(step)[:100]}")
11
12# Utilisation
13crew = Crew(
14    agents=[researcher, writer],
15    tasks=[research_task, writing_task],
16    task_callback=on_task_end,
17    step_callback=on_step,
18    verbose=True
19)

TaskOutput transporte la description de la tâche et le résultat brut dans le champ raw. Ne cherche pas de classe CrewCallback ni de module crewai.callbacks : CrewAI n'a ni l'un ni l'autre, et tu passes les fonctions de rappel justement par ces deux paramètres.

Exécution asynchrone

Tu peux lancer plusieurs équipages indépendants en même temps. C'est à ça que sert kickoff_async :

1import asyncio
2
3async def run_crew_async(topic: str) -> str:
4    """Exécution asynchrone de la crew."""
5    crew = Crew(
6        agents=[researcher, writer],
7        tasks=[research_task, writing_task],
8        process=Process.sequential
9    )
10
11    result = await crew.kickoff_async(inputs={"topic": topic})
12    return result
13
14# Lance plusieurs crews en parallèle
15async def run_multiple_crews():
16    topics = ["AI in medicine", "AI in finance", "AI in education"]
17
18    tasks = [run_crew_async(topic) for topic in topics]
19    results = await asyncio.gather(*tasks)
20
21    return dict(zip(topics, results))
22
23results = asyncio.run(run_multiple_crews())

asyncio.gather attend les trois équipages à la fois, et dict(zip(...)) associe les sujets aux résultats. En production, crée des objets agents et tâches séparés pour chaque équipage, car ici ils partagent tous les mêmes research_task et writing_task.

Exemple : content pipeline

Pour finir, on assemble tout dans un seul pipeline avec un spécialiste SEO en plus :

1def create_content_pipeline(topic: str) -> str:
2    """Pipeline complet de création de contenu."""
3
4    # Agents
5    seo_analyst = Agent(
6        role="SEO Specialist",
7        goal="Optimize content for search engines",
8        backstory="SEO expert with content marketing experience",
9        tools=[SerperDevTool()],
10        llm=llm
11    )
12
13    # Tâches
14    seo_task = Task(
15        description=f"Analyze keywords for the topic: {topic}",
16        expected_output="List of 10 keywords with their potential",
17        agent=seo_analyst
18    )
19
20    # Crew
21    pipeline = Crew(
22        agents=[seo_analyst, researcher, writer, editor],
23        tasks=[seo_task, research_task, writing_task, editing_task],
24        process=Process.sequential,
25        verbose=True
26    )
27
28    return pipeline.kickoff(inputs={"topic": topic})
29
30# Lance
31article = create_content_pipeline("Machine Learning in 2025")

La description de seo_task utilise une f-string, mais les tâches du chercheur et du rédacteur contiennent toujours {topic}, c'est pourquoi kickoff reçoit inputs. Sans ça, les agents verraient le texte littéral "{topic}". Je te conseille de commencer par le processus sequential et de n'activer la hiérarchie que pour un équipage plus grand, car le manager signifie des appels supplémentaires au modèle.

CrewAI est un framework puissant pour construire des équipes d'IA. Dans la prochaine leçon, tu vas découvrir les systèmes RAG en production - comment déployer ces technologies en entreprise !

Retiens : dans CrewAI, un agent est un membre d'équipage, une tâche est sa mission, et le crew est l'expédition tout entière.