- BrainTools - https://www.braintools.ru -
В 2026 году каждый второй стартап обещает заменить команду разработчиков роем AI-агентов. Звучит как мечта уставшего тимлида: один агент пишет код, второй ревьюит, третий деплоит, четвертый отвечает на вопросы в Slack, а пятый, наверное, уже сам заказывает пиццу в офис. Никаких больничных, никаких «я не успеваю», только железная продуктивность 24/7.
Я тоже купился. Взял CrewAI, собрал команду из трёх агентов для анализа конкурентов и генерации отчётов. Демо отработало идеально: агенты обменялись парой сообщений, выдали связный Markdown-файл и даже отправили его в Telegram. «Ну всё, — подумал я, — теперь можно увольнять аналитиков и копирайтеров. Будущее наступило».
Ровно через четыре часа после запуска на реальной задаче я наблюдал картину, достойную сюрреалистического полотна: пять AI-агентов устроили бесконечный митинг в духе худших корпоративных созвонов. Они перебивали друг друга, уточняли уже уточнённое, ходили по кругу и, кажется, начали обсуждать погоду. Один агент назначил себя лидом и раздавал указания, которые остальные игнорировали. Другой пытался писать в файл, который в этот момент читал третий. Спустя 127 вызовов LLM и сожжённые $4.30 на API-ключах я остановил этот цирк вручную.
В этой статье я расскажу, почему готовые мультиагентные фреймворки превращают вашу задачу в хаос, как мы построили систему, которая действительно работает, и в каких случаях проще вообще не связываться с мультиагентностью. Спойлер: LLM — не главная проблема. Проблема — в архитектуре оркестрации, которую многие принимают за магию.
Постановка задачи была типичной для внутреннего продукта: нужно проанализировать трёх конкурентов по заданным параметрам (цены, фичи, маркетинговые каналы), сформировать сводный отчёт в формате Markdown и отправить его в Telegram-чат команды. Звучит как идеальный кейс для мультиагентной системы: один агент ищет информацию, второй её структурирует, третий пишет человекочитаемый текст.
CrewAI обещает именно это: определяешь агентов с ролями, целями и бэкстори, задаёшь задачи и запускаешь последовательный процесс. Код получается настолько простым, что я сначала не поверил:
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from crewai_tools import SerperDevTool, FileReadTool, FileWriteTool
# Инициализация LLM и инструментов
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.2)
search_tool = SerperDevTool()
file_read = FileReadTool()
file_write = FileWriteTool()
# Агент-исследователь: ищет информацию о конкурентах в интернете
researcher = Agent(
role="Senior Market Researcher",
goal="Найти актуальную информацию о конкурентах: цены, ключевые фичи, каналы продвижения",
backstory="Вы опытный аналитик рынка с 10-летним стажем. Вы умеете находить даже скрытую информацию.",
tools=[search_tool],
llm=llm,
verbose=True
)
# Агент-аналитик: обрабатывает сырые данные, делает выводы
analyst = Agent(
role="Competitive Intelligence Analyst",
goal="Проанализировать собранные данные, выявить сильные и слабые стороны конкурентов, составить SWOT",
backstory="Вы бывший консультант McKinsey. Ваши отчёты всегда структурированы и содержат полезную информацию.",
tools=[file_read],
llm=llm,
verbose=True
)
# Агент-писатель: формирует финальный отчёт в Markdown
writer = Agent(
role="Technical Writer",
goal="Написать подробный отчёт в формате Markdown с чёткими выводами и рекомендациями",
backstory="Вы пишете документацию и аналитические отчёты для C-level аудитории.",
tools=[file_write],
llm=llm,
verbose=True
)
# Задачи
task_research = Task(
description="Найди информацию о конкурентах: Notion, Coda, Anytype. Интересуют цены, ключевые возможности, отзывы пользователей.",
expected_output="Структурированный документ с данными по каждому конкуренту.",
agent=researcher,
output_file="research_data.txt"
)
task_analysis = Task(
description="Проанализируй данные из файла research_data.txt. Составь сравнительную таблицу и SWOT-анализ.",
expected_output="Аналитическая записка с таблицей и выводами.",
agent=analyst,
output_file="analysis.txt"
)
task_write = Task(
description="На основе analysis.txt напиши итоговый отчёт в Markdown. Отправь его в Telegram (используй инструмент отправки).",
expected_output="Готовый Markdown-отчёт, отправленный в Telegram.",
agent=writer
)
# Запуск Crew с последовательным процессом
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[task_research, task_analysis, task_write],
process=Process.sequential,
verbose=True
)
result = crew.kickoff()
print("Работа завершена!")
Запустил — и магия случилась. В консоли замелькали разноцветные логи: агенты обмениваются сообщениями, исследователь что-то гуглит, аналитик читает файл, писатель формирует Markdown. Через пару минут в Telegram упало сообщение с красиво оформленным отчётом. Я был счастлив. Ровно 15 минут.
Потому что следующий запуск был уже на реальной задаче с нечёткими критериями и требованием параллельной работы. И вот тут началось.
Реальная задача отличалась от демо тремя критическими аспектами:
Неопределённость входных данных. Пользователь мог запросить анализ по произвольному списку конкурентов, иногда с дополнительными требованиями («сравни только enterprise-тарифы», «учти последние новости за март»).
Параллельная работа. Нужно было одновременно анализировать трёх конкурентов, а не последовательно, чтобы уложиться в разумное время.
Валидация результата. Перед отправкой в Telegram отчёт должен был пройти проверку на соответствие формату и отсутствие галлюцинаций.
Я перевёл процесс на Process.hierarchical в CrewAI (в надежде, что менеджер-агент всё разрулит) и добавил четвёртого агента-валидатора. И вот какие симптомы проявились практически сразу.
Агент-исследователь находил информацию, но аналитик начинал переспрашивать: «А точно ли эти цены актуальны? А где данные по фиче X?» Исследователь снова шёл в поиск, находил чуть больше, аналитик снова уточнял… Цикл повторялся, пока я не прервал выполнение на 37-й итерации. В логах это выглядело как диалог двух стажёров, которые боятся взять на себя ответственность:
[Researcher] -> [Analyst]: Я нашёл цены Notion: $8, $15, enterprise custom.
[Analyst] -> [Researcher]: Спасибо. А можешь уточнить, что входит в enterprise?
[Researcher] -> [Analyst]: Информации в открытых источниках нет.
[Analyst] -> [Researcher]: Может, поищешь на форумах?
[Researcher] -> [Analyst]: Нашёл упоминание, что enterprise включает SSO. Это добавить?
[Analyst] -> [Researcher]: Да, и ещё проверь, есть ли аудит логов. …
Проблема здесь в том, что агенты не имели чёткого критерия завершённости задачи. Они просто «общались», пока не упирались в лимит токенов или моё терпение.
В иерархическом режиме CrewAI назначает одного агента менеджером. В моём случае менеджером стал аналитик, который начал раздавать указания в стиле «Исследователь, срочно найди данные по Coda! Писатель, не пиши пока, жди!». Исследователь отвечал «Понял, выполняю», но продолжал гуглить Notion. Писатель и вовсе проигнорировал менеджера и начал генерировать отчёт на основе неполных данных.
Причина: в CrewAI менеджер не имеет реальных рычагов управления. Он лишь генерирует текст, который другие агенты могут интерпретировать как угодно. Это не оркестрация, это имитация совещания, где каждый слышит только себя.
Я добавил файловый инструмент, чтобы агенты могли сохранять промежуточные результаты. И тут же получил классическую гонку за файл:
Исследователь записывает данные в research_data.txt.
Аналитик начинает читать файл.
В этот момент писатель (который не должен был запускаться, но запустился из-за бага в оркестрации) пытается записать в тот же файл черновик отчёта.
Результат: файл повреждён, аналитик падает с ошибкой [1] парсинга.
В многопоточном программировании эту проблему решили ещё в 70-х семафорами и мьютексами. В мире AI-агентов про это, кажется, забыли.
Когда агентов стало пять (добавились валидатор и отправитель), контекст каждого агента раздулся до невообразимых размеров. CrewAI по умолчанию передаёт агенту всю историю сообщений, включая реплики других агентов, не относящиеся к его задаче. На пятой итерации исследователь начал «забывать», что он уже нашёл, и повторно гуглил одно и то же. Писатель вставлял в отчёт куски из случайных реплик менеджера, потому что они попали в его контекстное окно.
В документации CrewAI коммуникация агентов рисуется как аккуратная звезда или последовательная цепочка. В реальности мой граф сообщений выглядел перекати поле.
А с учётом того, что каждый агент мог отправить сообщение любому другому в любой момент, это превращалось в полносвязный граф, где количество рёбер растёт квадратично. Комбинаторный взрыв сообщений — вот что убивает производительность и бюджет.
Главный вывод этой главы: проблема не в LLM. GPT-4 отлично справляется с ролью отдельного агента. Проблема в архитектуре оркестрации, которая предполагает, что агенты сами договорятся.
Давайте честно разберём, почему CrewAI и AutoGen, прекрасно работающие на демо, ломаются на реальных задачах.
CrewAI предлагает два режима: Process.sequential и Process.hierarchical. Первый просто выполняет задачи одну за другой. Это надёжно, но не решает задачи с параллелизмом или условной логикой [2]. Как только вам нужно сказать «если анализ показал, что данных недостаточно, вернись к исследователю», вы выпадаете из парадигмы.
Разработчики предлагают использовать Tools для реализации условных переходов. То есть агент должен сам вызвать инструмент, который изменит состояние системы. На практике это приводит к монструозным промптам и костылям вроде такого:
# Костыль для CrewAI, чтобы реализовать условный возврат
from crewai import Agent, Task
from langchain.tools import tool
@tool
def request_more_research(topic: str) -> str:
"""
Вызови этот инструмент, если данных недостаточно.
ВНИМАНИЕ: это изменит порядок выполнения задач! (нет, не изменит)
"""
# В реальности мы просто пишем в глобальную переменную и надеемся,
# что внешний цикл её прочитает и перезапустит задачу.
global NEED_RESEARCH
NEED_RESEARCH = True
return "Запрос на дополнительное исследование зарегистрирован."
# Внешний цикл-костыль
while True:
result = crew.kickoff()
if not NEED_RESEARCH:
break
# Ручной сброс и повторный запуск с новыми параметрами...
Это не архитектура, это заклинания. И они нестабильны.
AutoGen от Microsoft предлагает более гибкую модель через GroupChat. Вы можете определить произвольный граф переходов между агентами с помощью speaker_selection_method. Звучит мощно. Но на практике:
По умолчанию используется auto — LLM решает, кто говорит следующим. Это порождает те самые бесконечные дебаты.
Жёсткие правила (round_robin, manual) требуют написания кастомной логики на Python, что возвращает нас к вопросу: «А зачем тогда фреймворк?»
Контекст опять же передаётся всем участникам, раздувая стоимость каждого вызова.
Ключевая архитектурная проблема обоих фреймворков — отсутствие явного контроллера состояния. Агенты работают по принципу «поговорим и решим», в то время как надёжная система требует «перейди из состояния А в состояние Б только при условии В, иначе в состояние С».
В традиционном программировании мы бы никогда не доверили бизнес-логику чату. Мы пишем конечные автоматы, workflow-движки, Sagas. Но в мире AI-агентов многие решили, что LLM сама разберётся. Не разберётся. LLM галлюцинирует, забывает [3], уходит в сторону. Это нормально для генерации текста, но катастрофично для оркестрации.
После нескольких недель боли [4] и сожжённых API-ключей мы переписали систему на LangGraph. Почему он? Потому что LangGraph изначально построен вокруг концепции направленного графа состояний, а не чата. Он заставляет думать в терминах узлов, рёбер и условий перехода — ровно то, что нужно для детерминированной оркестрации недетерминированных LLM-компонентов.
Перестаньте думать об агентах как о людях в Slack. Думайте о них как о микросервисах, которые вызываются по расписанию, получают строго ограниченный контекст и возвращают результат. А управляет всем оркестратор — граф, написанный на Python.
Мы разбили процесс на следующие узлы:
Planner — один раз анализирует входной запрос и формирует план работ (список шагов). Не участвует в дальнейшей дискуссии.
Workers — агенты, выполняющие конкретные задачи. Каждый worker получает только свой кусок состояния: описание задачи и результаты предыдущего шага. Никакой истории переписки.
Judge — проверяет результат worker’а. Принимает решение: перейти к следующему шагу, отправить на доработку или завершить с ошибкой.
Условия перехода — чистые Python-функции, проверяющие состояние.
Вот как это выглядит в коде на LangGraph:
from typing import TypedDict, List, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
# Определяем структуру состояния всего процесса
class AgentState(TypedDict):
input_query: str # Исходный запрос пользователя
plan: List[str] # План шагов от Planner'а
current_step: int # Индекс текущего шага (0..N)
step_results: dict # Результаты выполнения каждого шага
retry_count: int # Счётчик повторных попыток для текущего шага
final_report: str # Итоговый отчёт
error: str # Ошибка, если что-то пошло не так
# Инициализация модели
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.2)
# ========== Узел Planner ==========
def planner_node(state: AgentState) -> AgentState:
"""
На основе входного запроса формирует план шагов.
Выполняется ОДИН раз в начале.
"""
prompt = f"""
Пользователь запросил: {state['input_query']}
Составь план выполнения из последовательных шагов.
Каждый шаг должен быть атомарной задачей для AI-агента.
Верни список шагов в формате JSON: ["шаг 1", "шаг 2", ...]
"""
response = llm.invoke([HumanMessage(content=prompt)])
# Парсим ответ
import json
try:
plan = json.loads(response.content)
except:
plan = ["Собрать данные", "Проанализировать", "Написать отчёт"]
state['plan'] = plan
state['current_step'] = 0
state['step_results'] = {}
state['retry_count'] = 0
return state
# ========== Узел Worker (агент-исполнитель) ==========
def worker_node(state: AgentState) -> AgentState:
"""
Выполняет текущий шаг из плана.
ALERT: получает ТОЛЬКО описание шага и релевантные результаты предыдущих шагов.
"""
current_task = state['plan'][state['current_step']]
# Собираем контекст: только результаты ПРЕДЫДУЩИХ шагов, не всю историю!
context = ""
for step_idx, result in state['step_results'].items():
if int(step_idx) < state['current_step']:
context += f"nРезультат шага {step_idx}: {result}n"
prompt = f"""
Твоя задача: {current_task}
Контекст (результаты предыдущих шагов):
{context}
Исходный запрос пользователя: {state['input_query']}
Выполни задачу. Верни результат в виде структурированного текста.
"""
response = llm.invoke([HumanMessage(content=prompt)])
result = response.content
# Сохраняем результат текущего шага
state['step_results'][str(state['current_step'])] = result
return state
# ========== Узел Judge (валидатор) ==========
def judge_node(state: AgentState) -> AgentState:
"""
Проверяет результат текущего шага.
Не модифицирует состояние, только выставляет внутренние флаги для маршрутизации.
"""
# В реальном коде здесь может быть вызов LLM для валидации
# или проверка формата через regex/JSON schema
current_result = state['step_results'].get(str(state['current_step']), "")
# Простейшая эвристика: если результат слишком короткий или содержит "не знаю"
if len(current_result) < 50 or "не знаю" in current_result.lower():
state['error'] = "Результат невалиден"
else:
state['error'] = ""
return state
# ========== Функции маршрутизации ==========
def route_after_judge(state: AgentState) -> Literal["retry", "next", "finish"]:
"""
Решает, куда идти после проверки.
"""
MAX_RETRIES = 2
if state['error']:
if state['retry_count'] < MAX_RETRIES:
state['retry_count'] += 1
return "retry" # Повторяем текущий шаг
else:
return "finish" # Превышено число попыток, завершаем с ошибкой
# Если шаг выполнен успешно
state['retry_count'] = 0 # сбрасываем счётчик
if state['current_step'] < len(state['plan']) - 1:
state['current_step'] += 1
return "next" # Переходим к следующему шагу
else:
return "finish" # Все шаги выполнены
def route_after_planner(state: AgentState) -> Literal["work", "finish"]:
"""После планирования либо идём работать, либо завершаем (если план пуст)"""
if state['plan']:
return "work"
return "finish"
# ========== Сборка графа ==========
workflow = StateGraph(AgentState)
# Добавляем узлы
workflow.add_node("planner", planner_node)
workflow.add_node("worker", worker_node)
workflow.add_node("judge", judge_node)
# Устанавливаем точку входа
workflow.set_entry_point("planner")
# Добавляем рёбра с условиями
workflow.add_conditional_edges(
"planner",
route_after_planner,
{
"work": "worker",
"finish": END
}
)
workflow.add_edge("worker", "judge") # после worker всегда идём к judge
workflow.add_conditional_edges(
"judge",
route_after_judge,
{
"retry": "worker", # возврат на доработку
"next": "worker", # следующий шаг (тот же узел, но с обновлённым current_step)
"finish": END
}
)
# Компилируем граф
app = workflow.compile()
# ========== Запуск ==========
initial_state: AgentState = {
"input_query": "Проанализируй конкурентов Notion, Coda, Anytype. Нужен отчёт с ценами и фичами.",
"plan": [],
"current_step": 0,
"step_results": {},
"retry_count": 0,
"final_report": "",
"error": ""
}
# Выполнение с таймаутом
final_state = app.invoke(initial_state)
print("Финальный отчёт:", final_state['step_results'].get(str(len(final_state['plan'])-1)))
Контекст строго ограничен. Worker видит только результаты предыдущих шагов, а не всю историю переписки. Это решает проблему раздувания промпта и потери фокуса.
Детерминированные переходы. Решения принимает Python-код, а не LLM. route_after_judge — чистая функция, которая не галлюцинирует.
Встроенный стоп-кран. Счётчик retry_count гарантирует, что агент не уйдёт в бесконечный цикл уточнений.
Параллелизм легко добавить. LangGraph поддерживает параллельные ветки. Можно запустить трёх исследователей конкурентов одновременно и дождаться всех результатов перед анализом.
Вот как добавить параллельное выполнение для трёх конкурентов:
from langgraph.graph import StateGraph, END
from langgraph.types import Send
# Модифицируем состояние: добавляем список конкурентов
class ParallelAgentState(TypedDict):
competitors: List[str] # ["Notion", "Coda", "Anytype"]
research_results: dict # {"Notion": "...", "Coda": "..."}
input_query: str # Исходный запрос пользователя
plan: List[str] # План шагов от Planner'а
current_step: int # Индекс текущего шага (0..N)
step_results: dict # Результаты выполнения каждого шага
retry_count: int # Счётчик повторных попыток для текущего шага
final_report: str # Итоговый отчёт
error: str # Ошибка, если что-то пошло не так
def continue_to_research(state: ParallelAgentState):
"""
Возвращает список Send-объектов — по одному на каждого конкурента.
Это заставляет LangGraph запустить узел "researcher" параллельно для каждого.
"""
return [
Send("researcher", {"competitor": comp})
for comp in state['competitors']
]
# Узел-исследователь, который принимает параметр competitor
def researcher_node(state: ParallelAgentState, competitor: str):
# Выполняет поиск для конкретного конкурента
result = search_and_summarize(competitor)
return {"research_results": {competitor: result}}
# В графе добавляем параллельный переход
workflow.add_conditional_edges("planner", continue_to_research, ["researcher"])
Это чистая, предсказуемая параллельная обработка без гонок за файлы.
Мы прогнали одну и ту же задачу (анализ трёх конкурентов с формированием отчёта) через CrewAI (иерархический режим) и через наш LangGraph-оркестратор. Вот что получилось:
|
Метрика |
CrewAI (hierarchical) |
LangGraph (наш оркестратор) |
|---|---|---|
|
Время выполнения |
∞ (прервано вручную через 6 минут) |
47 секунд |
|
Количество вызовов LLM |
127 (остановлено) |
14 (включая Planner и Judge) |
|
Стоимость (GPT-4-turbo) |
~$4.30 (и росло) |
~$0.42 |
|
Успешное завершение |
0% (из 5 запусков — 0) |
100% (из 10 запусков — 10) |
|
Качество отчёта (субъективно) |
Случайное: от полного бреда до хорошего |
Стабильно приемлемое |
|
Гонки за ресурсы |
Постоянно |
Отсутствуют (синхронный граф) |
|
Параллелизм |
Заявлен, но не работает как ожидалось |
Реальный параллелизм через Send API |
Мы не изобрели новый фреймворк и не написали сверхсложный код. Мы просто применили принципы надёжного программирования — конечный автомат, ограничение контекста, явные условия перехода — к недетерминированной среде LLM. Оказалось, что этого достаточно, чтобы превратить хаос в рабочий конвейер.
После всего пережитого я обязан задать этот вопрос. Потому что, возможно, мы все стали жертвами хайпа.
Параллельные подпроцессы с разными «личностями». Классический пример: один агент генерирует идеи (креативщик с высокой температурой), второй их критикует (скептик с низкой температурой). Такой «внутренний диалог» действительно улучшает качество.
Симуляция множества точек зрения [5]. Если нужно промоделировать, как разные персоны отреагируют на продукт.
Сложные workflow с ветвлениями. Когда логика процесса нелинейна и зависит от промежуточных результатов, оркестратор на графе — правильное решение.
Линейные пайплайны обработки данных. ETL, последовательная генерация текста с шаблонными шагами. Один агент с хорошо составленным промптом и цепочкой вызовов функций справится быстрее, дешевле и надёжнее.
Простые RAG-системы. Зачем вам агент-ридер и агент-генератор, если можно одним промптом сказать: «Ответь на вопрос, используя эти документы»?
Задачи, где важна скорость и предсказуемость. Каждый дополнительный агент — это дополнительный вызов LLM и точка потенциального отказа.
Совет, который я даю себе полугодовой давности: прежде чем городить рой агентов, напиши цепочку промптов в одном скрипте. Если она решает задачу на 80% — остановись. Добавь пару условных переходов на Python. Если и этого мало — только тогда думай о LangGraph или, прости господи, CrewAI.
Мультиагентные системы — мощный, но опасный инструмент. Без дисциплины они превращаются в бесконечный митинг в Zoom, где каждый участник — это LLM с включённым verbose=True. Дисциплину должны задавать вы, а не языковая модель.
Три правила выживания, которые я вывел из этого опыта [6]:
Не доверяй чат-интерфейсу. Проектируй систему как конвейер с чёткими переходами состояний. Используй LangGraph или любой другой оркестратор, основанный на конечных автоматах. LLM не должна решать, кто говорит следующим — это ваша работа как инженера.
Ограничивай контекст. Не давай агенту читать всю переписку. Передавай только релевантные данные: описание его задачи и результат предыдущего шага. Это сэкономит деньги, токены и убережёт от галлюцинаций.
Всегда ставь стоп-кран. Таймауты, лимиты итераций, максимальное количество вызовов LLM. Ваш бюджет и нервная система [7] скажут вам спасибо. Помните: агенты не устают, они могут «совещаться» вечно. Ваша задача — вовремя сказать «хватит».
А продакт-менеджерам, которые прочитали эту статью и уже представляют, как заменят всю команду одним графом в LangGraph, я скажу так: AI-агенты — это не замена разработчикам. Это просто ещё один слой абстракции, который требует ещё более тщательного проектирования. И да, бесконечные созвоны в зуме никуда не денутся — просто теперь на них будут ходить ваши AI-агенты, пока вы пытаетесь понять, почему они обсуждают цены на AWS вместо фич конкурентов.
Удачи в оркестрации. Держите графы детерминированными, а промпты — короткими.
Автор: kardanShurup
Источник [8]
Сайт-источник BrainTools: https://www.braintools.ru
Путь до страницы источника: https://www.braintools.ru/article/29239
URLs in this post:
[1] ошибкой: http://www.braintools.ru/article/4192
[2] логикой: http://www.braintools.ru/article/7640
[3] забывает: http://www.braintools.ru/article/333
[4] боли: http://www.braintools.ru/article/9901
[5] зрения: http://www.braintools.ru/article/6238
[6] опыта: http://www.braintools.ru/article/6952
[7] нервная система: http://www.braintools.ru/nervous-system
[8] Источник: https://habr.com/ru/articles/1026856/?utm_source=habrahabr&utm_medium=rss&utm_campaign=1026856
Нажмите здесь для печати.