Tutorial de agent swarm: coordina agentes de IA con CrewAI

¿Qué es un agent swarm?

Un agent swarm es un grupo de agentes de IA que trabajan juntos para completar una tarea. En lugar de pedir a un único modelo que investigue, escriba, revise y finalice todo por su cuenta, un agent swarm reparte el trabajo entre varios agentes especializados. Cada agente tiene un rol claro, y el sistema los coordina para producir un resultado final más sólido.

La ventaja es la especialización. Un agente centrado en investigación puede dedicarse a encontrar y verificar fuentes; uno centrado en redacción puede enfocarse en la estructura y la claridad; y un revisor puede contrastar el borrador con la investigación original sin distraerse con lo que acaba de escribir. 

Un agente coordinador une los pasos y decide qué ocurre a continuación. En el swarm que estamos construyendo, esos roles se corresponden con Researcher, Writer, Reviewer y Manager.

Eso sí, no sale gratis. Más agentes implican más llamadas a modelos, más tiempo y más puntos donde el flujo puede fallar; una contrapartida a la que volveremos al final, cuando veas cuánto cuesta ejecutarlo.

¿Qué es CrewAI?

CrewAI es un framework de código abierto para crear sistemas multiagente. Te permite definir agentes, asignarles tareas y elegir cómo se coordinan de forma clara, para que te centres en los roles y el flujo de trabajo, no en la fontanería que los conecta.

Tres conceptos llevan el peso. 

• Agents son los trabajadores individuales, cada uno definido por un rol, un objetivo y un trasfondo que guía su comportamiento. 
• Tasks describen el trabajo a realizar y el output que debe producir cada una. 
• Una crew une agentes y tareas y los ejecuta con un proceso elegido

Este proceso puede ser secuencial, con tareas en orden fijo, o jerárquico, donde un agente manager delega trabajo y valida outputs. Aquí usaremos el proceso jerárquico, con el Manager encargándose de la delegación.

CrewAI también gestiona el acceso a modelos a través de LiteLLM, una librería open source que ofrece una interfaz unificada para muchos proveedores. Esto nos permite definir un único modelo Gemini 3.5 Flash y reutilizarlo en los cuatro agentes, como verás a continuación.

Crear un agent swarm con CrewAI y Gemini 3.5 Flash

Vamos con la configuración.

1. Requisitos previos

Antes de empezar, asegúrate de tener:

• Python 3.11 o superior instalado
• Jupyter Notebook o JupyterLab instalado
• Conocimientos básicos de Python y notebooks
• Una cuenta en Google AI Studio
• Facturación activada o créditos añadidos a tu cuenta de Google para usar Gemini
• Una cuenta gratuita de Olostep

2. Instala las dependencias

Inicia Jupyter Notebook o JupyterLab y abre un notebook con el kernel de Python más reciente disponible en tu entorno. Ejecuta esta celda para instalar los paquetes necesarios.

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

Cuando termine la instalación, ejecuta la siguiente celda para comprobar las versiones instaladas. Así confirmas que todo se instaló correctamente.

import importlib.metadata as md

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

Deberías ver algo parecido a esto:

         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. Configura las variables de entorno

Necesitarás dos claves de API para este proyecto. 

Primero, crea una clave de API de Gemini desde Google AI Studio. Abre Google AI Studio, ve a la sección de claves de API y crea una nueva clave para tu proyecto. Gemini ofrece una capa gratuita, pero con uso y límites de tasa restringidos. Para probar y ejecutar Gemini con más fiabilidad, vincula tu cuenta de facturación de Google o añade créditos para que el uso se facture cuando superes los límites de la capa gratuita.

Después, crea una clave de API de Olostep. Regístrate en Olostep, abre el panel de Olostep y genera una clave en la página de claves de API. Usaremos esta clave para dar al agente Researcher acceso web en vivo para búsqueda y scraping de páginas.

Cuando tengas ambas claves, guárdalas como variables de entorno. Gemini alimentará los agentes de CrewAI, mientras que Olostep permitirá que el Researcher recopile información actual de la web.

Tras crear ambas claves, guárdalas como variables de entorno antes de ejecutar el 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"

Ahora ejecuta la siguiente celda para comprobar que la clave de la API de Gemini se ha cargado correctamente.

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

Deberías ver: 

Gemini API key loaded

4. Configura Gemini 3.5 Flash con LiteLLM

En este proyecto usaremos Gemini 3.5 Flash como LLM compartido para todos los agentes de CrewAI. Gemini 3.5 Flash está pensado para cargas rápidas, agentic y centradas en código. Es útil aquí porque nuestros agentes deben razonar en varios pasos, seguir instrucciones, utilizar herramientas y producir outputs estructurados.

Gemini 3.5 Flash también admite una ventana de contexto amplia, outputs largos, reasoning avanzado y uso de herramientas, lo que lo convierte en una gran opción para flujos con agentes. Como un agent swarm puede llamar al modelo varias veces entre Researcher, Writer, Reviewer y Manager, usar un modelo Flash ayuda a mantener el flujo ágil sin sacrificar calidad de razonamiento y redacción.

Si quieres comparar el modelo con otros LLM de vanguardia, te recomiendo leer nuestras guías sobre Gemini 3.5 Flash vs GPT-5.5 y Gemini 3.5 Flash vs Claude Opus 4.8.

Accederemos a Gemini a través de LiteLLM. LiteLLM es una librería open source que ofrece una interfaz unificada para llamar a muchos proveedores de LLM, incluidos Gemini, OpenAI, Anthropic, Bedrock, Vertex AI y otros. En esta configuración, CrewAI usa LiteLLM por debajo para que podamos pasar el nombre del modelo de Gemini en un formato simple provider/model.

Ahora crearemos un LLM compartido en CrewAI y lo reutilizaremos en todos los agentes. Así mantenemos la configuración sencilla porque Researcher, Writer, Reviewer y Manager usarán el mismo modelo de 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)")

Deberías ver algo parecido a esto:

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

Esto confirma que CrewAI puede acceder a Gemini a través de LiteLLM. En los siguientes pasos pasaremos este objeto compartido gemini_llm a cada agente.

5. Añade acceso web en vivo

El agente Researcher necesita acceso a información actual. Para ello, crearemos una herramienta de búsqueda de Olostep personalizada y se la asignaremos después.

Esta herramienta puede trabajar en dos modos. Puede devolver solo resúmenes de búsqueda, más rápido, o hacer scraping de las páginas de resultados y devolver contenido en markdown. También añadimos un pequeño presupuesto de búsquedas para que el agente no abuse de la herramienta.

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

Primero, carga la clave de API de Olostep desde el entorno.

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

Ahora crea la herramienta personalizada de CrewAI.

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)

Por último, crea una instancia de la herramienta.

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

Deberías ver algo así:

Olostep API key loaded.
Tool ready: olostep_web_search

Esto confirma que la herramienta de búsqueda web de Olostep está lista. Más adelante la conectaremos al agente Researcher para que pueda recopilar información actual de la web.

6. Crea los agentes

Ahora crearemos los agentes de nuestro swarm. Definiremos tres agentes de trabajo y un agente manager. Los agentes de trabajo ejecutan las tareas principales, mientras que el manager coordina el flujo.

El agente Researcher usará la herramienta de Olostep para encontrar información actual. El agente Writer convertirá la investigación en un borrador. El agente Reviewer mejorará el borrador. El agente Manager decidirá cómo delegar el trabajo en la crew jerárquica.

Crea los agentes de trabajo

Empecemos por los tres agentes de trabajo.

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

A continuación, crea el agente 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,
)

Ahora crea el agente 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,
)

Crea el agente manager

Por último, crea el 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,
)

El manager se configura aparte cuando construimos la crew jerárquica. Por eso aún no se ha añadido a la lista de agentes de trabajo.

7. Define las tareas

Ahora definiremos las tareas del agent swarm. Estas tareas no se asignan directamente a los agentes de trabajo. En su lugar, el agente Manager decidirá qué especialista debe encargarse de cada una.

En este flujo, la primera tarea es investigar, la segunda escribir y la tercera revisar. Cada tarea incluye también un output esperado, para que los agentes sepan exactamente qué deben producir.

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

A continuación, crea la tarea de redacción. Esta tarea usa la de investigación como contexto, para que el Writer construya el artículo a partir de lo recopilado.

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

Ahora, crea la tarea de revisión. Esta tarea usa como contexto tanto la investigación como la redacción, para que el Reviewer contraste el artículo con la investigación original.

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

Por último, añade las tres tareas a una lista.

tasks = [research_task, write_task, review_task]

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

Deberías ver:

Defined 3 manager-delegated tasks.

8. Construye el flujo

Ahora conectaremos los agentes y las tareas en un único flujo de CrewAI. Usaremos un proceso jerárquico en el que el Manager coordina a los agentes de trabajo.

En esta configuración, Researcher, Writer y Reviewer se añaden como agentes de trabajo. El Manager se pasa aparte mediante manager_agent, porque es el responsable de la delegación y la coordinación.

from crewai import Crew, Process

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

Ahora imprime los detalles de la crew para confirmar que el flujo se montó correctamente.

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

Deberías ver algo parecido a esto:

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

Esto confirma que el agent swarm jerárquico está listo. En el siguiente paso, ejecutaremos la crew con un tema y generaremos la respuesta final.

9. Ejecuta el agent swarm

Ahora podemos ejecutar el agent swarm. Proporcionaremos un tema e iniciaremos el flujo con kickoff_async().

Jupyter ya ejecuta un event loop en segundo plano. Por eso primero aplicamos nest_asyncio, para que el flujo asíncrono de CrewAI pueda ejecutarse dentro del 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")

Deberías ver:

nest_asyncio applied: CrewAI can now run inside this kernel.

Después, define un tema y crea una función auxiliar para ejecutar 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)

Cuando arranca el flujo, CrewAI primero asigna la tarea al agente Manager. El Manager delega la investigación al Senior Research Analyst. El Researcher usa la herramienta olostep_web_search para buscar información sobre umbrales adaptativos, uso de herramientas por LLM, memoria y decisiones de llamadas a herramientas.

En el log, el Researcher usa tanto llamadas solo de búsqueda como con scraping. El presupuesto de búsqueda se agota, llegando a 6 de 6 llamadas web totales y 3 de 3 llamadas de descubrimiento. Una solicitud adicional sin scraping se bloquea porque ya se alcanzó el límite de descubrimiento. Esto demuestra que las reglas de presupuesto dentro de nuestra herramienta personalizada de Olostep funcionan correctamente.

Tras la fase de investigación, el Manager continúa el flujo delegando los pasos de redacción y revisión. Los agentes producen notas de investigación, redactan el artículo y luego pulen la respuesta final. La ejecución termina con éxito y devuelve un objeto CrewOutput.

Deberías ver un output similar a este:

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

Esto significa que el swarm completo se ejecutó correctamente. El Manager coordinó el flujo, el Researcher recopiló información actual, el Writer creó el borrador y el Reviewer mejoró el resultado final.

10. Visualiza el resultado

Cuando termine el swarm, podemos mostrar el artículo final generado por la crew. El valor result.raw contiene la respuesta final del flujo.

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

Esto renderizará el output final como markdown con formato dentro del notebook. 

En la mayoría de los casos, el blog generado ya estará bien estructurado, con varios encabezados, secciones claras, diagramas o explicaciones visuales cuando corresponda, y una conclusión sólida. 

La combinación de investigación, redacción y revisión editorial ayuda a producir contenido listo para publicar. Tras revisar el output, personalmente le daría luz verde para su publicación.

A continuación, podemos imprimir un rastro de ejecución. Esto ayuda a entender cómo trabajó el swarm, qué agentes participaron, cuántas llamadas web se usaron y cómo se procesó cada tarea.

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

Deberías ver un output similar a este:

Esta traza muestra que el swarm jerárquico funcionó como se esperaba. El Manager coordinó las tres tareas, el Researcher llevó la fase de investigación, el Writer creó el borrador y el Reviewer mejoró la respuesta final.

Sin embargo, también deja clara una limitación importante: los flujos multiagente pueden salir caros. 

En esta ejecución, el flujo usó 6 llamadas web externas y múltiples llamadas al LLM entre Manager, Researcher, Writer y Reviewer. Una sola ejecución puede costar alrededor de 0,28 $, y crear esta guía costó unos 2,78 $ durante las pruebas. Es un coste alto para un tutorial sencillo.

Reflexiones finales

Los agent swarms y los flujos multiagente pintan muy bien sobre el papel. La idea de tener un Researcher, un Writer, un Reviewer y un Manager trabajando juntos es potente y, en esta guía, el resultado final fue realmente publicable con ediciones menores.

Pero en la práctica, esta configuración puede costar más, tardar más y crear más puntos de fallo. En mi caso, una ejecución cuesta en torno a 0,28 $, y crear esta guía costó unos 2,78 $. Es mucho para un proyecto tutorial simple.

Para aplicaciones reales, no siempre optaría por un swarm multiagente completo. Preferiría una configuración más programática donde las tareas simples las resuelvan código o reglas, y solo las complejas pasen a agentes. También podemos recortar costes limitando las llamadas a herramientas, usando menos agentes, acortando prompts y pasando a un flujo lineal en lugar de jerárquico.

Claro que, si optimizamos demasiado, puede que deje de ser un agent swarm como tal y pase a ser más bien un flujo de IA basado en reglas. Por eso la clave es el equilibrio: usa swarms cuando aporten valor real y mantén la sencillez cuando no sea así.