Tutoriel agent swarm : coordonner des agents IA avec CrewAI

Qu’est-ce qu’un essaim d’agents ?

Un essaim d’agents est un groupe d’agents IA qui collaborent pour accomplir une tâche. Au lieu de demander à un seul modèle de rechercher, rédiger, relire et finaliser, un essaim répartit le travail entre plusieurs agents spécialisés. Chaque agent a un rôle clair, et le système les coordonne pour produire un meilleur résultat final.

L’avantage, c’est la spécialisation. Un agent dédié à la recherche se concentre sur la découverte et la vérification des sources, un agent focalisé sur la rédaction s’attache à la structure et à la clarté, et un relecteur contrôle le brouillon par rapport à la recherche initiale sans être distrait par sa propre écriture. 

Un agent coordinateur relie les étapes et décide de la suite. Dans l’essaim que nous construisons, ces rôles correspondent directement au Researcher, au Writer, au Reviewer et au Manager.

Mais ce n’est pas gratuit. Plus d’agents signifie plus d’appels au modèle, plus de temps et plus de points de défaillance potentiels ; un arbitrage sur lequel nous reviendrons à la fin, une fois que vous aurez vu le coût réel d’exécution.

Qu’est-ce que CrewAI ?

CrewAI est un framework open source pour construire des systèmes multi-agents. Il offre une façon claire de définir des agents, de leur assigner des tâches et de choisir leur mode de coordination, pour vous concentrer sur les rôles et le workflow plutôt que sur la plomberie qui les relie.

Trois concepts font l’essentiel du travail. 

• Agents sont les travailleurs individuels, chacun défini par un rôle, un objectif et un contexte qui façonnent son comportement. 
• Tasks décrivent le travail à effectuer et la production attendue pour chacune. 
• Une crew réunit agents et tâches et les exécute selon un processus choisi

Ce processus peut être séquentiel, avec des tâches lancées dans un ordre fixe, ou hiérarchique, où un agent manager délègue le travail et valide les sorties. Nous utiliserons ici le processus hiérarchique, le Manager étant en charge de la délégation.

CrewAI gère également l’accès au modèle via LiteLLM, une bibliothèque open source qui offre une interface unifiée pour de nombreux fournisseurs. C’est ce qui nous permet de définir un unique modèle Gemini 3.5 Flash et de le réutiliser sur les quatre agents, que nous allons configurer ci-dessous.

Créer un essaim d’agents avec CrewAI et Gemini 3.5 Flash

Passons à la mise en place.

1. Prérequis

Avant de commencer, assurez-vous d’avoir :

• Python 3.11 ou version ultérieure installé
• Jupyter Notebook ou JupyterLab installé
• Des bases en Python et en notebooks
• Un compte Google AI Studio
• La facturation activée, ou des crédits ajoutés à votre compte Google pour l’usage de Gemini
• Un compte gratuit Olostep

2. Installer les dépendances

Lancez Jupyter Notebook ou JupyterLab et ouvrez un notebook avec le dernier noyau Python disponible dans votre environnement. Exécutez ensuite la cellule suivante pour installer les paquets requis.

%pip install --quiet -U \
    crewai \
    crewai-tools \
    google-genai \
    litellm \
    olostep \
    nest-asyncio

Une fois l’installation terminée, exécutez la cellule suivante pour vérifier les versions installées. Cela permet de confirmer que tout est correctement installé.

import importlib.metadata as md

for pkg in [
    "crewai",
    "crewai-tools",
    "google-genai",
    "litellm",
    "olostep",
    "nest-asyncio",
]:
    print(f"{pkg:>16}  {md.version(pkg)}")

Vous devriez voir une sortie similaire :

         crewai  1.14.6
    crewai-tools  1.14.6
      google-genai  2.8.0
          litellm  1.87.1
          olostep  1.1.0
    nest-asyncio  1.6.0

3. Définir les variables d’environnement

Vous aurez besoin de deux clés API pour ce projet. 

Commencez par créer une clé API Gemini depuis Google AI Studio. Ouvrez Google AI Studio, allez dans la section API key et créez une nouvelle clé pour votre projet. Gemini propose un niveau gratuit, avec un usage et un débit limités. Pour tester et exécuter Gemini plus fiablement, connectez votre compte de facturation Google ou ajoutez des crédits à votre compte Google afin que l’utilisation soit facturée lorsque vous dépassez les limites du niveau gratuit.

Ensuite, créez une clé API Olostep. Inscrivez-vous sur Olostep, ouvrez le tableau de bord Olostep et générez une clé API depuis la page des clés API. Nous utiliserons cette clé pour donner au Researcher un accès web en direct pour la recherche et l’extraction de pages.

Une fois les deux clés prêtes, enregistrez-les comme variables d’environnement. Gemini alimentera les agents CrewAI, tandis qu’Olostep permettra au Researcher de collecter des informations à jour sur le web.

Après avoir créé les deux clés, enregistrez-les comme variables d’environnement avant d’exécuter le notebook.

PowerShell :

$env:GEMINI_API_KEY="your-gemini-key"
$env:OLOSTEP_API_KEY="your-olostep-key"

macOS/Linux :

export GEMINI_API_KEY="your-gemini-key"
export OLOSTEP_API_KEY="your-olostep-key"

Exécutez maintenant la cellule suivante pour vérifier que la clé API Gemini est correctement chargée.

import os

api_key = os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")

if not api_key:
    raise RuntimeError(
        "Set GEMINI_API_KEY or GOOGLE_API_KEY in your environment, "
        "then restart the notebook kernel."
    )
print(f"Gemini API key loaded")

Vous devriez voir :

Gemini API key loaded

4. Configurer Gemini 3.5 Flash via LiteLLM

Dans ce projet, nous utiliserons Gemini 3.5 Flash comme LLM partagé pour tous les agents CrewAI. Gemini 3.5 Flash est conçu pour des charges rapides, agentiques et orientées code. Il est utile ici, car nos agents doivent raisonner en plusieurs étapes, suivre des instructions, utiliser des outils et produire des sorties structurées.

Gemini 3.5 Flash prend également en charge une grande fenêtre de contexte, des sorties longues, le raisonnement et l’usage d’outils, ce qui en fait un excellent choix pour des workflows d’agents. Comme un essaim peut appeler le modèle plusieurs fois entre Researcher, Writer, Reviewer et Manager, utiliser un modèle Flash aide à conserver une bonne réactivité tout en offrant un raisonnement et une qualité de rédaction solides.

Si vous souhaitez comparer ce modèle à d’autres LLM de pointe, nous vous recommandons de lire nos guides Gemini 3.5 Flash vs GPT-5.5 et Gemini 3.5 Flash vs Claude Opus 4.8.

Nous accéderons à Gemini via LiteLLM. LiteLLM est une bibliothèque open source qui fournit une interface unifiée pour appeler de nombreux fournisseurs de LLM, dont Gemini, OpenAI, Anthropic, Bedrock, Vertex AI, et d’autres. Dans cette configuration, CrewAI utilise LiteLLM en arrière-plan afin que nous puissions passer le nom du modèle Gemini au format simple provider/model.

Nous allons maintenant créer un LLM CrewAI partagé et le réutiliser pour tous les agents. Cela simplifie la configuration, puisque le Researcher, le Writer, le Reviewer et le Manager utiliseront tous le même modèle Gemini.

from crewai import LLM

GEMINI_MODEL = "gemini/gemini-3.5-flash"

gemini_llm = LLM(
    model=GEMINI_MODEL,
    api_key=os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY"),
    temperature=1.0,
)

print(f"LLM ready: {gemini_llm.model}")
print("Provider API: Google Gemini (via LiteLLM)")

Vous devriez voir une sortie similaire :

LLM ready: gemini-3.5-flash
Provider API: Google Gemini (via LiteLLM)

Cela confirme que CrewAI peut désormais accéder à Gemini via LiteLLM. Dans les étapes suivantes, nous passerons cet objet partagé gemini_llm à chaque agent.

5. Ajouter l’accès web en direct

L’agent Researcher a besoin d’accéder à des informations récentes. Pour cela, nous allons créer un outil de recherche Olostep personnalisé et l’attribuer ensuite au Researcher.

Cet outil peut fonctionner en deux modes. Il peut renvoyer uniquement des résumés de recherche, plus rapides, ou scraper les pages de résultats et renvoyer le contenu en markdown. Nous ajoutons aussi un petit budget de recherche afin d’éviter que l’agent n’appelle trop souvent l’outil web.

from pydantic import Field, PrivateAttr
from crewai.tools import BaseTool

try:
    from olostep import Olostep, Olostep_BaseError
except ImportError:
    raise ImportError(
        "The 'olostep' package is not installed. Run %pip install -U olostep."
    )

Commencez par charger la clé API Olostep depuis l’environnement.

olostep_api_key = os.getenv("OLOSTEP_API_KEY")

if not olostep_api_key or olostep_api_key == "your-olostep-api-key-here":
    print(
        "The Olostep tool will fail at call time unless OLOSTEP_API_KEY "
        "is set in the environment."
    )
else:
    print("Olostep API key loaded.")

_olostep_client = Olostep(api_key=olostep_api_key) if olostep_api_key else None

Créez maintenant l’outil CrewAI personnalisé.

class OlostepSearchTool(BaseTool):
    """Search the web with a strict research-call budget."""

    name: str = "olostep_web_search"
    description: str = (
        "Search the live web. Set scrape=false for fast discovery and "
        "scrape=true for full-page evidence. You have at most 3 discovery "
        "searches without scraping and 6 external search calls total. After "
        "3 discovery searches, use scrape=true on the strongest query or "
        "domain. Each call returns at most 3 results. Inputs: query, scrape, "
        "limit, and optional include_domains."
    )

    default_limit: int = Field(default=3, ge=1, le=3)
    max_chars_per_page: int = Field(default=4000, ge=500, le=12000)

    _total_calls: int = PrivateAttr(default=0)
    _discovery_calls: int = PrivateAttr(default=0)
    _call_history: list[dict] = PrivateAttr(default_factory=list)

    def _run(
        self,
        query: str,
        scrape: bool = False,
        limit: int | None = None,
        include_domains: list[str] | None = None,
    ) -> str:
        if _olostep_client is None:
            return (
                "OLOSTEP_API_KEY is not set. Set it in the environment "
                "and restart the kernel."
            )

        if self._total_calls >= 6:
            return (
                "Research search budget exhausted: 6 external calls have "
                "already been used. Stop searching and complete the task "
                "with the collected sources."
            )

        if not scrape and self._discovery_calls >= 3:
            return (
                "Discovery-search limit reached. Do not run another "
                "scrape=False search. Use scrape=True on the strongest query "
                "or selected domains, or finish with the sources already found."
            )

        clean_query = query.strip()
        if not clean_query:
            return "Search query cannot be empty."

        result_limit = max(1, min(limit or self.default_limit, 3))
        kwargs = {"query": clean_query, "limit": result_limit}

        if scrape:
            kwargs["scrape_options"] = {
                "formats": ["markdown"],
                "timeout": 25,
            }

        if include_domains:
            kwargs["include_domains"] = include_domains

        self._total_calls += 1
        if not scrape:
            self._discovery_calls += 1

        call_record = {
            "number": self._total_calls,
            "query": clean_query,
            "scrape": scrape,
            "limit": result_limit,
            "domains": include_domains or [],
            "status": "started",
            "results": 0,
        }
        self._call_history.append(call_record)

        try:
            search = _olostep_client.searches.create(**kwargs)
        except Olostep_BaseError as error:
            call_record["status"] = "error"
            call_record["error"] = f"{type(error).__name__}: {error}"
            return (
                f"Olostep search error: {type(error).__name__}: {error}\n"
                f"Budget used: {self._total_calls}/6 total calls; "
                f"{self._discovery_calls}/3 discovery calls."
            )

        if not search.links:
            call_record["status"] = "no_results"
            return (
                f"No results found for query: {clean_query!r}\n"
                f"Budget used: {self._total_calls}/6 total calls; "
                f"{self._discovery_calls}/3 discovery calls."
            )

        call_record["status"] = "success"
        call_record["results"] = len(search.links)

        chunks = [
            f"## Web results for: {clean_query!r}",
            f"Page scraping: {'enabled' if scrape else 'disabled'}",
            (
                f"Budget used: {self._total_calls}/6 total calls; "
                f"{self._discovery_calls}/3 discovery calls."
            ),
        ]

        for index, link in enumerate(search.links, 1):
            url = link.get("url", "")
            title = link.get("title") or url or "Untitled result"
            description = link.get("description") or "No description provided."

            chunks.append(
                f"### {index}. {title}\n"
                f"URL: {url}\n"
                f"Summary: {description}"
            )

            markdown = link.get("markdown_content")
            if scrape and markdown:
                body = markdown.strip()
                if len(body) > self.max_chars_per_page:
                    body = body[: self.max_chars_per_page] + "\n...[truncated]"
                chunks.append(f"Page content:\n{body}")

        return "\n\n".join(chunks)

Enfin, créez une instance de l’outil.

olostep_search_tool = OlostepSearchTool()
print(f"Tool ready: {olostep_search_tool.name}")

Vous devriez voir une sortie similaire :

Olostep API key loaded.
Tool ready: olostep_web_search

Cela confirme que l’outil de recherche web Olostep est prêt. Plus tard, nous l’attacherons au Researcher afin qu’il puisse collecter des informations à jour sur le web.

6. Créer les agents

Nous allons maintenant créer les agents de notre essaim. Nous définirons trois agents exécutants et un agent manager. Les agents exécutants font le travail principal, tandis que le manager coordonne le workflow.

Le Researcher utilisera l’outil Olostep pour trouver des informations à jour. Le Writer transformera la recherche en brouillon. Le Reviewer améliorera ce brouillon. Le Manager décidera de la délégation du travail dans la crew hiérarchique.

Créer les agents exécutants

Commençons par les trois agents exécutants.

from crewai import Agent

researcher = Agent(
    role="Senior Research Analyst",
    goal=(
        "Find accurate, current, and well-sourced information about {topic}. "
        "Produce concrete, verifiable points for the writer."
    ),
    backstory=(
        "You are a meticulous research analyst. You decide whether each web "
        "search needs page scraping. Use scrape=False when titles, URLs, and "
        "summaries are enough. Use scrape=True when you need full-page evidence "
        "to verify a claim or understand a source in detail. Cite source URLs."
    ),
    llm=gemini_llm,
    tools=[olostep_search_tool],
    verbose=True,
    allow_delegation=False,
)

Créez ensuite l’agent Writer.

writer = Agent(
    role="Blog Post Writer",
    goal=(
        "Turn the research into a complete, publication-ready educational "
        "article on {topic}. Include a strong title, an unlabeled italic "
        "summary, TL;DR, detailed body sections, conclusion, and FAQs. Add a "
        "table only when it makes a comparison or decision easier to understand."
    ),
    backstory=(
        "You are a seasoned technology educator and writer. You create "
        "practical, approachable articles similar in structure to high-quality "
        "DataCamp content, without copying its wording or branding. You use "
        "clear headings, short paragraphs, concrete examples, and sourced facts."
    ),
    llm=gemini_llm,
    verbose=True,
    allow_delegation=False,
)

Créez maintenant l’agent Reviewer.

reviewer = Agent(
    role="Senior Editor",
    goal=(
        "Review the draft for factual accuracy, clarity, structure, and "
        "completeness. Ensure it follows the required article order and return "
        "a polished, publication-ready version."
    ),
    backstory=(
        "You are a detail-oriented senior editor for an educational technology "
        "publication. You verify claims, improve flow, remove repetition, and "
        "ensure the title is followed by an unlabeled italic summary and TL;DR. "
        "You keep a table only when it adds real value, remove any Key Takeaways "
        "section, and ensure Conclusion appears immediately before FAQs."
    ),
    llm=gemini_llm,
    verbose=True,
    allow_delegation=False,
)

Créer l’agent manager

Enfin, créez l’agent Manager.

manager = Agent(
    role="Content Coordinator",
    goal=(
        "Coordinate research, writing, and review for {topic}. Delegate each "
        "task to the correct specialist and ensure the reviewed article is the "
        "final result."
    ),
    backstory=(
        "You are an experienced editorial lead. You keep the workflow simple: "
        "research first, writing second, and editorial review last."
    ),
    llm=gemini_llm,
    verbose=True,
    allow_delegation=True,
)

Le manager est configuré séparément lors de la création de la crew hiérarchique. C’est pourquoi il n’a pas encore été ajouté à la liste des agents exécutants.

7. Définir les tâches

Nous allons maintenant définir les tâches de l’essaim d’agents. Ces tâches ne sont pas directement assignées aux agents exécutants. C’est l’agent Manager qui décidera quel exécutant doit prendre chaque tâche.

Dans ce workflow, la première tâche est la recherche, la deuxième la rédaction et la troisième la relecture. Chaque tâche inclut aussi une production attendue, afin que les agents sachent précisément quoi livrer.

from crewai import Task

research_task = Task(
    description=(
        "Delegate to the Senior Research Analyst. Research '{topic}' using at "
        "most 6 web-tool calls. Use no more than 3 calls with scrape=False, "
        "then use scrape=True only when detailed evidence is needed. Collect "
        "5-7 useful findings, source URLs, and common reader questions."
    ),
    expected_output=(
        "Clear markdown research notes with 5-7 sourced findings and 3-5 "
        "potential FAQ questions."
    ),
)

Créez ensuite la tâche de rédaction. Cette tâche utilise la recherche comme contexte, afin que le Writer puisse construire l’article à partir des éléments collectés.

write_task = Task(
    description=(
        "Delegate to the Blog Post Writer. Write one complete educational "
        "markdown article on '{topic}', approximately 1,000-1,500 words. Start "
        "with one H1 title and an unlabeled italic summary. Then include TL;DR, "
        "an introduction, useful body sections, an optional table only when it "
        "improves understanding, a Conclusion, and FAQs. Use source links."
    ),
    expected_output=(
        "One continuous publication-ready markdown article with title, italic "
        "summary, TL;DR, body, Conclusion, and 3-5 FAQs."
    ),
    context=[research_task],
)

Créez maintenant la tâche de relecture. Elle utilise à la fois la recherche et la rédaction comme contexte, afin que le Reviewer puisse vérifier l’article par rapport aux sources.

review_task = Task(
    description=(
        "Delegate to the Senior Editor. Review the article for accuracy, "
        "clarity, structure, and source support. Return only the corrected full "
        "article. Keep the same continuous article format and ensure FAQs are "
        "the final section."
    ),
    expected_output=(
        "Only the final reviewed markdown article, beginning with its H1 title "
        "and ending with the final FAQ answer."
    ),
    context=[research_task, write_task],
)

Enfin, ajoutez les trois tâches à une liste.

tasks = [research_task, write_task, review_task]

print(f"Defined {len(tasks)} manager-delegated tasks.")

Vous devriez voir :

Defined 3 manager-delegated tasks.

8. Construire le workflow

Nous allons maintenant relier les agents et les tâches dans un workflow CrewAI unique. Nous utiliserons un processus hiérarchique dans lequel l’agent Manager coordonne les agents exécutants.

Dans cette configuration, le Researcher, le Writer et le Reviewer sont ajoutés comme agents exécutants. Le Manager est passé séparément via manager_agent, car il est responsable de la délégation et de la coordination.

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=tasks,
    manager_agent=manager,
    process=Process.hierarchical,
    verbose=True,
    memory=False,
)

Affichez maintenant les détails de la crew pour confirmer que le workflow a été assemblé correctement.

print("Hierarchical crew assembled.")
print(f"  Workers : {[a.role for a in crew.agents]}")
print(f"  Manager : {crew.manager_agent.role}")
print(f"  Tasks   : {len(crew.tasks)}")
print(f"  Process : {crew.process}")

Vous devriez voir une sortie similaire :

Hierarchical crew assembled.
  Workers : ['Senior Research Analyst', 'Blog Post Writer', 'Senior Editor']
  Manager : Content Coordinator
  Tasks   : 3
  Process : Process.hierarchical

Cela confirme que l’essaim hiérarchique d’agents est prêt. À l’étape suivante, nous exécuterons la crew sur un sujet et générerons la réponse finale.

9. Exécuter l’essaim d’agents

Nous pouvons maintenant exécuter l’essaim d’agents. Nous fournirons un sujet et démarrerons le workflow avec kickoff_async().

Jupyter exécute déjà une boucle d’événements en arrière-plan. Pour cette raison, nous appliquons d’abord nest_asyncio afin que le workflow asynchrone CrewAI puisse s’exécuter dans le notebook.

try:
    import nest_asyncio

    nest_asyncio.apply()

    print("nest_asyncio applied: CrewAI can now run inside this kernel.")

except ModuleNotFoundError:
    print("nest_asyncio not installed. Run: pip install nest-asyncio")

Vous devriez voir :

nest_asyncio applied: CrewAI can now run inside this kernel.

Ensuite, définissez un sujet et créez une fonction utilitaire pour exécuter la crew.

from datetime import datetime

topic = "Learning Adaptive Thresholds for LLM Tool Use: When to Rely on Memory and When to Call a Tool"

async def run_crew(topic: str):
    """Run the CrewAI crew asynchronously."""
    print(f"Starting crew at {datetime.now().isoformat(timespec='seconds')}")
    print(f"Topic: {topic!r}\n")

    result = await crew.kickoff_async(inputs={"topic": topic})

    print(f"\nFinished at {datetime.now().isoformat(timespec='seconds')}")
    print(f"Result type: {type(result).__name__}")
    return result

result = await run_crew(topic)

Au démarrage, CrewAI confie d’abord la tâche à l’agent Manager. Le Manager délègue ensuite la recherche au Senior Research Analyst. Le Researcher utilise l’outil olostep_web_search pour rechercher des informations sur les seuils adaptatifs, l’usage d’outils par les LLM, la mémoire et les décisions d’appel d’outil.

Dans le journal, le Researcher utilise à la fois des appels de recherche simple et de scraping. Le budget de recherche est entièrement consommé, atteignant 6 appels web sur 6 et 3 recherches exploratoires sur 3. Une requête supplémentaire sans scraping est bloquée car la limite d’exploration est atteinte. Cela montre que les règles de budget de notre outil Olostep personnalisé fonctionnent correctement.

Après la phase de recherche, le Manager poursuit le workflow en déléguant la rédaction puis la relecture. Les agents produisent des notes de recherche, rédigent l’article, puis polissent la réponse finale. L’exécution se termine avec succès et renvoie un objet CrewOutput.

Vous devriez voir une sortie similaire :

Starting crew at 2026-06-05T13:54:07
Topic: 'Learning Adaptive Thresholds for LLM Tool Use: When to Rely on Memory and When to Call a Tool'

Crew Execution Started
Task Started
Agent: Content Coordinator
Agent: Senior Research Analyst
Tool: olostep_web_search
Budget used: 6/6 total calls; 3/3 discovery calls

Finished at 2026-06-05T13:59:29
Result type: CrewOutput

Cela signifie que l’essaim complet a bien tourné. Le Manager a coordonné le workflow, le Researcher a collecté des informations à jour, le Writer a créé le brouillon et le Reviewer a amélioré la version finale.

10. Afficher le résultat

Après l’exécution, nous pouvons afficher l’article final généré par la crew. La valeur result.raw contient la réponse finale du workflow.

from IPython.display import Markdown, display

raw = result.raw

if "Blog Post" in raw:
    raw = raw.split("Blog Post", 1)[1].strip()

display(Markdown(raw))

Cela va rendre la sortie finale au format markdown dans le notebook. 

Dans la plupart des cas, l’article généré sera déjà bien structuré, avec plusieurs intertitres, des sections claires, des schémas ou des explications visuelles si nécessaire, et une conclusion solide. 

La combinaison recherche, rédaction et relecture éditoriale aide à produire un contenu prêt à publier. Après relecture, je lui donnerais personnellement le feu vert pour publication.

Ensuite, nous pouvons imprimer une trace d’exécution. Cela aide à comprendre comment l’essaim a fonctionné, quels agents ont participé, combien d’appels web ont été utilisés et comment chaque tâche a été traitée.

from collections import Counter

def role_name(value):
    """Return a readable role/name for CrewAI agent values."""
    if value is None:
        return "Not reported"
    if isinstance(value, str):
        return value
    return getattr(value, "role", None) or getattr(value, "name", None) or str(value)

print("=" * 70)
print("SWARM EXECUTION TRACE")
print("=" * 70)

# Hierarchical crew structure
print("\nCOORDINATION")
print(f"Process : {crew.process}")
print(f"Manager : {role_name(crew.manager_agent)}")
print(f"Workers : {', '.join(role_name(agent) for agent in crew.agents)}")
print(
    "Flow    : Manager receives each task, delegates specialist work, "
    "validates the response, and passes approved context to the next task."
)

Now print the web-tool usage recorded by the custom Olostep tool.
history = list(getattr(olostep_search_tool, "_call_history", []))
discovery_calls = sum(not item["scrape"] for item in history)
scrape_calls = sum(item["scrape"] for item in history)
successful_calls = sum(item["status"] == "success" for item in history)

print("\nWEB TOOL USAGE")
print(f"Total external calls : {len(history)}/6")
print(f"Without scraping     : {discovery_calls}/3")
print(f"With scraping        : {scrape_calls}")
print(f"Successful calls     : {successful_calls}")

if history:
    for item in history:
        mode = "scrape" if item["scrape"] else "search only"
        domains = ", ".join(item["domains"]) if item["domains"] else "all domains"
        print(
            f"  {item['number']}. {mode}; status={item['status']}; "
            f"results={item['results']}; scope={domains}"
        )
        print(f"     Query: {item['query']}")
else:
    print("  No Olostep calls were recorded.")

Finally, print the task-by-task trace.
print("\nTASK-BY-TASK TRACE")
agent_activity = Counter()

for index, task_out in enumerate(result.tasks_output, start=1):
    task = crew.tasks[index - 1] if index <= len(crew.tasks) else None
    reported_agent = role_name(getattr(task_out, "agent", None))
    processed_by = sorted(getattr(task, "processed_by_agents", set()) or [])
    participants = processed_by or [reported_agent]

    for participant in participants:
        agent_activity[role_name(participant)] += 1

    delegations = getattr(task, "delegations", 0) if task else 0
    used_tools = getattr(task, "used_tools", 0) if task else 0
    tool_errors = getattr(task, "tools_errors", 0) if task else 0
    start_time = getattr(task, "start_time", None) if task else None
    end_time = getattr(task, "end_time", None) if task else None
    duration = end_time - start_time if start_time and end_time else None

    print(f"\n--- Task {index} ---")
    print(f"Reported owner : {reported_agent}")
    print(f"Processed by   : {', '.join(map(role_name, participants))}")
    print(f"Delegations    : {delegations}")
    print(f"Tool uses      : {used_tools}")
    print(f"Tool errors    : {tool_errors}")
    if duration is not None:
        print(f"Duration       : {duration}")

    preview = getattr(task_out, "raw", "").strip().replace("\n", " ")
    print(f"Output preview : {preview[:300]}{'...' if len(preview) > 300 else ''}")

print("\nAGENT ACTIVITY")
for agent, count in agent_activity.most_common():
    print(f"{agent}: participated in {count} task(s)")

if not agent_activity:
    print("CrewAI did not expose per-agent processing metadata for this run.")

Vous devriez voir une sortie similaire :

Cette trace montre que l’essaim hiérarchique a fonctionné comme prévu. Le Manager a coordonné les trois tâches, le Researcher a géré la recherche, le Writer a rédigé le brouillon, et le Reviewer a amélioré la réponse finale.

Cependant, elle met aussi en évidence une limite importante : les workflows multi-agents peuvent devenir coûteux. 

Lors de cette exécution, le workflow a utilisé 6 appels web externes et plusieurs appels LLM répartis entre Manager, Researcher, Writer et Reviewer. Une exécution peut coûter jusqu’à environ 0,28 $, et la création de ce guide a coûté environ 2,78 $ pendant les tests. C’est élevé pour un simple tutoriel.

Conclusion

Sur le papier, les essaims d’agents et les workflows multi-agents sont séduisants. L’idée d’un Researcher, d’un Writer, d’un Reviewer et d’un Manager travaillant de concert est puissante, et dans ce guide, le résultat final était effectivement publiable après quelques retouches.

Mais en pratique, cette configuration peut coûter plus cher, prendre plus de temps et multiplier les points de défaillance. Dans mon cas, une exécution coûte environ 0,28 $, et la rédaction de ce guide a coûté environ 2,78 $. C’est beaucoup pour un simple tutoriel.

Pour des applications réelles, je n’opterais pas toujours pour un essaim multi-agents complet. Je préférerais une approche plus programmatique, où les tâches simples sont gérées par du code ou des règles, et seules les tâches complexes sont confiées à des agents. On peut aussi réduire les coûts en limitant les appels d’outils, en utilisant moins d’agents, en raccourcissant les prompts et en privilégiant un workflow linéaire plutôt que hiérarchique.

Évidemment, si l’on optimise trop, il ne s’agit plus vraiment d’un essaim d’agents : on se rapproche d’un workflow IA piloté par des règles. L’objectif est donc l’équilibre : utilisez des essaims d’agents lorsqu’ils apportent une vraie valeur, mais restez simple lorsqu’ils n’en apportent pas.