Нещодавній проєкт: клієнт із фінансового сектору хотів автоматизувати 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() Зв'яжіться з нами для обговорення вашого проєкту. Отримайте консультацію — ми допоможемо обрати оптимальну архітектуру.