CrewAI - Framework Multi-Agent

W poprzedniej lekcji sami napisaliśmy klasę agenta, orkiestratora i łańcuch wiadomości. Działało, ale każdą nową funkcję, taką jak narzędzia, pamięć czy delegowanie zadań, trzeba by dopisywać ręcznie. Gdy wyprawa rośnie, lepiej skorzystać z gotowego regulaminu obozu.

CrewAI to framework Pythona do tworzenia zespołów agentów AI, które współpracują przy rozwiązywaniu złożonych zadań. To jak montowanie załogi ekspedycji, gdzie każdy ma swoją rolę. Cały framework opiera się na trzech pojęciach: Agent (kto), Task (co) i Crew (jak razem).

Instalacja i podstawy

Instalujemy dwa pakiety i importujemy podstawowe klasy. Model językowy konfigurujemy klasą LLM z samego CrewAI:

1# pip install crewai crewai-tools
2
3from crewai import Agent, Task, Crew, Process, LLM
4
5# Model LLM
6llm = LLM(model="gpt-4o-mini", temperature=0.7)

Uwaga na zmianę w bibliotece: starsze przykłady w sieci używają tu ChatOpenAI z LangChain, ale aktualne wersje CrewAI nie zależą od LangChain i odrzucają taki obiekt przy walidacji agenta. LLM(model="gpt-4o-mini") korzysta pod spodem z klucza w zmiennej OPENAI_API_KEY.

Tworzenie agentów

Agent w CrewAI to trzy zdania o nim samym: rola (role), cel (goal) i historia (backstory). Model dostaje je jako opis postaci, w którą ma się wcielić. Oto badacz:

1# Agent researcher
2researcher = Agent(
3    role="Senior Research Analyst",
4    goal="Przeprowadzać dokładne badania i dostarczać wiarygodne informacje",
5    backstory="""Jesteś doświadczonym analitykiem z 15-letnim doświadczeniem
6    w badaniu trendów technologicznych. Specjalizujesz się w analizie AI i ML.
7    Zawsze weryfikujesz fakty i szukasz wiarygodnych źródeł.""",
8    verbose=True,
9    allow_delegation=False,
10    llm=llm
11)

verbose=True wypisuje w konsoli rozumowanie agenta, a allow_delegation=False zabrania mu przekazywać pracy innym. W aktualnych wersjach False to i tak wartość domyślna.

Pisarz i redaktor wyglądają podobnie, zmienia się tylko opis postaci:

1# Agent writer
2writer = Agent(
3    role="Content Writer",
4    goal="Tworzyć angażujące i edukacyjne treści na temat technologii",
5    backstory="""Jesteś utalentowanym pisarzem technicznym z pasją do
6    tłumaczenia złożonych koncepcji na prosty język. Twoje artykuły
7    są zawsze dobrze zorganizowane i przyjemne w czytaniu.""",
8    verbose=True,
9    allow_delegation=True,
10    llm=llm
11)
12
13# Agent editor
14editor = Agent(
15    role="Senior Editor",
16    goal="Zapewnić najwyższą jakość publikowanych treści",
17    backstory="""Jesteś doświadczonym redaktorem z okiem do szczegółów.
18    Dbasz o spójność stylu, poprawność faktyczną i klarowność przekazu.
19    Nie boisz się proponować zmian, które poprawią tekst.""",
20    verbose=True,
21    allow_delegation=False,
22    llm=llm
23)

Pisarz ma allow_delegation=True, więc gdy czegoś nie wie, może zapytać kolegę z załogi. To jedyna różnica techniczna, reszta to charakter zapisany w backstory.

Definiowanie zadań

Task opisuje pracę do wykonania. Każde zadanie ma opis, oczekiwany wynik (expected_output) i przypisanego agenta:

1# Zadanie badawcze
2research_task = Task(
3    description="""Przeprowadź dokładne badanie na temat: {topic}
4
5    Twoje zadanie:
6    1. Znajdź najnowsze informacje i trendy
7    2. Zidentyfikuj kluczowych graczy i technologie
8    3. Przeanalizuj potencjalne zastosowania
9    4. Oceń przyszłe kierunki rozwoju
10
11    Dostarcz szczegółowy raport z badań.""",
12    expected_output="Szczegółowy raport badawczy z cytatami i źródłami",
13    agent=researcher
14)

Zapis {topic} w opisie to miejsce na zmienną. CrewAI podstawi tu wartość z parametru inputs przy uruchomieniu zespołu.

Zadanie pisarskie korzysta z wyniku badań. Służy do tego parametr context:

1# Zadanie pisarskie
2writing_task = Task(
3    description="""Na podstawie raportu badawczego napisz artykuł blogowy.
4
5    Wymagania:
6    1. Wciągający wstęp
7    2. Jasna struktura z nagłówkami
8    3. Praktyczne przykłady
9    4. Podsumowanie z kluczowymi wnioskami
10    5. Długość: 1000-1500 słów
11
12    Używaj przystępnego języka dla czytelników technicznych.""",
13    expected_output="Kompletny artykuł blogowy gotowy do publikacji",
14    agent=writer,
15    context=[research_task]  # Używa wyniku poprzedniego zadania
16)

context=[research_task] przekazuje pisarzowi raport badacza, więc nie musimy ręcznie kleić tekstów, jak robił to nasz orkiestrator.

Redaktor dostaje w kontekście artykuł pisarza:

1# Zadanie redakcyjne
2editing_task = Task(
3    description="""Zrecenzuj i popraw artykuł.
4
5    Sprawdź:
6    1. Poprawność gramatyczną i stylistyczną
7    2. Spójność argumentacji
8    3. Klarowność przekazu
9    4. Angażujący ton
10
11    Wprowadź niezbędne poprawki.""",
12    expected_output="Zredagowany, finalny artykuł",
13    agent=editor,
14    context=[writing_task]
15)

Opis, oczekiwany wynik, agent i kontekst to cztery elementy, z których składasz każde zadanie.

Tworzenie i uruchamianie Crew

Crew łączy agentów i zadania. Parametr process decyduje, jak załoga pracuje:

1# Tworzenie zespołu
2content_crew = Crew(
3    agents=[researcher, writer, editor],
4    tasks=[research_task, writing_task, editing_task],
5    process=Process.sequential,  # lub Process.hierarchical
6    verbose=True
7)
8
9# Uruchomienie z inputami
10result = content_crew.kickoff(inputs={"topic": "RAG w enterprise AI"})
11print(result)

Process.sequential wykonuje zadania po kolei, w kolejności z listy. CrewAI wspiera dwa typy procesów: sequential i hierarchical. kickoff zwraca obiekt CrewOutput, którego tekst znajdziesz w polu raw, a print wypisze go od razu.

Narzędzia dla agentów

Agent bez narzędzi zna tylko to, czego nauczył się model. Pakiet crewai-tools daje gotowe narzędzia do wyszukiwarki, stron i plików:

1from crewai_tools import (
2    SerperDevTool,      # Wyszukiwanie Google
3    WebsiteSearchTool,  # Przeszukiwanie stron
4    FileReadTool,       # Czytanie plików
5    PDFSearchTool,      # Przeszukiwanie PDF
6)
7
8# Konfiguracja narzędzi
9search_tool = SerperDevTool()
10web_tool = WebsiteSearchTool()
11file_tool = FileReadTool()
12
13# Agent z narzędziami
14researcher_with_tools = Agent(
15    role="Research Analyst",
16    goal="Zbierać i analizować informacje z różnych źródeł",
17    backstory="Ekspert w wyszukiwaniu informacji...",
18    tools=[search_tool, web_tool, file_tool],
19    verbose=True,
20    llm=llm
21)

SerperDevTool wymaga klucza w zmiennej SERPER_API_KEY. Agent sam decyduje, kiedy sięgnąć po które narzędzie, na podstawie ich opisów.

Własne narzędzie to klasa dziedzicząca po BaseTool. Obecnie importujemy ją z crewai.tools, a nie z crewai_tools, jak w starszych przykładach:

1# Własne narzędzie
2from crewai.tools import BaseTool
3from pydantic import BaseModel, Field
4
5class DatabaseQueryInput(BaseModel):
6    query: str = Field(description="SQL query to execute")
7
8class DatabaseQueryTool(BaseTool):
9    name: str = "Database Query"
10    description: str = "Wykonuje zapytania SQL na bazie danych"
11    args_schema: type[BaseModel] = DatabaseQueryInput
12
13    def _run(self, query: str) -> str:
14        # Implementacja zapytania
15        return f"Wynik zapytania: {query}"

args_schema to model Pydantic opisujący argumenty, a metoda _run zawiera właściwą logikę. name i description czyta model językowy, więc pisz je jasno.

Hierarchiczny proces

W procesie hierarchicznym zadania rozdziela menedżer. Tworzymy go jak zwykłego agenta:

1# Manager koordynujący
2manager = Agent(
3    role="Project Manager",
4    goal="Koordynować pracę zespołu i zapewnić terminową realizację",
5    backstory="Doświadczony PM z umiejętnością zarządzania zespołami AI",
6    verbose=True,
7    llm=llm
8)
9
10# Crew z hierarchią
11hierarchical_crew = Crew(
12    agents=[researcher, writer, editor],
13    tasks=[research_task, writing_task, editing_task],
14    process=Process.hierarchical,
15    manager_agent=manager,  # lub manager_llm=llm
16    verbose=True
17)

Menedżer nie trafia na listę agents, tylko do manager_agent. Zamiast własnego menedżera możesz podać manager_llm, a CrewAI stworzy go sam. Jedno z dwóch jest wymagane w procesie hierarchicznym.

Pamięć i kontekst

Załoga może pamiętać wcześniejsze ustalenia. Pamięć włączamy flagą memory=True, a embedder wskazuje model embeddingów, którym CrewAI zapisuje wspomnienia:

1# Crew z pamięcią
2crew_with_memory = Crew(
3    agents=[researcher, writer],
4    tasks=[research_task, writing_task],
5    process=Process.sequential,
6    memory=True,  # Włącza pamięć
7    embedder={
8        "provider": "openai",
9        "config": {"model": "text-embedding-3-small"}
10    },
11    verbose=True
12)
13
14# Agent z pamięcią
15agent_with_memory = Agent(
16    role="Assistant",
17    goal="Pomagać użytkownikowi",
18    backstory="...",
19    memory=True,
20    llm=llm
21)

To te same embeddingi, które poznaliśmy dwie lekcje temu, tylko używane do przechowywania wspomnień załogi zamiast dokumentów.

Callbacks i monitorowanie

Przy dłuższych zadaniach chcesz wiedzieć, co się dzieje. Crew przyjmuje dwie funkcje zwrotne: task_callback po każdym zadaniu i step_callback po każdym kroku agenta:

1from crewai.tasks.task_output import TaskOutput
2
3def on_task_end(output: TaskOutput) -> None:
4    """Wywoływana po zakończeniu każdego zadania."""
5    print(f"Zakończono: {output.description[:50]}")
6    print(f"   Output: {output.raw[:100]}...")
7
8def on_step(step) -> None:
9    """Wywoływana po każdym kroku agenta."""
10    print(f"Krok agenta: {str(step)[:100]}")
11
12# Użycie
13crew = Crew(
14    agents=[researcher, writer],
15    tasks=[research_task, writing_task],
16    task_callback=on_task_end,
17    step_callback=on_step,
18    verbose=True
19)

TaskOutput niesie opis zadania i surowy wynik w polu raw. Nie szukaj klasy CrewCallback ani modułu crewai.callbacks: w CrewAI ich nie ma, funkcje zwrotne przekazujesz właśnie przez te dwa parametry.

Asynchroniczne wykonanie

Kilka niezależnych załóg możesz uruchomić jednocześnie. Służy do tego kickoff_async:

1import asyncio
2
3async def run_crew_async(topic: str) -> str:
4    """Asynchroniczne uruchomienie crew."""
5    crew = Crew(
6        agents=[researcher, writer],
7        tasks=[research_task, writing_task],
8        process=Process.sequential
9    )
10
11    result = await crew.kickoff_async(inputs={"topic": topic})
12    return result
13
14# Równoległe uruchomienie wielu crew
15async def run_multiple_crews():
16    topics = ["AI w medycynie", "AI w finansach", "AI w edukacji"]
17
18    tasks = [run_crew_async(topic) for topic in topics]
19    results = await asyncio.gather(*tasks)
20
21    return dict(zip(topics, results))
22
23results = asyncio.run(run_multiple_crews())

asyncio.gather czeka na wszystkie trzy załogi naraz, a dict(zip(...)) paruje tematy z wynikami. W produkcji twórz osobne obiekty agentów i zadań dla każdej załogi, bo tu wszystkie współdzielą te same research_task i writing_task.

Przykład: Content Pipeline

Na koniec składamy wszystko w jeden pipeline z dodatkowym specjalistą SEO:

1def create_content_pipeline(topic: str) -> str:
2    """Kompletny pipeline tworzenia treści."""
3
4    # Agenci
5    seo_analyst = Agent(
6        role="SEO Specialist",
7        goal="Optymalizować treści pod kątem wyszukiwarek",
8        backstory="Ekspert SEO z doświadczeniem w content marketingu",
9        tools=[SerperDevTool()],
10        llm=llm
11    )
12
13    # Zadania
14    seo_task = Task(
15        description=f"Przeanalizuj słowa kluczowe dla tematu: {topic}",
16        expected_output="Lista 10 słów kluczowych z ich potencjałem",
17        agent=seo_analyst
18    )
19
20    # Crew
21    pipeline = Crew(
22        agents=[seo_analyst, researcher, writer, editor],
23        tasks=[seo_task, research_task, writing_task, editing_task],
24        process=Process.sequential,
25        verbose=True
26    )
27
28    return pipeline.kickoff(inputs={"topic": topic})
29
30# Uruchomienie
31article = create_content_pipeline("Machine Learning w 2025")

Opis seo_task używa f-stringa, ale zadania badacza i pisarza wciąż mają {topic}, dlatego kickoff dostaje inputs. Bez tego agenci zobaczyliby dosłowny tekst "{topic}". Polecam zaczynać od procesu sequential, a hierarchię włączać dopiero przy większej załodze, bo menedżer to dodatkowe wywołania modelu.

CrewAI to potężny framework do budowania zespołów AI. W następnej lekcji poznasz produkcyjne systemy RAG - jak wdrażać te technologie w enterprise!

Zapamiętaj: w CrewAI agent to członek załogi, zadanie to jego przydział, a crew to cała ekspedycja.