CrewAI Processes : sequential, hierarchical et consensual

Introduction

Dans CrewAI, un Process définit la manière dont les agents coordonnent leur travail. Il détermine l'ordre d'exécution, la prise de décision, et la qualité du résultat final.

Trois processus disponibles :

Process	Mécanisme	Cas d'usage	Complexité
Sequential	Ordre linéaire	ETL, pipelines	Faible
Hierarchical	Agent manager	Tâches complexes	Moyenne
Consensual	Vote entre agents	Code review, contenu multi-perspective	Élevée

Ce guide couvre chaque processus avec du code Python reproductible, les pièges à éviter, et les critères de choix.

💡 Vous débutez avec CrewAI ? Consultez d'abord le guide d'installation CrewAI pour mettre en place votre premier projet.

---

Qu'est-ce qu'un Process dans CrewAI ?

Un Process contrôle trois dimensions d'une crew :

1. L'ordre d'exécution des tâches.
2. Qui décide de l'affectation.
3. Comment la crew converge vers un résultat.

from crewai import Crew, Process

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2, task3],
    process=Process.sequential
)

Le choix du process impacte directement le coût (appels LLM), la latence, et la qualité du output.

---

Section 1 — Sequential Process

Comment ça fonctionne

Le Sequential Process exécute les tâches dans l'ordre exact de la liste tasks. La tâche N+1 commence uniquement quand la tâche N est terminée.

from crewai import Crew, Agent, Task, Process

# Agents
scraper = Agent(
    role="Web Scraper",
    goal="Extraire le contenu pertinent d'une page web",
    backstory="Expert en extraction de données web"
)

transformer = Agent(
    role="Data Transformer",
    goal="Transformer les données brutes en format structuré",
    backstory="Spécialiste de la transformation de données"
)

loader = Agent(
    role="Data Loader",
    goal="Charger les données transformées en base",
    backstory="Expert en ingestion de données"
)

# Tâches ordonnées
extract_task = Task(
    description="Scraper la page https://example.com et extraire le contenu texte.",
    agent=scraper,
    expected_output="Texte brut extrait de la page."
)

transform_task = Task(
    description="Nettoyer et structurer le texte extrait en paragraphs.",
    agent=transformer,
    expected_output="Texte structuré en paragraphs."
)

load_task = Task(
    description="Préparer les données pour insertion en base MongoDB.",
    agent=loader,
    expected_output="Document JSON prêt pour insertion."
)

# Crew Sequential
crew = Crew(
    agents=[scraper, transformer, loader],
    tasks=[extract_task, transform_task, load_task],
    process=Process.sequential
)

result = crew.kickoff()

Quand l'utiliser

Le Sequential Process est optimal pour :

• Pipelines ETL — chaque étape dépend de la précédente (Extract → Transform → Load).
• Workflows de contenu — recherche → rédaction → révision.
• Tout cas où l'ordre est une contrainte métier non négociable.

Limites

• Une tâche échoue → le pipeline entier s'arrête.
• Pas de parallélisation : vous payez la latence cumulée.
• Un agent inactif pendant l'attente = appels LLM gaspillés.

Bonnes pratiques

# ❌ Erreur : oublier le paramètre process
crew = Crew(agents=[...], tasks=[...])

# ✅ Explicite
crew = Crew(agents=[...], tasks=[...], process=Process.sequential)

---

Section 2 — Hierarchical Process

Comment ça fonctionne

Le Hierarchical Process utilise un agent manager pour orchestrer le travail. Le manager ne fait pas le travail lui-même : il décide qui fait quoi, quand, et synthétise les résultats.

from crewai import Crew, Agent, Task, Process

# Agent manager — il orchestre
manager = Agent(
    role="Chef de Projet IA",
    goal="Orchestrer les agents pour un rapport complet et actionnable",
    backstory=(
        "Directeur de projet IA avec 10 ans d'expérience. "
        "Vous savez diviser les tâches complexes et coordonner des équipes."
    ),
    allow_delegation=True  # Critique : autorise la délégation
)

# Agents exécutants
analyst = Agent(
    role="Data Analyst",
    goal="Analyser les données financières",
    backstory="Analyste financier spécialisé en IA."
)

writer = Agent(
    role="Rédacteur Financier",
    goal="Rédiger un résumé executive clair",
    backstory="Rédacteur spécialisé en veille financière."
)

# Tâches
analysis_task = Task(
    description="Analyser les données Q1 2026 et identifier les 3 tendances majeures.",
    agent=analyst,
    expected_output="Liste des 3 tendances avec données de soutien."
)

writing_task = Task(
    description="Rédiger un executive summary de 200 mots basé sur l'analyse.",
    agent=writer,
    expected_output="Résumé executive en 200 mots."
)

# Crew Hierarchical
crew = Crew(
    agents=[manager, analyst, writer],
    tasks=[analysis_task, writing_task],
    process=Process.hierarchical,
    manager_agent=manager
)

result = crew.kickoff()

Comment le manager décide

Le manager reçoit les descriptions de toutes les tâches et les compétences de chaque agent. Il utilise le LLM pour décider :

1. Quelle tâche assigner à quel agent.
2. Dans quel ordre.
3. Quand aggregator les résultats.

Ce comportement est entièrement géré par le LLM sous-jacent — pas de règle programmatique.

Quand l'utiliser

• Tâches complexes multi-étapes nécessitant une coordination humaine-style.
• Projets créatifs où un lead manage des contributeurs.
• Affectation dynamique — le manager adapte selon le contexte.

Limites

• Plus cher : le manager fait des appels LLM supplémentaires.
• Moins prévisible : l'ordre dépend des décisions du LLM.
• Un prompt de manager mal写的 = orchestration sous-optimale.

Configuration recommandée

crew = Crew(
    agents=[manager, agent1, agent2, agent3],
    tasks=[...],
    process=Process.hierarchical,
    manager_agent=manager,
    manager_llm=ChatOpenAI(model="gpt-4o")  # LLM dédié au manager
)

---

Section 3 — Consensual Process

Comment ça fonctionne

Le Consensual Process fait voter les agents pour converger vers une réponse commune. Chaque agent exprime son avis, puis un mécanisme de vote détermine la réponse finale.

from crewai import Crew, Agent, Task, Process

# Agents avec perspectives complémentaires
optimist = Agent(
    role="Analyste Optimiste",
    goal="Évaluer le potentiel de croissance du projet",
    backstory="Analyste tech orienté croissance et opportunités."
)

skeptic = Agent(
    role="Analyste Critique",
    goal="Identifier les risques et faiblesses du projet",
    backstory="Consultant en gestion de risques."
)

mediator = Agent(
    role="Synthésiseur",
    goal="Converger les opinions vers une recommandation équilibrée",
    backstory="Expert en décision multi-critères."
)

# Tâches de vote
perspective_task = Task(
    description="Analyser ce projet IA sous l'angle de l'opportunité de marché.",
    agent=optimist,
    expected_output="Analyse des opportunités avec données quantitatives."
)

risk_task = Task(
    description="Analyser ce projet IA sous l'angle des risques opérationnels.",
    agent=skeptic,
    expected_output="Analyse des risques avec exemples concrets."
)

synthesis_task = Task(
    description="Sur base des analyses, voter pour la décision finale (go / no-go / conditional-go).",
    agent=mediator,
    expected_output="Décision motivée : go, no-go ou conditional-go."
)

# Crew Consensual
crew = Crew(
    agents=[optimist, skeptic, mediator],
    tasks=[perspective_task, risk_task, synthesis_task],
    process=Process.consensual
)

result = crew.kickoff()

Configuration avancée du vote

crew = Crew(
    agents=[optimist, skeptic, mediator],
    tasks=[perspective_task, risk_task, synthesis_task],
    process=Process.consensual,
    consensus_config={
        "threshold": 0.6,      # 60% d'accord minimum
        "voting_method": "weighted",  # pondéré par expertise
        "max_rounds": 3        # tours de vote max
    }
)

Quand l'utiliser

• Code review distribué — plusieurs expertises review le même code.
• Contenu multi-perspective — comparer plusieurs angles avant de converger.
• Décisions collégiales — quand aucune perspective ne doit dominer seule.
• Recherche exploratoire — générer plusieurs hypothèses, puis voter sur la plus prometteuse.

Limites

• Latence élevée — chaque agent doit s'exprimer, puis voter.
• Coût — N agents × leurs tâches + votes = nombreux appels LLM.
• Deadlock possible — si les opinions sont trop divergentes.

---

Section 4 — Combiner les Processes : Nested Crews

La vraie puissance de CrewAI émerge quand vous combinez les processus au sein d'une même application.

Architecture type

Main Crew (Hierarchical)
├── Sub-Crew ETL (Sequential)
│   ├── Scraper Agent
│   ├── Transformer Agent
│   └── Loader Agent
└── Sub-Crew Analysis (Consensual)
    ├── Analyst A
    ├── Analyst B
    └── Synthesizer

Exemple : pipeline complet

from crewai import Crew, Agent, Task, Process

# === MAIN CREW (Hierarchical) ===
manager = Agent(
    role="Chef de Projet",
    goal="Orchestrer le pipeline de bout en bout",
    backstory="Chef de projet IA senior.",
    allow_delegation=True
)

# === ETL Sub-Crew (Sequential) ===
scraper = Agent(role="Scraper", goal="Extraire les données", backstory="...")
transformer = Agent(role="Transformer", goal="Nettoyer les données", backstory="...")
loader = Agent(role="Loader", goal="Stocker les données", backstory="...")

extract = Task(description="Scraper les données", agent=scraper)
transform = Task(description="Transformer les données", agent=transformer)
load = Task(description="Charger les données", agent=loader)

etl_crew = Crew(
    agents=[scraper, transformer, loader],
    tasks=[extract, transform, load],
    process=Process.sequential,
    crew_name="etl_crew"
)

# === Analysis Sub-Crew (Consensual) ===
analyst_a = Agent(role="Analyste A", goal="Analyse technique", backstory="...")
analyst_b = Agent(role="Analyste B", goal="Analyse marché", backstory="...")

analysis_a = Task(description="Analyse technique", agent=analyst_a)
analysis_b = Task(description="Analyse marché", agent=analyst_b)

analysis_crew = Crew(
    agents=[analyst_a, analyst_b],
    tasks=[analysis_a, analysis_b],
    process=Process.consensual,
    crew_name="analysis_crew"
)

# === MAIN CREW ===
etl_task = Task(
    description="Exécuter le pipeline ETL complet.",
    crew=etl_crew,
    expected_output="Données nettoyées et stockées."
)

analysis_task = Task(
    description="Analyser les données avec les deux analysts.",
    crew=analysis_crew,
    expected_output="Rapport d'analyse consensus."
)

main_crew = Crew(
    agents=[manager],
    tasks=[etl_task, analysis_task],
    process=Process.hierarchical
)

result = main_crew.kickoff()

Tableau récapitulatif des combinaisons

Architecture	Use case
Hierarchical + Sequential	Pipeline ETL supervisé par un manager
Hierarchical + Consensual	Décision complexe avec vote
Sequential + Consensual	Pipeline classique + revue par les pairs

---

Section 5 — Process et LLM : impact sur les coûts et la latence

Comparaison des processus

Process	Appels LLM	Latence	Coût
Sequential	N (1 par tâche)	Élevée (cumulée)	++
Hierarchical	N + M (manager)	Moyenne	+++
Consensual	N + V (votes)	Très élevée	++++

Comment réduire les coûts

1. Sequential — combinez plusieurs sous-tâches en une seule任务 plus large.
2. Hierarchical — utilisez un modèle moins cher pour le manager.
3. Consensual — limitez les tours de vote avec max_rounds.

# ❌ 5 tasks = 5 appels LLM minimum
tasks = [task1, task2, task3, task4, task5]

# ✅ 2 tasks = 2 appels LLM (combinaison intelligente)
tasks = [data_collection_task, data_analysis_task]

---

Section 6 — Debugging un Process

Activer le mode verbose

crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    verbose=True  # Affiche chaque étape
)

result = crew.kickoff()

Output détaillé pour audit

crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.hierarchical,
    verbose=2  # Niveau 2 = output complet
)

result = crew.kickoff()

# Accéder aux traces
print(result.token_usage)    # Stats d'usage LLM
print(result.tasks_output)    # Output de chaque tâche

Analyser les logs

for i, task_result in enumerate(result.tasks_output):
    print(f"Tâche {i+1}: {task_result.agent.role}")
    print(f"  Output: {task_result.raw}")
    print(f"  Token usage: {task_result.token_usage}")

Erreurs fréquentes

Erreur	Cause	Solution
Task bloquante	Agent en attente infinie	Ajouter un timeout sur le task
Manager non-déléguant	allow_delegation=False	Mettre allow_delegation=True
Vote en deadlock	Trop d'agents avec avis divergents	Réduire le nombre d'agents ou augmenter max_rounds
Coût explosif	Trop de tâches séquentielles	Combiner les tâches

---

Questions fréquentes

Quels sont les 3 types de processus dans CrewAI ?

Sequential (exécution ordonnée), Hierarchical (un agent manager orchestre), et Consensual (les agents votent pour converger).

Comment utiliser un Sequential Process ?

Définissez process=Process.Sequential dans la création de votre Crew. Les tâches s'exécutent dans l'ordre de la liste tasks.

Qu'est-ce que le Hierarchical Process ?

Un agent manager orchestre les tâches et les distribue aux agents subordonnés. Définissez un manager_agent avec allow_delegation=True et utilisez process=Process.hierarchical.

Comment fonctionne le Consensual Process ?

Les agents votent pour converger vers une réponse commune. Configurez le seuil de consensus avec consensus_config.

Peut-on combiner plusieurs processus ?

Oui, via les nested crews : une crew principale peut contenir des sub-crews utilisant des processus différents. Par exemple, une crew Hierarchical contenant une sub-crew Sequential pour un pipeline ETL.

---

Aller plus loin avec CrewAI

• Guide complet CrewAI — panorama complet de l'écosystème
• CrewAI Tools intégrés — les outils natifs disponibles
• Architectures Multi-Agents — patterns avancés d'agents multiples