Agent-Swarm-Tutorial: KI-Agenten mit CrewAI koordinieren

Was ist ein Agenten-Schwarm?

Ein Agenten-Schwarm ist eine Gruppe von KI-Agenten, die gemeinsam eine Aufgabe erledigen. Statt ein einzelnes Modell mit Recherche, Schreiben, Review und Finalisierung zu betrauen, teilt ein Schwarm die Arbeit auf mehrere spezialisierte Agenten auf. Jeder Agent hat eine klare Rolle, und das System koordiniert sie, um ein besseres Endergebnis zu erzielen.

Der Vorteil ist Spezialisierung. Ein Research-Agent kann sich auf das Finden und Verifizieren von Quellen konzentrieren, ein Schreib-Agent auf Struktur und Verständlichkeit, und ein Reviewer kann den Entwurf gegen die ursprüngliche Recherche prüfen, ohne vom eigenen Schreiben abgelenkt zu sein. 

Ein koordinierender Agent verbindet die Schritte und entscheidet über den nächsten Schritt. In unserem Schwarm entsprechen diese Rollen direkt Researcher, Writer, Reviewer und Manager.

Das ist allerdings nicht umsonst. Mehr Agenten bedeuten mehr Modellaufrufe, mehr Zeit und mehr potenzielle Fehlerstellen im Workflow — ein Trade-off, zu dem wir am Ende zurückkehren, wenn du gesehen hast, was der Betrieb tatsächlich kostet.

Was ist CrewAI?

CrewAI ist ein Open-Source-Framework zum Aufbau von Multi-Agenten-Systemen. Es bietet eine saubere Möglichkeit, Agenten zu definieren, ihnen Aufgaben zuzuweisen und die Koordination festzulegen, sodass du dich auf Rollen und Workflow konzentrieren kannst statt auf die technische Verkabelung.

Drei Konzepte leisten die Hauptarbeit. 

• Agents sind die einzelnen Arbeiter, jeweils definiert durch Rolle, Ziel und eine Hintergrundgeschichte, die ihr Verhalten prägt. 
• Tasks beschreiben die Arbeit und welches Output jeweils erwartet wird. 
• Eine Crew bringt Agenten und Aufgaben zusammen und führt sie in einem gewählten Prozess aus

Dieser Prozess kann sequentiell sein, in dem Tasks in fester Reihenfolge laufen, oder hierarchisch, in dem ein Manager-Agent Arbeit delegiert und Ergebnisse prüft. Wir nutzen hier den hierarchischen Prozess, der Manager übernimmt die Delegation.

CrewAI übernimmt außerdem den Modellzugang über LiteLLM, eine Open-Source-Bibliothek mit einer einheitlichen Schnittstelle für viele Anbieter. So können wir ein einziges Gemini 3.5 Flash-Modell definieren und über alle vier Agenten hinweg wiederverwenden, was wir in den folgenden Schritten einrichten.

Einen Agenten-Schwarm mit CrewAI und Gemini 3.5 Flash aufbauen

Los geht’s mit dem Setup.

1. Voraussetzungen

Bevor du startest, stelle sicher, dass du Folgendes hast:

• Python 3.11 oder neuer
• Jupyter Notebook oder JupyterLab
• Grundkenntnisse in Python und Notebooks
• Ein Google AI Studio-Konto
• Abrechnung aktiviert oder Guthaben in deinem Google-Konto für die Gemini-Nutzung
• Ein kostenloses Olostep-Konto

2. Abhängigkeiten installieren

Starte Jupyter Notebook oder JupyterLab und öffne ein Notebook mit dem neuesten verfügbaren Python-Kernel in deiner Umgebung. Führe dann die folgende Zelle aus, um die benötigten Pakete zu installieren.

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

Nachdem die Installation abgeschlossen ist, führe die nächste Zelle aus, um die installierten Paketversionen zu prüfen. So stellst du sicher, dass alles korrekt installiert wurde.

import importlib.metadata as md

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

Die Ausgabe sollte in etwa so aussehen:

         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. Umgebungsvariablen setzen

Für dieses Projekt benötigst du zwei API-Schlüssel. 

Erstelle zuerst einen Gemini-API-Schlüssel im Google AI Studio. Öffne Google AI Studio, gehe zum Bereich API key und erstelle einen neuen Schlüssel für dein Projekt. Gemini bietet einen Free-Tarif, dieser hat jedoch Nutzungs- und Rate-Limits. Um Gemini verlässlicher zu testen und auszuführen, verknüpfe dein Google-Abrechnungskonto oder füge Guthaben hinzu, damit Nutzung über die Free-Grenzen abgerechnet werden kann.

Erstelle als Nächstes einen Olostep-API-Schlüssel. Registriere dich bei Olostep, öffne das Olostep-Dashboard und generiere einen API-Schlüssel auf der Seite API keys. Wir nutzen diesen Schlüssel, um dem Researcher-Agenten für Suche und Pagescraping Live-Webzugang zu geben.

Sobald beide Schlüssel bereit sind, speichere sie als Umgebungsvariablen. Gemini versorgt die CrewAI-Agenten, während Olostep dem Researcher-Agenten erlaubt, aktuelle Webinformationen zu sammeln.

Nachdem du beide Schlüssel erstellt hast, speichere sie als Umgebungsvariablen, bevor du das Notebook ausführst.

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"

Führe nun die folgende Zelle aus, um zu prüfen, dass der Gemini-API-Schlüssel korrekt geladen wurde.

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

Du solltest Folgendes sehen: 

Gemini API key loaded

4. Gemini 3.5 Flash über LiteLLM konfigurieren

In diesem Projekt verwenden wir Gemini 3.5 Flash als gemeinsames LLM für alle CrewAI-Agenten. Gemini 3.5 Flash ist für schnelle, agentische und coding-orientierte Workloads ausgelegt. Es ist hier sinnvoll, weil unsere Agenten in mehreren Schritten überlegen, Anweisungen befolgen, mit Tools arbeiten und strukturierte Outputs erzeugen müssen.

Gemini 3.5 Flash unterstützt außerdem große Kontextfenster, lange Ausgaben, Thinking und Tool-Nutzung und ist damit gut für Agent-Workflows geeignet. Da ein Agenten-Schwarm das Modell über Researcher, Writer, Reviewer und Manager hinweg mehrfach aufruft, hilft ein Flash-Modell, die Reaktionszeit hoch zu halten und dennoch starke Argumentations- und Schreibqualität zu liefern.

Wenn du das Modell mit anderen State-of-the-Art-LLMs vergleichen möchtest, lies unsere Guides zu Gemini 3.5 Flash vs GPT-5.5 und Gemini 3.5 Flash vs Claude Opus 4.8.

Wir greifen über LiteLLM auf Gemini zu. LiteLLM ist eine Open-Source-Bibliothek mit einer einheitlichen Schnittstelle für viele LLM-Anbieter, darunter Gemini, OpenAI, Anthropic, Bedrock, Vertex AI und weitere. In diesem Setup nutzt CrewAI LiteLLM im Hintergrund, sodass wir den Gemini-Modellnamen schlicht im Format provider/model übergeben können.

Jetzt erstellen wir ein gemeinsames CrewAI-LLM und verwenden es für alle Agenten wieder. Das hält das Setup einfach, weil Researcher, Writer, Reviewer und Manager dasselbe Gemini-Modell nutzen.

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

Die Ausgabe sollte in etwa so aussehen:

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

Das bestätigt, dass CrewAI jetzt über LiteLLM auf Gemini zugreifen kann. Im nächsten Schritt übergeben wir dieses gemeinsame gemini_llm-Objekt an jeden Agenten.

5. Live-Webzugang hinzufügen

Der Researcher-Agent benötigt Zugriff auf aktuelle Informationen. Dazu erstellen wir ein benutzerdefiniertes Olostep-Suchtool und geben es später dem Researcher-Agenten.

Dieses Tool kann in zwei Modi arbeiten. Es kann nur Suchzusammenfassungen zurückgeben, was schneller ist, oder die Ergebnisseiten scrapen und Seiteninhalte in Markdown liefern. Außerdem setzen wir ein kleines Suchbudget, damit der Agent das Web-Tool nicht zu oft aufruft.

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

Lade zuerst den Olostep-API-Schlüssel aus der Umgebung.

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

Erstelle nun das benutzerdefinierte CrewAI-Tool.

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)

Erzeuge zum Schluss eine Instanz des Tools.

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

Die Ausgabe sollte in etwa so aussehen:

Olostep API key loaded.
Tool ready: olostep_web_search

Damit ist das Olostep-Websuche-Tool bereit. Später hängen wir es an den Researcher-Agenten, damit er aktuelle Informationen aus dem Web sammeln kann.

6. Die Agenten erstellen

Jetzt erstellen wir die Agenten für unseren Schwarm. Wir definieren drei Worker-Agenten und einen Manager-Agenten. Die Worker erledigen die Hauptaufgaben, der Manager koordiniert den Ablauf.

Der Researcher nutzt das Olostep-Tool für aktuelle Informationen. Der Writer verwandelt die Recherche in einen Entwurf. Der Reviewer verbessert den Entwurf. Der Manager entscheidet in der hierarchischen Crew über die Delegation.

Worker-Agenten erstellen

Starten wir mit den drei Worker-Agenten.

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

Als Nächstes den Writer-Agenten.

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

Jetzt den Reviewer-Agenten.

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

Manager-Agent erstellen

Zum Schluss den Manager-Agenten anlegen.

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

Der Manager wird separat konfiguriert, wenn wir die hierarchische Crew bauen. Deshalb wurde er noch nicht zur Worker-Liste hinzugefügt.

7. Aufgaben definieren

Jetzt definieren wir die Tasks für den Agenten-Schwarm. Diese Tasks werden nicht direkt den Workern zugewiesen. Stattdessen entscheidet der Manager-Agent, welcher Worker welche Aufgabe übernimmt.

In diesem Workflow ist die erste Aufgabe Recherche, die zweite Schreiben und die dritte Review. Jede Aufgabe enthält außerdem ein erwartetes Output, damit die Agenten genau wissen, was sie liefern sollen.

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

Als Nächstes die Schreibaufgabe. Diese Aufgabe nutzt die Recherche als Kontext, damit der Writer den Artikel aus den gesammelten Informationen erstellen kann.

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

Jetzt die Review-Aufgabe. Diese nutzt sowohl Recherche- als auch Schreib-Task als Kontext, damit der Reviewer den Artikel gegen die ursprüngliche Recherche prüfen kann.

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

Füge zum Schluss alle drei Aufgaben in einer Liste zusammen.

tasks = [research_task, write_task, review_task]

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

Du solltest sehen:

Defined 3 manager-delegated tasks.

8. Den Workflow bauen

Jetzt verbinden wir die Agenten und Aufgaben zu einem CrewAI-Workflow. Wir nutzen einen hierarchischen Prozess, in dem der Manager-Agent die Worker koordiniert.

In diesem Setup werden Researcher, Writer und Reviewer als Worker hinzugefügt. Der Manager wird separat über manager_agent übergeben, da er für Delegation und Koordination verantwortlich ist.

from crewai import Crew, Process

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

Drucke jetzt die Crew-Details, um zu bestätigen, dass der Workflow korrekt zusammengesetzt wurde.

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

Die Ausgabe sollte in etwa so aussehen:

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

Damit ist der hierarchische Agenten-Schwarm bereit. Im nächsten Schritt führen wir die Crew zu einem Thema aus und generieren die finale Antwort.

9. Den Agenten-Schwarm ausführen

Nun können wir den Agenten-Schwarm starten. Wir geben ein Thema vor und starten den Workflow mit kickoff_async().

Jupyter führt bereits eine Event-Loop im Hintergrund aus. Deshalb wenden wir zuerst nest_asyncio an, damit der asynchrone CrewAI-Workflow im Notebook laufen kann.

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

Du solltest sehen:

nest_asyncio applied: CrewAI can now run inside this kernel.

Definiere als Nächstes ein Thema und eine Helferfunktion zum Ausführen der 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)

Wenn der Workflow startet, erhält zuerst der Manager-Agent die Aufgabe. Der Manager delegiert die Recherchearbeit an den Senior Research Analyst. Der Researcher nutzt das Tool olostep_web_search, um das Web nach Informationen zu adaptiven Schwellenwerten, LLM-Toolnutzung, Memory und Tool-Call-Entscheidungen zu durchsuchen.

Im Log nutzt der Researcher sowohl reine Suchaufrufe als auch Scraping-Aufrufe. Das Suchbudget wird vollständig ausgeschöpft: 6 von 6 Webaufrufen und 3 von 3 Discovery-Aufrufen. Eine zusätzliche reine Suche wird blockiert, weil das Discovery-Limit bereits erreicht ist. Das zeigt, dass die Budgetregeln in unserem Olostep-Custom-Tool korrekt greifen.

Nach der Recherche setzt der Manager den Workflow fort, delegiert Schreiben und Review. Die Agenten erstellen Recherchenotizen, verfassen den Artikel und polieren anschließend die finale Antwort. Der Lauf endet erfolgreich und gibt ein CrewOutput-Objekt zurück.

Die Ausgabe sollte in etwa so aussehen:

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

Das bedeutet, der vollständige Schwarm lief erfolgreich. Der Manager koordinierte den Workflow, der Researcher sammelte aktuelle Informationen, der Writer erstellte den Entwurf und der Reviewer verbesserte das finale Ergebnis.

10. Ergebnis ansehen

Nachdem der Schwarm fertig ist, können wir den finalen Artikel anzeigen, den die Crew generiert hat. Der Wert result.raw enthält die finale Antwort aus dem 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))

Dadurch wird die finale Ausgabe im Notebook als formatiertes Markdown gerendert. 

In den meisten Fällen ist der generierte Blog bereits gut strukturiert, mit mehreren Überschriften, klaren Abschnitten, Diagrammen oder Visualisierungen wo sinnvoll und einem überzeugenden Fazit. 

Die Kombination aus Recherche, Schreiben und redaktioneller Prüfung liefert Inhalte, die publikationsreif sind. Nach Sichtung der Ausgabe würde ich persönlich grünes Licht für die Veröffentlichung geben.

Als Nächstes können wir eine Execution-Trace ausgeben. So verstehst du, wie der Schwarm gearbeitet hat, welche Agenten beteiligt waren, wie viele Webaufrufe genutzt wurden und wie jeder Task verarbeitet wurde.

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

Die Ausgabe sollte in etwa so aussehen:

Der Trace zeigt, dass der hierarchische Schwarm wie erwartet gearbeitet hat. Der Manager koordinierte alle drei Aufgaben, der Researcher übernahm die Recherche, der Writer erstellte den Entwurf und der Reviewer verbesserte die finale Antwort.

Er zeigt aber auch eine wichtige Einschränkung: Multi-Agenten-Workflows können teuer werden. 

In diesem Lauf wurden 6 externe Webaufrufe und mehrere LLM-Aufrufe über Manager, Researcher, Writer und Reviewer genutzt. Ein einzelner Lauf kann bis zu ca. 0,28 $ kosten, und die Erstellung dieses Guides kostete in den Tests rund 2,78 $. Das ist viel für ein einfaches Tutorial-Projekt.

Fazit

Agenten-Schwärme und Multi-Agenten-Workflows klingen auf dem Papier großartig. Die Idee, dass Researcher, Writer, Reviewer und Manager zusammenarbeiten, hat Power — und in diesem Guide war das Endergebnis tatsächlich mit kleinen Anpassungen publikationsreif.

In der Praxis können Setup, Zeitbedarf und Fehlerrisiko jedoch steigen. In meinem Fall kostet ein Lauf etwa 0,28 $, und die Erstellung dieses Guides rund 2,78 $. Das ist für ein simples Tutorial relativ viel.

Für echte Anwendungen würde ich nicht immer einen vollständigen Multi-Agenten-Schwarm einsetzen. Ich bevorzuge ein stärker programmatisches Setup, in dem einfache Aufgaben durch Code oder Regeln erledigt werden und nur komplexe Schritte an Agenten gehen. Außerdem können wir Kosten senken, indem wir Toolaufrufe begrenzen, weniger Agenten nutzen, Prompts kürzen und einen linearen statt hierarchischen Workflow wählen.

Klar, wenn wir zu stark optimieren, ist es kein echter Agenten-Schwarm mehr, sondern eher ein regelbasierter KI-Workflow. Ziel ist daher Balance: Setze Agenten-Schwärme ein, wenn sie echten Mehrwert bringen, und halte es einfach, wenn nicht.