Разработка мульти-агентных AI-систем под ключ с гарантией

Недавний проект: клиент из финсектора хотел автоматизировать 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

Практический кейс: due diligence за 3 дня

Мы разработали систему для финтех-компании, обрабатывающую M&A сделки. Состав агентов: Financial Analyst (МСФО, РСБУ), Legal (договоры, суды), HR (текучка, ключевые сотрудники), Risk (сводный риск), Report (финальный документ). Инфраструктура: LangGraph с RAG-индексами для каждого агента. Результаты: время сокращено с 4 недель до 3 дней, покрытие аспектов выросло с 78% до 94%, стоимость одного DD снижена на 67% — экономия составила от $20k–50k. Человек-в-петле — финальная валидация каждого раздела.

Что входит в разработку мульти-агентной системы

• Архитектурная схема с описанием ролей агентов, протоколов обмена, точек отказа.
• Baseline-агенты (3–5 штук) с настройками LLM, инструментов, RAG-индексов.
• Система оркестрации (LangGraph или кастомный Supervisor) с обработкой ошибок и повторными попытками.
• Дашборд мониторинга (latency p99, token usage, rerun rate) — часть MLOps агентов.
• Тестовый полигон с 10+ сценариями, покрывающими штатные и краевые случаи.
• Документация в формате README + декабрьская презентация для стейкхолдеров.
• Обучение команды (2–3 вебинара по доработке агентов).

Этапы разработки

Сроки и стоимость

Ориентировочные сроки — от 7 до 12 недель в зависимости от сложности. Стоимость разработки MAS под ключ рассчитывается индивидуально после аудита вашей задачи. Закажите консультацию — мы оценим ваш проект и предложим архитектуру.

Коммуникация между агентами

# Паттерн: агенты передают структурированные сообщения через 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() 

Свяжитесь с нами для обсуждения вашего проекта. Получите консультацию — мы поможем выбрать оптимальную архитектуру.