Samouczek Agent Swarm: koordynuj agentów AI z CrewAI

Czym jest Agent Swarm?

Agent swarm to grupa agentów AI współpracujących nad wykonaniem zadania. Zamiast prosić jeden model, by sam zbadał, napisał, zrecenzował i dopiął całość, rój dzieli pracę między wiele wyspecjalizowanych agentów. Każdy agent ma jasną rolę, a system koordynuje ich działania, by uzyskać lepszy efekt końcowy.

Przewagą jest specjalizacja. Agent nastawiony na research może skupić się na wyszukiwaniu i weryfikacji źródeł, agent piszący — na strukturze i klarowności, a recenzent sprawdzi szkic względem oryginalnych badań bez rozpraszania się pisaniem, które sam przed chwilą stworzył. 

Agent koordynujący spina kroki i decyduje, co dalej. W budowanym przez nas roju role te odpowiadają bezpośrednio Researcherowi, Writerowi, Reviewerowi i Managerowi.

To jednak nie jest za darmo. Więcej agentów oznacza więcej wywołań modelu, więcej czasu i więcej miejsc, w których przepływ pracy może się posypać; to kompromis, do którego wrócimy na końcu, gdy zobaczysz realne koszty uruchomienia.

Czym jest CrewAI?

CrewAI to otwartoźródłowy framework do budowy systemów wieloagentowych. Daje przejrzysty sposób definiowania agentów, przypisywania im zadań i wyboru sposobu koordynacji, dzięki czemu skupiasz się na rolach i przepływie pracy, a nie na łączeniu elementów.

Trzy pojęcia robią tu większość roboty. 

• Agenci to poszczególni „pracownicy”, każdy zdefiniowany przez rolę, cel i historię, które kształtują jego zachowanie. 
• Zadania opisują pracę do wykonania i oczekiwany rezultat każdego z nich. 
• „Załoga” łączy agentów i zadania, uruchamiając je w wybranym procesie

Proces może być sekwencyjny — zadania w stałej kolejności — albo hierarchiczny, gdzie agent‑menedżer deleguje pracę innym i weryfikuje ich wyniki. Tutaj użyjemy procesu hierarchicznego, z delegacją po stronie Managera.

CrewAI obsługuje też dostęp do modeli przez LiteLLM, otwartoźródłową bibliotekę zapewniającą ujednolicony interfejs dla wielu dostawców. Dzięki temu możemy zdefiniować pojedynczy model Gemini 3.5 Flash i wykorzystać go we wszystkich czterech agentach, co skonfigurujemy w kolejnych krokach.

Budowa roju agentów z CrewAI i Gemini 3.5 Flash

Zaczynajmy od przygotowania środowiska.

1. Wymagania wstępne

Zanim zaczniesz, upewnij się, że masz:

• Zainstalowany Python 3.11 lub nowszy
• Zainstalowany Jupyter Notebook lub JupyterLab
• Podstawową znajomość Pythona i notebooków
• Konto w Google AI Studio
• Włączone rozliczenia lub dodane kredyty na twoim koncie Google do użycia Gemini
• Darmowe konto Olostep

2. Zainstaluj zależności

Uruchom Jupyter Notebook lub JupyterLab i otwórz notebook z najnowszym jądrem Pythona dostępnym w twoim środowisku. Następnie uruchom poniższą komórkę, aby zainstalować wymagane pakiety.

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

Po zakończeniu instalacji uruchom następną komórkę, aby sprawdzić wersje zainstalowanych pakietów. To pomaga potwierdzić, że wszystko zainstalowało się poprawnie.

import importlib.metadata as md

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

Powinieneś zobaczyć wynik podobny do tego:

         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. Ustaw zmienne środowiskowe

Będziesz potrzebować dwóch kluczy API do tego projektu. 

Najpierw utwórz klucz API Gemini w Google AI Studio. Otwórz Google AI Studio, przejdź do sekcji API key i utwórz nowy klucz dla swojego projektu. Gemini oferuje darmowy poziom, ale z ograniczeniami użycia i limitami zapytań. Aby bardziej niezawodnie testować i uruchamiać Gemini, połącz konto rozliczeniowe Google lub dodaj kredyty na koncie Google, by rozliczać użycie po przekroczeniu limitów darmowego poziomu.

Następnie utwórz klucz API Olostep. Zarejestruj się w Olostep, otwórz pulpit Olostep i wygeneruj klucz API na stronie kluczy API. Użyjemy go, aby dać agentowi Researcher dostęp do sieci na żywo do wyszukiwania i scrapowania stron.

Gdy oba klucze będą gotowe, zapisz je jako zmienne środowiskowe. Gemini będzie zasilać agentów CrewAI, a Olostep pozwoli Researcherowi zbierać aktualne informacje z sieci.

Po utworzeniu obu kluczy zapisz je jako zmienne środowiskowe przed uruchomieniem notebooka.

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"

Teraz uruchom poniższą komórkę, aby sprawdzić, czy klucz API Gemini został poprawnie załadowany.

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

Powinieneś zobaczyć: 

Gemini API key loaded

4. Skonfiguruj Gemini 3.5 Flash przez LiteLLM

W tym projekcie użyjemy Gemini 3.5 Flash jako wspólnego LLM dla wszystkich agentów CrewAI. Gemini 3.5 Flash jest zaprojektowany do szybkich, agentowych i ukierunkowanych na kod zadań. Przydaje się tutaj, bo nasi agenci muszą rozumować wieloetapowo, wykonywać instrukcje, korzystać z narzędzi i tworzyć ustrukturyzowane wyniki.

Gemini 3.5 Flash obsługuje też duże okno kontekstu, długie wyjścia, myślenie i użycie narzędzi, co czyni go mocnym wyborem do przepływów pracy agentów. Ponieważ rój może wielokrotnie wywoływać model w ramach Researchera, Writera, Reviewera i Managera, użycie modelu Flash pomaga utrzymać responsywność, zachowując przy tym solidną jakość rozumowania i pisania.

Jeśli chcesz porównać model z innymi najnowszymi LLM, polecam nasze przewodniki: Gemini 3.5 Flash vs GPT-5.5 oraz Gemini 3.5 Flash vs Claude Opus 4.8.

Dostęp do Gemini uzyskamy przez LiteLLM. LiteLLM to otwartoźródłowa biblioteka z ujednoliconym interfejsem do wywoływania wielu dostawców LLM, w tym Gemini, OpenAI, Anthropic, Bedrock, Vertex AI i innych. W tej konfiguracji CrewAI używa LiteLLM pod spodem, więc możemy podać nazwę modelu Gemini w prostym formacie provider/model.

Teraz utworzymy współdzielony LLM CrewAI i wykorzystamy go we wszystkich agentach. To upraszcza konfigurację, bo Researcher, Writer, Reviewer i Manager będą korzystać z tego samego modelu 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)")

Powinieneś zobaczyć wynik podobny do tego:

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

To potwierdza, że CrewAI może teraz uzyskać dostęp do Gemini przez LiteLLM. W kolejnych krokach przekażemy ten współdzielony obiekt gemini_llm każdemu agentowi.

5. Dodaj dostęp do sieci na żywo

Agent Researcher potrzebuje dostępu do aktualnych informacji. W tym celu stworzymy niestandardowe narzędzie wyszukiwania Olostep i później przekażemy je Researcherowi.

To narzędzie może działać w dwóch trybach. Może zwracać tylko podsumowania wyników wyszukiwania — szybciej — albo scrapować strony i zwracać treść w markdown. Dodajemy też mały budżet wyszukiwań, aby agent nie wywoływał narzędzia sieciowego zbyt wiele razy.

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

Najpierw załaduj klucz API Olostep ze środowiska.

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

Teraz utwórz niestandardowe narzędzie 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)

Na koniec utwórz instancję narzędzia.

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

Powinieneś zobaczyć wynik podobny do tego:

Olostep API key loaded.
Tool ready: olostep_web_search

To potwierdza, że narzędzie wyszukiwania sieci Olostep jest gotowe. Później dołączymy je do agenta Researcher, aby mógł zbierać aktualne informacje z sieci.

6. Utwórz agentów

Teraz utworzymy agentów do naszego roju. Zdefiniujemy trzech „pracowników” i jednego menedżera. Agenci‑pracownicy wykonują główne zadania, a menedżer koordynuje przepływ pracy.

Agent Researcher użyje narzędzia Olostep do znalezienia aktualnych informacji. Agent Writer zamieni research w szkic. Agent Reviewer poprawi szkic. Agent Manager zdecyduje, jak delegować pracę w załodze hierarchicznej.

Utwórz agentów‑pracowników

Zacznijmy od trzech agentów‑pracowników.

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

Następnie utwórz agenta 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,
)

Teraz utwórz agenta 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,
)

Utwórz agenta‑menedżera

Na koniec utwórz agenta 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,
)

Manager jest konfigurowany osobno podczas budowy załogi hierarchicznej. Dlatego nie dodaliśmy go jeszcze do listy agentów‑pracowników.

7. Zdefiniuj zadania

Teraz zdefiniujemy zadania dla roju agentów. Te zadania nie są przypisane bezpośrednio do agentów‑pracowników. Zamiast tego agent Manager zdecyduje, który „pracownik” powinien obsłużyć każde zadanie.

W tym przepływie pracy pierwszym zadaniem jest research, drugim — pisanie, a trzecim — recenzja. Każde zadanie zawiera też oczekiwany rezultat, aby agenci dokładnie wiedzieli, co mają dostarczyć.

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

Następnie utwórz zadanie pisania. To zadanie używa researchu jako kontekstu, aby Writer mógł zbudować artykuł na podstawie zebranych materiałów.

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

Teraz utwórz zadanie recenzji. To zadanie korzysta z researchu i pisania jako kontekstu, aby Reviewer mógł sprawdzić artykuł względem oryginalnych materiałów.

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

Na koniec dodaj wszystkie trzy zadania do listy.

tasks = [research_task, write_task, review_task]

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

Powinieneś zobaczyć:

Defined 3 manager-delegated tasks.

8. Zbuduj przepływ pracy

Teraz połączymy agentów i zadania w jeden przepływ pracy CrewAI. Użyjemy procesu hierarchicznego, w którym agent Manager koordynuje agentów‑pracowników.

W tej konfiguracji Researcher, Writer i Reviewer są dodani jako agenci‑pracownicy. Manager jest przekazywany osobno przez manager_agent, bo odpowiada za delegację i koordynację.

from crewai import Crew, Process

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

Teraz wydrukuj szczegóły załogi, aby potwierdzić, że przepływ pracy został poprawnie złożony.

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

Powinieneś zobaczyć wynik podobny do tego:

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

To potwierdza, że hierarchiczny rój agentów jest gotowy. W kolejnym kroku uruchomimy załogę na podanym temacie i wygenerujemy finalną odpowiedź.

9. Uruchom rój agentów

Możemy już uruchomić rój agentów. Podamy temat i wystartujemy przepływ pracy za pomocą kickoff_async().

Jupyter już uruchamia pętlę zdarzeń w tle. Dlatego najpierw zastosujemy nest_asyncio, aby asynchroniczny przepływ CrewAI mógł działać w notebooku.

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

Powinieneś zobaczyć:

nest_asyncio applied: CrewAI can now run inside this kernel.

Następnie zdefiniuj temat i utwórz pomocniczą funkcję do uruchomienia załogi.

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)

Gdy przepływ ruszy, CrewAI najpierw przekazuje zadanie agentowi Manager. Manager deleguje research do Senior Research Analyst. Researcher używa narzędzia olostep_web_search, by przeszukać sieć pod kątem informacji o adaptacyjnych progach, użyciu narzędzi przez LLM, pamięci i decyzjach o wywołaniu narzędzia.

W logu Researcher używa zarówno wywołań tylko‑wyszukiwania, jak i scrapowania. Budżet wyszukiwań jest w pełni wykorzystany — 6 z 6 całkowitych wywołań sieci i 3 z 3 wywołań odkrywczych. Jedno dodatkowe żądanie tylko‑wyszukiwania jest zablokowane, bo limit wyszukiwań odkrywczych został osiągnięty. To pokazuje, że reguły budżetu w naszym niestandardowym narzędziu Olostep działają poprawnie.

Po etapie researchu Manager kontynuuje przepływ, delegując kroki pisania i recenzji. Agenci tworzą notatki badawcze, przygotowują szkic artykułu, a następnie szlifują finalną odpowiedź. Uruchomienie kończy się sukcesem i zwraca obiekt CrewOutput.

Powinieneś zobaczyć wynik podobny do tego:

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

To oznacza, że cały rój wykonał się pomyślnie. Manager koordynował przepływ, Researcher zebrał aktualne informacje, Writer stworzył szkic, a Reviewer poprawił wynik końcowy.

10. Zobacz wynik

Po zakończeniu działania roju możemy wyświetlić finalny artykuł wygenerowany przez załogę. Wartość result.raw zawiera końcową odpowiedź z przepływu pracy.

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

To wyrenderuje finalny wynik jako sformatowany markdown w notebooku. 

W większości przypadków wygenerowany blog będzie już dobrze ustrukturyzowany — z wieloma nagłówkami, klarownymi sekcjami, diagramami lub wizualnymi objaśnieniami tam, gdzie to uzasadnione — oraz mocnym podsumowaniem.

Połączenie researchu, pisania i redakcyjnej recenzji pomaga tworzyć treści gotowe do publikacji. Po przejrzeniu wyniku sam dałbym mu zielone światło do publikacji.

Następnie możemy wydrukować ślad wykonania. To pomaga zrozumieć, jak działał rój, którzy agenci uczestniczyli, ile było wywołań sieci i jak przetwarzano każde zadanie.

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

Powinieneś zobaczyć wynik podobny do tego:

Ten ślad pokazuje, że hierarchiczny rój zadziałał zgodnie z oczekiwaniami. Manager koordynował trzy zadania, Researcher obsłużył krok researchu, Writer stworzył szkic artykułu, a Reviewer poprawił finalną odpowiedź.

Widać tu jednak ważne ograniczenie: przepływy wieloagentowe mogą być kosztowne. 

W tym przebiegu użyto 6 zewnętrznych wywołań sieci i wielu wywołań LLM w ramach Managera, Researchera, Writera i Reviewera. Pojedynczy przebieg może kosztować około 0,28 USD, a stworzenie tego przewodnika kosztowało podczas testów około 2,78 USD. To sporo jak na prosty tutorial.

Końcowe przemyślenia

Roje agentów i przepływy wieloagentowe świetnie wyglądają na papierze. Pomysł, że Researcher, Writer, Reviewer i Manager pracują razem, brzmi mocno — i w tym przewodniku efekt końcowy faktycznie nadawał się do publikacji po drobnych poprawkach.

W praktyce jednak takie ustawienie może kosztować więcej, zajmować więcej czasu i tworzyć więcej punktów awarii. W moim przypadku jeden przebieg kosztuje około 0,28 USD, a przygotowanie tego przewodnika — około 2,78 USD. To dużo jak na prosty projekt tutorialowy.

W realnych zastosowaniach nie zawsze używałbym pełnego roju wieloagentowego. Wolałbym bardziej programistyczne podejście, gdzie proste zadania obsługuje kod lub reguły, a tylko złożone są przekazywane agentom. Koszty można też obniżyć, ograniczając wywołania narzędzi, używając mniejszej liczby agentów, skracając prompty i wybierając przepływ liniowy zamiast hierarchicznego.

Oczywiście, jeśli zoptymalizujemy za bardzo, to przestaje być „prawdziwy” rój agentów. Bardziej przypomina regułowy przepływ AI. Dlatego kluczowa jest równowaga: używaj rojów agentów, gdy realnie dodają wartość, a w przeciwnym razie trzymaj rzeczy proste.