Розробка мульти-агентної AI-системи

Розвиток мульти-агентної AI-системи

Мульти-агентна система (MAS) — архітектура, в якій кілька спеціалізованих AI-агентів взаємодіють для виконання складних завдань, недоступних одному агенту. Розділення відповідальності зменшує складність кожного агента, підвищує якість у спеціалізованих завданнях та дозволяє масштабувати систему горизонтально.

Архітектури мульти-агентних систем

Supervisor (Оркестратор): центральний агент розподіляє завдання між спеціалізованими агентами й агрегує результати. Добре керується, але є вузьке місце в оркестраторі.

Peer-to-peer (P2P): агенти спілкуються безпосередньо, без центрального координатора. Більш стійка до відмов, важче налагоджується.

Hierarchical (Ієрархічна): багаторівнева структура — агенти верхнього рівня керують агентами нижнього рівня. Використовується для дуже складних робочих процесів.

Pipeline: кожен агент виконує свою частину завдання й передає результат наступному. Простий, передбачуваний, обмежений лінійністю.

Реалізація паттерна Supervisor з LangGraph

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()

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 за Q1 2026",
    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": "Q1 2026"})

Практичний кейс: система автоматизації due diligence

Завдання: комплексна перевірка компанії при M&A угоді. Включає юридичний, фінансовий, операційний та HR аналіз.

Склад агентів:

• Financial Analyst Agent: аналіз звітності МСФО/ГААП
• Legal Agent: перевірка договірної бази, судових суперечок
• HR Agent: аналіз структури персоналу, текучості, ключових співробітників
• Risk Agent: зведений аналіз ризиків
• Report Agent: фінальний звіт Due Diligence

Інфраструктура: LangGraph, кожен агент з доступом до спеціалізованих RAG-індексів.

Результати:

• Скорочення часу DD: з 4 тижнів до 3 днів (для стандартних кейсів)
• Покриття (частка перевіреної аспектів): 78% → 94%
• Human-in-the-loop: фінальна валідація кожного розділу
• Вартість одного DD: зменшення на 67%

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

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

Графік

• Проектування архітектури MAS: 1–2 тижні
• Розвиток базових агентів (3–5): 3–5 тижнів
• Інтеграція комунікації та тестування: 2–3 тижні
• Моніторинг і production hardening: 1–2 тижні
• Всього: 7–12 тижнів