Używamy cookies, żeby zwiększyć Twoje doświadczenia na stronie
CrewAI - Framework Multi-Agent
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 drużyny specjalistów, gdzie każdy ma swoją rolę!
Instalacja i podstawy
1# pip install crewai crewai-tools
2
3from crewai import Agent, Task, Crew, Process
4from langchain_openai import ChatOpenAI
5
6# Model LLM
7llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)Tworzenie agentów
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)
12
13# Agent writer
14writer = Agent(
15 role="Content Writer",
16 goal="Tworzyć angażujące i edukacyjne treści na temat technologii",
17 backstory="""Jesteś utalentowanym pisarzem technicznym z pasją do
18 tłumaczenia złożonych koncepcji na prosty język. Twoje artykuły
19 są zawsze dobrze zorganizowane i przyjemne w czytaniu.""",
20 verbose=True,
21 allow_delegation=True,
22 llm=llm
23)
24
25# Agent editor
26editor = Agent(
27 role="Senior Editor",
28 goal="Zapewnić najwyższą jakość publikowanych treści",
29 backstory="""Jesteś doświadczonym redaktorem z okiem do szczegółów.
30 Dbasz o spójność stylu, poprawność faktyczną i klarowność przekazu.
31 Nie boisz się proponować zmian, które poprawią tekst.""",
32 verbose=True,
33 allow_delegation=False,
34 llm=llm
35)Definiowanie zadań
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)
15
16# Zadanie pisarskie
17writing_task = Task(
18 description="""Na podstawie raportu badawczego napisz artykuł blogowy.
19
20 Wymagania:
21 1. Wciągający wstęp
22 2. Jasna struktura z nagłówkami
23 3. Praktyczne przykłady
24 4. Podsumowanie z kluczowymi wnioskami
25 5. Długość: 1000-1500 słów
26
27 Używaj przystępnego języka dla czytelników technicznych.""",
28 expected_output="Kompletny artykuł blogowy gotowy do publikacji",
29 agent=writer,
30 context=[research_task] # Używa wyniku poprzedniego zadania
31)
32
33# Zadanie redakcyjne
34editing_task = Task(
35 description="""Zrecenzuj i popraw artykuł.
36
37 Sprawdź:
38 1. Poprawność gramatyczną i stylistyczną
39 2. Spójność argumentacji
40 3. Klarowność przekazu
41 4. Angażujący ton
42
43 Wprowadź niezbędne poprawki.""",
44 expected_output="Zredagowany, finalny artykuł",
45 agent=editor,
46 context=[writing_task]
47)Tworzenie i uruchamianie Crew
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)Narzędzia dla agentó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)
22
23# Własne narzędzie
24from crewai_tools import BaseTool
25from pydantic import BaseModel, Field
26
27class DatabaseQueryInput(BaseModel):
28 query: str = Field(description="SQL query to execute")
29
30class DatabaseQueryTool(BaseTool):
31 name: str = "Database Query"
32 description: str = "Wykonuje zapytania SQL na bazie danych"
33 args_schema: type[BaseModel] = DatabaseQueryInput
34
35 def _run(self, query: str) -> str:
36 # Implementacja zapytania
37 return f"Wynik zapytania: {query}"Hierarchiczny proces
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)Pamięć i kontekst
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)Callbacks i monitorowanie
1from crewai.callbacks import CrewCallback
2
3class ProgressCallback(CrewCallback):
4 """Callback do śledzenia postępu."""
5
6 def on_task_start(self, task: Task):
7 print(f"🚀 Rozpoczynam: {task.description[:50]}...")
8
9 def on_task_end(self, task: Task, output: str):
10 print(f"✅ Zakończono: {task.description[:50]}")
11 print(f" Output: {output[:100]}...")
12
13 def on_agent_action(self, agent: Agent, action: str):
14 print(f"🤖 {agent.role}: {action}")
15
16# Użycie
17crew = Crew(
18 agents=[researcher, writer],
19 tasks=[research_task, writing_task],
20 callbacks=[ProgressCallback()],
21 verbose=True
22)Asynchroniczne wykonanie
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())Przykład: Content Pipeline
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()
29
30# Uruchomienie
31article = create_content_pipeline("Machine Learning w 2025")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!