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

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

CrewAI — фреймворк для построения мульти-агентных систем, где каждый агент имеет роль, цель и набор инструментов. Концепция «crew» (команды) делает архитектуру интуитивной: агенты работают как специалисты команды, а задачи делегируются по ролям. В отличие от 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 reviews
4. Strategy Analyst — синтез данных, выявление стратегических паттернов
5. Report Writer — финальный отчёт с executive summary

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

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

• Время подготовки отчёта: 3 недели → 4 часа автономной работы + 2 часа ревью аналитика
• Покрытие конкурентов: 5 → 12 компаний
• Глубина анализа: покрытие 23 аспектов vs 11 ранее
• Пропущенных значимых событий: аналитики оценили в 2–3 случая/квартал → 0

CrewAI vs LangGraph

Сроки

• Прототип CrewAI с 3 агентами: 2–4 дня
• Production-crew с кастомными инструментами: 1–2 недели
• Сложный Flow с conditional routing: 2–3 недели
• Интеграция с корпоративными системами: +1–2 недели