Інтеграція CrewAI для мульти-агентних систем

Інтеграція CrewAI для мульти-агентних систем

CrewAI — це фреймворк для побудови мульти-агентних систем, де кожен агент має роль, мету та набір інструментів. Концепція "команди" робить архітектуру інтуїтивною: агенти працюють як спеціалісти команди, а завдання розподіляються за ролями. На відміну від LangGraph з його явним описом переходів графа, CrewAI використує декларативний підхід з автоматичним керуванням потоком.

Базова структура: Агенти та завдання

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool, FileWriterTool
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

# Визначити агентів
researcher = Agent(
    role="Senior Research Analyst",
    goal="Знайти актуальну та точну інформацію на задану тему",
    backstory="""Ви — дослідник-аналітик з 10 років досвіду.
    Спеціалізуєтесь у технологічному секторі.
    Завжди перевіряєте джерела та вказуєте на суперечності.""",
    tools=[SerperDevTool(), ScrapeWebsiteTool()],
    llm=llm,
    verbose=True,
    max_iter=5,  # Максимум ітерацій агента
    memory=True,
)

writer = Agent(
    role="Content Strategist",
    goal="Створити структурований аналітичний звіт",
    backstory="Досвідчений технічний письменник, спеціаліст у бізнес-аналітиці.",
    tools=[FileWriterTool()],
    llm=llm,
    verbose=True,
)

# Визначити завдання
research_task = Task(
    description="""Дослідіть ринок {topic} на 2025 рік.
    Охопліть: ключові гравці, обсяг ринку, тренди, прогнози.
    Знайдіть мінімум 5 актуальних джерел.""",
    expected_output="Структуровані дослідницькі дані з джерелами",
    agent=researcher,
    async_execution=False,
)

write_task = Task(
    description="""На основі наданого дослідження створіть аналітичний звіт.
    Формат: вступ, ключові висновки (таблиця), тренди, висновки.
    Обсяг: 1500–2000 слів.""",
    expected_output="Готовий аналітичний звіт у форматі markdown",
    agent=writer,
    context=[research_task],  # Використовує вихід research_task як контекст
    output_file="report.md",
)

# Створити команду
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # Послідовне виконання
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "ринок LLM-рішень для корпоративного сектора"})

Hierarchical Process: Менеджер координує агентів

manager_llm = ChatOpenAI(model="gpt-4o", temperature=0)

hierarchical_crew = Crew(
    agents=[researcher, analyst, writer, qa_reviewer],
    tasks=[research_task, analysis_task, writing_task, review_task],
    process=Process.hierarchical,
    manager_llm=manager_llm,  # LLM-менеджер приймає рішення про делегування
    verbose=True,
)

У ієрархічному режимі менеджер автоматично вирішує, якому агенту делегувати завдання та чи потребує результат переробки.

CrewAI Flows: Імпіративне керування

from crewai.flow.flow import Flow, listen, start, router
from pydantic import BaseModel

class ContentState(BaseModel):
    topic: str = ""
    research_result: str = ""
    analysis_result: str = ""
    quality_score: float = 0.0
    final_content: str = ""

class ContentCreationFlow(Flow[ContentState]):

    @start()
    def initialize(self):
        print(f"Початок роботи над темою: {self.state.topic}")

    @listen(initialize)
    def run_research(self):
        research_crew = Crew(agents=[researcher], tasks=[research_task], process=Process.sequential)
        result = research_crew.kickoff(inputs={"topic": self.state.topic})
        self.state.research_result = result.raw

    @listen(run_research)
    def run_analysis(self):
        analysis_crew = Crew(agents=[analyst], tasks=[analysis_task])
        result = analysis_crew.kickoff(inputs={"research": self.state.research_result})
        self.state.analysis_result = result.raw

    @router(run_analysis)
    def check_quality(self):
        # Оцініть якість
        score = evaluate_quality(self.state.analysis_result)
        self.state.quality_score = score
        if score >= 0.8:
            return "write_content"
        return "improve_analysis"

    @listen("improve_analysis")
    def improve_analysis(self):
        # Повторний аналіз з додатковим контекстом
        pass

    @listen("write_content")
    def write_final_content(self):
        write_crew = Crew(agents=[writer], tasks=[write_task])
        result = write_crew.kickoff(inputs={"analysis": self.state.analysis_result})
        self.state.final_content = result.raw

flow = ContentCreationFlow()
flow.kickoff(inputs={"topic": "Застосування AI в логістиці"})

Користувацькі інструменти

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class DatabaseQueryInput(BaseModel):
    sql_query: str = Field(description="SQL-запит для виконання")
    database: str = Field(description="Ім'я бази даних", default="analytics")

class DatabaseQueryTool(BaseTool):
    name: str = "query_database"
    description: str = "Виконати SQL-запит до бази даних аналітики"
    args_schema: type[BaseModel] = DatabaseQueryInput

    def _run(self, sql_query: str, database: str = "analytics") -> str:
        # Валідація: лише SELECT
        if not sql_query.strip().upper().startswith("SELECT"):
            return "Помилка: дозволені лише SELECT-запити"
        result = db.execute(sql_query, database=database)
        return result.to_json()

db_tool = DatabaseQueryTool()
analyst.tools.append(db_tool)

Практичний приклад: Автоматизація конкурентного аналізу

Завдання: квартальний конкурентний аналіз для відділу стратегії. Раніше займав 3 тижні силами 2 аналітиків.

Команда CrewAI:

1. Scout Agent — моніторинг новин конкурентів, прес-релізи, вакансії (SerperDevTool + ScrapeWebsiteTool)
2. Financial Agent — аналіз публічної фінансової звітності (користувацький інструмент парсингу SEC/МСБО)
3. Product Agent — аналіз змін продукту, changelog, рецензії app store
4. Strategy Analyst — синтез даних, виявлення стратегічних паттернів
5. Report Writer — фінальний звіт з executive summary

Конфігурація: Process.hierarchical з manager_llm, завдання виконуються частково паралельно (Scout, Financial, Product — одночасно; Strategy Analyst чекає на всі три).

Результати за квартал:

• Час підготовки звіту: 3 тижні → 4 години автономної роботи + 2 години огляду аналітика
• Охоплення конкурентів: 5 → 12 компаній
• Глибина аналізу: 23 аспекти порівняно з 11 раніше
• Пропущені значимі события: аналітики оцінили 2–3 на квартал → 0

CrewAI vs LangGraph

Графік реалізації

• Прототип CrewAI з 3 агентами: 2–4 дні
• Production crew з користувацькими інструментами: 1–2 тижні
• Складний Flow з условною маршрутизацією: 2–3 тижні
• Інтеграція з корпоративними системами: +1–2 тижні