Tutorial Agent Swarm: coordina agenti AI con CrewAI

Che cos’è uno sciame di agenti?

Uno sciame di agenti è un gruppo di agenti AI che collaborano per completare un task. Invece di chiedere a un solo modello di ricercare, scrivere, revisionare e finalizzare tutto da solo, uno sciame di agenti divide il lavoro tra più agenti specializzati. Ogni agente ha un ruolo chiaro e il sistema li coordina per produrre un risultato finale più solido.

Il vantaggio è la specializzazione. Un agente focalizzato sulla ricerca può concentrarsi sul trovare e verificare le fonti, uno focalizzato sulla scrittura può curare struttura e chiarezza, e un revisore può controllare la bozza rispetto alla ricerca originale senza essere distratto dal testo che ha appena prodotto. 

Un agente coordinatore collega i passaggi e decide cosa succede dopo. Nello sciame che stiamo costruendo, questi ruoli corrispondono direttamente a Ricercatore, Scrittore, Revisore e Manager.

Non è però “gratis”. Più agenti significano più chiamate al modello, più tempo e più punti in cui il workflow può fallire; un compromesso a cui torneremo alla fine, dopo aver visto quanto costa davvero l’esecuzione.

Che cos’è CrewAI?

CrewAI è un framework open source per costruire sistemi multi-agente. Offre un modo pulito per definire gli agenti, assegnare loro i task e scegliere come coordinarli, così puoi concentrarti su ruoli e workflow invece che sulla “plumbing” che li collega.

Tre concetti fanno la maggior parte del lavoro. 

• Agenti sono i lavoratori individuali, ognuno definito da un ruolo, un obiettivo e una backstory che ne guida il comportamento. 
• Task descrivono il lavoro da svolgere e l’output che ciascuno dovrebbe produrre. 
• Una crew riunisce agenti e task ed esegue il tutto con un processo scelto

Questo processo può essere sequenziale, con i task in ordine fisso, oppure gerarchico, dove un agente manager delega il lavoro e ne valida l’output. Qui useremo il processo gerarchico, con il Manager a gestire la delega.

CrewAI gestisce anche l’accesso ai modelli tramite LiteLLM, una libreria open source che offre un’interfaccia unificata per molti provider. È ciò che ci permette di definire un unico modello Gemini 3.5 Flash e riutilizzarlo in tutti e quattro gli agenti, che imposteremo nei passaggi sotto.

Costruire uno sciame di agenti con CrewAI e Gemini 3.5 Flash

Iniziamo con la configurazione.

1. Prerequisiti

Prima di iniziare, assicurati di avere:

• Python 3.11 o successivo installato
• Jupyter Notebook o JupyterLab installato
• Nozioni di base di Python e notebook
• Un account Google AI Studio
• Fatturazione attiva o crediti aggiunti al tuo account Google per l’uso di Gemini
• Un account Olostep gratuito

2. Installa le dipendenze

Avvia Jupyter Notebook o JupyterLab e apri un notebook con il kernel Python più recente disponibile nel tuo ambiente. Poi esegui la seguente cella per installare i pacchetti richiesti.

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

Dopo l’installazione, esegui la prossima cella per controllare le versioni dei pacchetti installati. Aiuta a confermare che tutto sia stato installato correttamente.

import importlib.metadata as md

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

Dovresti vedere un output simile a questo:

         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. Imposta le variabili d’ambiente

Ti serviranno due chiavi API per questo progetto. 

Per prima cosa, crea una chiave API Gemini da Google AI Studio. Apri Google AI Studio, vai alla sezione API key e crea una nuova chiave per il tuo progetto. Gemini offre un livello gratuito, ma con limiti di utilizzo e di rate. Per verifiche ed esecuzioni più affidabili con Gemini, collega il tuo account di fatturazione Google o aggiungi crediti al tuo account, così l’uso può essere fatturato oltre i limiti del free tier.

Poi crea una chiave API Olostep. Registrati su Olostep, apri la dashboard e genera una chiave API dalla pagina API keys. Useremo questa chiave per dare all’agente Ricercatore l’accesso live al web per ricerche e scraping delle pagine.

Una volta pronte entrambe le chiavi, salvale come variabili d’ambiente. Gemini alimenterà gli agenti CrewAI, mentre Olostep consentirà al Ricercatore di raccogliere informazioni aggiornate dal web.

Dopo aver creato entrambe le chiavi, salvale come variabili d’ambiente prima di eseguire il 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"

Ora esegui la seguente cella per verificare che la chiave API di Gemini sia caricata correttamente.

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")

Dovresti vedere: 

Gemini API key loaded

4. Configura Gemini 3.5 Flash tramite LiteLLM

In questo progetto useremo Gemini 3.5 Flash come LLM condiviso per tutti gli agenti CrewAI. Gemini 3.5 Flash è progettato per carichi di lavoro rapidi, agentici e orientati al coding. È utile qui perché gli agenti devono ragionare su più passi, seguire istruzioni, usare strumenti e produrre output strutturati.

Gemini 3.5 Flash supporta anche una grande finestra di contesto, output lunghi, “thinking” e tool use, rendendolo una scelta forte per i workflow basati su agenti. Poiché uno sciame può chiamare il modello più volte tra Ricercatore, Scrittore, Revisore e Manager, usare un modello Flash aiuta a mantenere il workflow reattivo offrendo comunque buon ragionamento e qualità di scrittura.

Se vuoi confrontare il modello con altri LLM all’avanguardia, ti consiglio di leggere le nostre guide su Gemini 3.5 Flash vs GPT-5.5 e Gemini 3.5 Flash vs Claude Opus 4.8.

Accederemo a Gemini tramite LiteLLM. LiteLLM è una libreria open source che offre un’interfaccia unificata per chiamare molti provider LLM, inclusi Gemini, OpenAI, Anthropic, Bedrock, Vertex AI e altri. In questa configurazione, CrewAI usa LiteLLM dietro le quinte così possiamo passare il nome del modello Gemini in un semplice formato provider/model.

Ora creeremo un LLM CrewAI condiviso e lo riuseremo per tutti gli agenti. Mantiene la configurazione semplice perché Ricercatore, Scrittore, Revisore e Manager useranno tutti lo stesso modello 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)")

Dovresti vedere un output simile a questo:

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

Questo conferma che CrewAI può ora accedere a Gemini tramite LiteLLM. Nei prossimi passi passeremo questo oggetto condiviso gemini_llm a ciascun agente.

5. Aggiungi l’accesso web live

L’agente Ricercatore ha bisogno di accedere a informazioni aggiornate. Per farlo, creeremo uno strumento di ricerca Olostep personalizzato e lo assegneremo al Ricercatore in seguito.

Questo tool può lavorare in due modalità. Può restituire solo sommari di ricerca, più veloce, oppure può fare scraping delle pagine dei risultati e restituire il contenuto in markdown. Aggiungiamo anche un piccolo budget di ricerca per evitare che l’agente chiami lo strumento web troppe volte.

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."
    )

Per prima cosa, carica la chiave API Olostep dall’ambiente.

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

Ora crea lo strumento CrewAI personalizzato.

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)

Infine, crea un’istanza dello strumento.

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

Dovresti vedere un output simile a questo:

Olostep API key loaded.
Tool ready: olostep_web_search

Questo conferma che lo strumento di ricerca web Olostep è pronto. Più avanti lo collegheremo al Ricercatore così potrà raccogliere informazioni aggiornate dal web.

6. Crea gli agenti

Ora creeremo gli agenti per il nostro sciame. Definiremo tre agenti operativi e un agente manager. Gli agenti operativi svolgono i task principali, mentre il manager coordina il workflow.

Il Ricercatore userà lo strumento Olostep per trovare informazioni aggiornate. Lo Scrittore trasformerà la ricerca in una bozza. Il Revisore migliorerà la bozza. Il Manager deciderà come delegare il lavoro nella crew gerarchica.

Crea gli agenti operativi

Partiamo dai tre agenti operativi.

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,
)

Ora crea l’agente Scrittore.

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,
)

Ora crea l’agente Revisore.

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,
)

Crea l’agente manager

Infine, crea l’agente 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,
)

Il manager viene configurato separatamente quando costruiamo la crew gerarchica. Per questo non è stato ancora aggiunto alla lista degli agenti operativi.

7. Definisci i task

Ora definiremo i task per lo sciame di agenti. Questi task non sono assegnati direttamente agli agenti operativi. Sarà l’agente Manager a decidere quale operatore debba gestire ciascun task.

In questo workflow, il primo task è la ricerca, il secondo è la scrittura e il terzo è la revisione. Ogni task include anche un output atteso, così gli agenti sanno esattamente cosa produrre.

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."
    ),
)

Successivamente, crea il task di scrittura. Questo task usa il task di ricerca come contesto, così lo Scrittore può costruire l’articolo partendo dalla ricerca raccolta.

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],
)

Ora crea il task di revisione. Questo task usa come contesto sia la ricerca che la scrittura, così il Revisore può confrontare l’articolo con la ricerca originale.

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],
)

Infine, aggiungi tutti e tre i task a una lista.

tasks = [research_task, write_task, review_task]

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

Dovresti vedere:

Defined 3 manager-delegated tasks.

8. Costruisci il workflow

Ora collegheremo agenti e task in un unico workflow CrewAI. Useremo un processo gerarchico in cui l’agente Manager coordina gli agenti operativi.

In questa configurazione, Ricercatore, Scrittore e Revisore sono aggiunti come agenti operativi. Il Manager è passato separatamente tramite manager_agent, perché è responsabile di delega e coordinamento.

from crewai import Crew, Process

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

Ora stampa i dettagli della crew per confermare che il workflow sia stato assemblato correttamente.

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}")

Dovresti vedere un output simile a questo:

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

Questo conferma che lo sciame di agenti gerarchico è pronto. Nel prossimo passo eseguiremo la crew su un argomento e genereremo la risposta finale.

9. Esegui lo sciame di agenti

Ora possiamo eseguire lo sciame. Forniremo un argomento e avvieremo il workflow con kickoff_async().

Jupyter esegue già un event loop in background. Per questo, applichiamo prima nest_asyncio, così il workflow asincrono di CrewAI può girare dentro il 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")

Dovresti vedere:

nest_asyncio applied: CrewAI can now run inside this kernel.

Poi definisci un argomento e crea una funzione di supporto per eseguire 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)

Quando il workflow parte, CrewAI affida per prima cosa il task al Manager. Il Manager delega poi il lavoro di ricerca al Senior Research Analyst. Il Ricercatore usa lo strumento olostep_web_search per cercare informazioni sul web riguardo soglie adattive, uso degli strumenti da parte degli LLM, memoria e decisioni di chiamata degli strumenti.

Nel log, il Ricercatore usa sia chiamate di sola ricerca sia chiamate con scraping. Il budget di ricerca viene usato completamente, arrivando a 6 chiamate web su 6 totali e 3 chiamate di discovery su 3. Una richiesta extra di sola ricerca viene bloccata perché il limite di discovery è stato raggiunto. Questo mostra che le regole di budget nel nostro strumento Olostep personalizzato funzionano correttamente.

Dopo la fase di ricerca, il Manager prosegue delegando scrittura e revisione. Gli agenti producono note di ricerca, redigono l’articolo e poi rifiniscono la risposta finale. L’esecuzione termina con successo e restituisce un oggetto CrewOutput.

Dovresti vedere un output simile a questo:

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

Questo significa che l’intero sciame ha girato con successo. Il Manager ha coordinato il workflow, il Ricercatore ha raccolto informazioni aggiornate, lo Scrittore ha creato la bozza e il Revisore ha migliorato l’output finale.

10. Visualizza il risultato

Dopo che lo sciame ha finito di girare, possiamo mostrare l’articolo finale generato dalla crew. Il valore result.raw contiene la risposta finale del 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))

Questo renderizzerà l’output finale come markdown formattato dentro il notebook. 

Nella maggior parte dei casi, il blog generato sarà già ben strutturato, con più intestazioni, sezioni chiare, diagrammi o spiegazioni visuali dove opportuno e una conclusione solida. 

La combinazione di ricerca, scrittura e revisione editoriale aiuta a produrre contenuti pronti per la pubblicazione. Dopo aver rivisto l’output, personalmente darei il via libera alla pubblicazione.

Successivamente, possiamo stampare una traccia di esecuzione. Aiuta a capire come ha lavorato lo sciame, quali agenti hanno partecipato, quante chiamate web sono state usate e come è stato elaborato ciascun task.

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.")

Dovresti vedere un output simile a questo:

Questa traccia mostra che lo sciame gerarchico ha funzionato come previsto. Il Manager ha coordinato tutti e tre i task, il Ricercatore ha gestito la fase di ricerca, lo Scrittore ha creato la bozza dell’articolo e il Revisore ha migliorato la risposta finale.

Tuttavia, evidenzia anche un limite importante: i workflow multi-agente possono diventare costosi. 

In questa esecuzione, il workflow ha usato 6 chiamate web esterne e più chiamate LLM tra Manager, Ricercatore, Scrittore e Revisore. Una singola run può costare fino a circa $0,28 e creare questa guida è costato circa $2,78 durante i test. È alto per un semplice progetto tutorial.

Considerazioni finali

Gli sciami di agenti e i workflow multi-agente sulla carta sono fantastici. L’idea di avere Ricercatore, Scrittore, Revisore e Manager che lavorano insieme è potente e, in questa guida, l’output finale era davvero abbastanza buono da pubblicare con minime modifiche.

Ma nella pratica questa configurazione può costare di più, richiedere più tempo e creare più punti di failure. Nel mio caso, una run costa circa $0,28 e creare questa guida è costato circa $2,78. È tanto per un progetto tutorial semplice.

Per applicazioni reali, non userei sempre un intero sciame di agenti. Preferirei una configurazione più programmatica in cui i task semplici sono gestiti da codice o regole, e solo i task complessi sono affidati agli agenti. Possiamo anche ridurre i costi limitando le chiamate agli strumenti, usando meno agenti, accorciando i prompt e scegliendo un workflow lineare invece che gerarchico.

Certo, se ottimizziamo troppo, potrebbe non essere più un vero sciame di agenti. Diventa più simile a un workflow AI basato su regole. Per questo l’obiettivo è l’equilibrio: usa gli sciami quando aggiungono vero valore, ma mantieni la semplicità quando non lo fanno.