Розробка AI-оркестратора агентів (Agent Orchestration)

Розробка AI-оркестратора агентів (Agent Orchestration)

Уявіть: у вас п'ять AI-агентів. Кожен вміє своє — аналізувати фінанси, перевіряти юридичні документи, оцінювати ринок. Ви даєте команду провести due diligence компанії. Без оркестратора починається хаос: агенти викликаються послідовно, при помилці все перезапускається, паралельні завдання не виконуються, а p99 latency сягає за хвилину. Ми зіткнулися з цим на реальному проєкті для нашого клієнта — великої інвестиційної компанії — та побудували оркестратор на LangGraph, який скоротив час виконання DD з 4 тижнів до 3 днів. В оркестраторі все в 9 разів швидше, а покриття аспектів перевірки сягає 94% (проти 78% раніше). Економія для клієнта: вартість кожного DD знизилася з $100 000 до $29 000, що дало понад $1 млн на рік при 15 перевірках. Це не межа — можна досягти ще більшої ефективності.

Чому без оркестратора агенти неефективні?

Типова проблема — кожен агент працює ізольовано. Ви послідовно передаєте запит від дослідника до аналітика, потім до письменника. Помилка на будь-якому етапі перезапускає весь ланцюжок. Якщо один агент завис, весь процес зупиняється. Паралельні завдання відсутні. У підсумку — низька утилізація GPU (близько 15%), величезна затримка та постійні ручні виправлення. Наприклад, при ручній обробці due diligence витрачається до 4 тижнів, а вартість такого проєкту може сягати значних сум.

Як LangGraph спрощує оркестрацію?

LangGraph — графовий фреймворк, що дозволяє описати залежності між агентами як орієнтований граф. Вузли — агенти, ребра — переходи за умовами. Вбудовані checkpoint'и через MemorySaver зберігають стан після кожного кроку. Retry-логіка та fallback — з коробки. Нижче — реалізація оркестратора з LangGraph, яку ми використовуємо в комерційних проєктах.

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
import operator
import json

class OrchestratorState(TypedDict):
    user_request: str
    task_plan: list[dict]           # [{task_id, description, agent, status, result}]
    current_task_index: int
    agent_results: Annotated[dict, lambda a, b: {**a, **b}]
    final_response: str
    error_count: int

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Реєстр агентів
AGENT_REGISTRY = {
    "researcher": ResearcherAgent(),
    "analyst": AnalystAgent(),
    "writer": WriterAgent(),
    "sql_agent": SQLAgent(),
    "code_interpreter": CodeInterpreterAgent(),
    "file_processor": FileProcessorAgent(),
}

def plan_tasks(state: OrchestratorState) -> OrchestratorState:
    """Оркестратор розбиває завдання на підзавдання та призначає агентів"""

    available_agents = list(AGENT_REGISTRY.keys())

    response = llm.invoke(f"""
Розбий наступне завдання на підзавдання та признач агента для кожного.
Доступні агенти: {available_agents}

Завдання: {state["user_request"]}

Поверни JSON-список:
[{{"task_id": "t1", "description": "...", "agent": "researcher", "dependencies": []}}]
Залежності: список task_id, які мають завершитися перед цим завданням.
""")

    task_plan = json.loads(response.content)
    for task in task_plan:
        task["status"] = "pending"
        task["result"] = None

    return {**state, "task_plan": task_plan, "current_task_index": 0}

def execute_next_task(state: OrchestratorState) -> OrchestratorState:
    """Виконує наступне готове завдання"""
    task_plan = state["task_plan"].copy()

    # Знаходимо наступне завдання, усі залежності якого виконано
    next_task = None
    for task in task_plan:
        if task["status"] == "pending":
            deps_completed = all(
                any(t["task_id"] == dep and t["status"] == "completed"
                    for t in task_plan)
                for dep in task.get("dependencies", [])
            )
            if deps_completed:
                next_task = task
                break

    if not next_task:
        return {**state, "current_task_index": -1}  # Усі завдання завершено

    # Виконуємо завдання через відповідного агента
    agent = AGENT_REGISTRY.get(next_task["agent"])
    if not agent:
        next_task["status"] = "failed"
        next_task["result"] = f"Agent {next_task['agent']} not found"
    else:
        # Передаємо результати залежностей як контекст
        dependency_results = {
            dep: state["agent_results"].get(dep)
            for dep in next_task.get("dependencies", [])
        }

        try:
            result = agent.execute(
                task=next_task["description"],
                context=dependency_results,
            )
            next_task["status"] = "completed"
            next_task["result"] = result
        except Exception as e:
            next_task["status"] = "failed"
            next_task["result"] = str(e)

    # Оновлюємо plan
    updated_plan = [
        task if task["task_id"] != next_task["task_id"] else next_task
        for task in task_plan
    ]

    return {
        **state,
        "task_plan": updated_plan,
        "agent_results": {next_task["task_id"]: next_task["result"]},
    }

def should_continue(state: OrchestratorState) -> str:
    """Визначає наступний крок оркестратора"""
    pending = [t for t in state["task_plan"] if t["status"] == "pending"]
    failed = [t for t in state["task_plan"] if t["status"] == "failed"]

    if failed and state["error_count"] >= 3:
        return "finalize_with_errors"
    if not pending:
        return "aggregate_results"
    return "execute_next"

def aggregate_results(state: OrchestratorState) -> OrchestratorState:
    """Збирає результати всіх агентів у фінальну відповідь"""
    all_results = {t["task_id"]: t["result"] for t in state["task_plan"]}

    final = llm.invoke(f"""
На основі результатів від різних агентів сформуй фінальну відповідь.
Оригінальний запит: {state["user_request"]}
Результати: {json.dumps(all_results, ensure_ascii=False)}
""").content

    return {**state, "final_response": final}

# Побудова графа
graph = StateGraph(OrchestratorState)
graph.add_node("plan", plan_tasks)
graph.add_node("execute_next", execute_next_task)
graph.add_node("aggregate_results", aggregate_results)

graph.set_entry_point("plan")
graph.add_edge("plan", "execute_next")
graph.add_conditional_edges("execute_next", should_continue, {
    "execute_next": "execute_next",
    "aggregate_results": "aggregate_results",
    "finalize_with_errors": "aggregate_results",
})
graph.add_edge("aggregate_results", END)

orchestrator = graph.compile(checkpointer=MemorySaver())

Як оркестратор виконує завдання паралельно?

import asyncio

async def execute_parallel_tasks(tasks_batch: list[dict]) -> list[dict]:
    """Паралельне виконання незалежних завдань"""
    coroutines = []
    for task in tasks_batch:
        agent = AGENT_REGISTRY.get(task["agent"])
        if agent:
            coroutines.append(asyncio.to_thread(agent.execute, task=task["description"]))

    results = await asyncio.gather(*coroutines, return_exceptions=True)

    for task, result in zip(tasks_batch, results):
        if isinstance(result, Exception):
            task["status"] = "failed"
            task["result"] = str(result)
        else:
            task["status"] = "completed"
            task["result"] = result

    return tasks_batch

Практичний кейс: оркестратор для due diligence (наш клієнт — велика інвестиційна компанія)

З нашої практики — автоматизована перевірка компанії при M&A. Паралельна робота 5 агентів:

1. Financial Agent: аналіз 3 років звітності
2. Legal Agent: перевірка судових спорів, обмежень
3. HR Agent: структура персоналу, плинність
4. Market Agent: положення на ринку, конкуренти
5. Risk Agent: синтез ризиків з усіх джерел

Граф виконання:

• t1 (financial), t2 (legal), t3 (hr), t4 (market) — паралельно
• t5 (risk) — залежить від t1, t2, t3, t4
• t6 (final_report) — залежить від t5

Результати (для нашого клієнта):

• Час DD: 4 тижні → 3 дні (в 9 разів швидше)
• Покриття аспектів: 78% → 94% (покращення на 16 п.п.)
• Вартість одного DD знизилася на 71% — економія становить понад $70 000 на кожному проєкті. Для нашого клієнта з 15 перевірками на рік це дало $1 065 000 економії.

Порівняємо підходи:

Розподіл завдань між агентами

Моніторинг і трасування

import mlflow

def log_orchestration_run(state: OrchestratorState):
    with mlflow.start_run():
        mlflow.log_metrics({
            "total_tasks": len(state["task_plan"]),
            "completed_tasks": sum(1 for t in state["task_plan"] if t["status"] == "completed"),
            "failed_tasks": sum(1 for t in state["task_plan"] if t["status"] == "failed"),
        })
        mlflow.log_text(json.dumps(state["task_plan"], indent=2), "task_execution_log.json")

Що входить у роботу

• Проєктування архітектури мульти-агентної системи під вашу задачу
• Реалізація оркестратора на LangGraph або аналогічному фреймворку
• Розробка спеціалізованих AI-агентів (до 10) з інтеграцією до ваших даних
• Налаштування паралельного виконання, retry-логіки та fallback
• Моніторинг і трасування через MLflow, Weights & Biases
• Документація схеми графа, API агентів, інструкція з експлуатації
• Навчання вашої команди (2–3 сесії)

Терміни орієнтовно

• Проєктування оркестратора: 1–2 тижні
• Реалізація базових агентів (3–5): 3–5 тижнів
• Інтеграція паралельного виконання: 1 тиждень
• Error handling і моніторинг: 1–2 тижні
• Разом: 6–10 тижнів

Гарантуємо стабільну роботу в продакшені — всі рішення покриті unit-тестами та інтеграційними тестами. У нас за плечима 10+ років досвіду в AI/ML, і ми впроваджували оркестратори для financial due diligence, автоматизації звітності та генерації контенту.

Замовте консультацію з впровадження оркестратора — це безкоштовно і ні до чого не зобов'язує. Зв'яжіться з нами — оцінимо ваш проєкт, підберемо архітектуру та терміни впровадження.

Практичний розбір LLM: fine-tuning, RAG, агенти, деплой

Модель GPT‑4 або Claude 3.5 Sonnet через публічне API — не рішення, а просто інструмент. Коли приходить вимога «зробити як ChatGPT, але на наших даних», за нею стоїть реальна інженерна задача: від налаштування промптів до навчання 70B‑моделі на власній інфраструктурі. LLM розробка під ключ — це складний стек, і ми займаємося цим понад 5 років. За цей час реалізовано понад 20 проєктів у галузі генеративного AI: від RAG‑систем для юридичних департаментів до кастомних агентів для техпідтримки. Де саме знаходиться ваша задача — залежить від даних, latency‑вимог, бюджету та того, наскільки критична конфіденційність.

Типова ситуація: клієнт уже спробував ChatGPT, але результати нестабільні — то відповідає точно, то галюцинує. Або потрібна інтеграція в корпоративний портал з дотриманням політик безпеки. Розберемо кожен шар стеку в деталях — від RAG до production‑деплою.

Чому RAG‑системи ламаються і як це виправити?

RAG (Retrieval‑Augmented Generation) виглядає просто: знайшли релевантні документи, поклали в контекст, модель відповіла. На практиці збоїть у кількох місцях.

Chunking без перекриття. Класична помилка: chunk_size=512, overlap=0. Якщо відповідь лежить на межі двох чанків, retrieval не знайде жодного з достатньою впевненістю. Рішення: overlap 15–25% від chunk_size, а краще sentence‑aware splitting через spaCy або NLTK, а не наївне розбиття за символами.

Поганий embedder. Текст‑embedding‑ada‑002 — хороший для загального випадку, але на юридичних або медичних текстах програє спеціалізованим моделям: E5‑large‑v2, BGE‑M3 або fine‑tuned sentence‑transformers на доменних даних. Різниця в Recall@5 може становити 15–25%.

Відсутність re‑ranking. Векторний пошук оптимізований за швидкістю, не за релевантністю. Cross‑encoder re‑ranker (ms‑marco‑MiniLM‑L‑6‑v2, bge‑reranker‑large) після первинного retrieval піднімає точність топ‑3 при прийнятній затримці (+50–150 ms). Це часто важливіше за покращення embedding‑моделі.

Гібридний пошук. Тільки dense вектори погано працюють на точних запитах: імена, артикули, коди. BM25 (sparse) добре знаходить точні збіги, але не розуміє семантику. Гібрид через RRF (Reciprocal Rank Fusion) — оптимальний компроміс. Qdrant, Weaviate та pgvector 0.7+ підтримують гібридний пошук нативно.

Коли варто fine‑tune, а не промпт‑інжиніринг?

Промпт‑інжиніринг вирішує ~70% завдань адаптації LLM під домен. Решта 30% вимагають донавчання. Три ознаки: модель ігнорує специфічний формат виведення навіть при детальному описі в промпті; задача вимагає глибокого знання спеціалізованої лексики (медицина, право); потрібно значно знизити витрати на токени, замінивши велику модель меншою спеціалізованою.

LoRA та QLoRA — стандарт для SFT. LoRA додає trainable low‑rank матриці до attention‑шарів. Типова конфігурація для Llama‑3 8B: r=64, lora_alpha=128, target_modules=["q_proj","v_proj","k_proj","o_proj"] — параметрів, що навчаються, ~0.8%, навчання на одній A100 40GB. QLoRA додає 4‑бітну квантизацію (NF4) і дозволяє fine‑tune 70B модель на двох A100 40GB, хоча швидкість падає вдвічі порівняно з bf16.

DPO замість RLHF. Direct Preference Optimization вимагає лише пари (chosen, rejected), а не скалярні reward‑сигнали. DPOTrainer з бібліотеки trl (Hugging Face) реалізує це кількома десятками рядків.

Типова помилка. Датасет з 500 прикладів, 5 епох, validation loss 0.8 — здається норм. Але на тесті модель деградувала на загальних інструкціях. Причина: catastrophic forgetting. Рішення — додати 10–20% загальних instruction‑following прикладів (Alpaca, FLAN) у навчальну вибірку, щоб не зруйнувати вихідні здібності.

Як обрати базову модель: 8B чи 70B?

Для більшості задач fine‑tuning 8B моделі достатньо. 70B потрібен, коли потрібне глибоке міркування або baseline 8B не досягає потрібної якості навіть після донавчання. Вартість інференсу Llama‑3 8B через vLLM на A100 значно нижча, ніж у GPT‑4, що робить його економічно вигідним.

Що дає PagedAttention в production?

vLLM — перший вибір для serving open‑source моделей. PagedAttention — ключове технічне рішення: KV‑cache керується як virtual memory в ОС, без фрагментації. Це дає throughput у 2–4 рази вище порівняно з наївним HuggingFace Transformers inference. Документація vLLM підтверджує: continuous batching та PagedAttention — стандарт для високонавантажених LLM‑сервісів.

Типові числа на A100 80GB для Llama‑3 8B (bf16): 400–600 req/s, P50 latency 200–400ms, P99 latency 600–900ms при concurrency 64. Для 70B на двох A100 з tensor parallelism: 80–120 req/s, P99 latency 1.5–2.5s. Квантизація AWQ або GPTQ знижує споживання пам'яті в 2 рази при втраті якості в межах 1–3%.

Мультиагентні системи

Агенти — LLM з доступом до інструментів: пошук, виконання коду, запити до API, робота з БД. Основні патерни:

• ReAct (Reason + Act): модель розмірковує → обирає інструмент → спостерігає результат → знову розмірковує. LangChain та LlamaIndex реалізують з коробки.
• Multi‑agent orchestration: кілька спеціалізованих агентів з координатором зверху. Приклад: coordinator → researcher (пошук + summarization) → coder (генерація та виконання коду) → critic (перевірка). Інструменти: AutoGen (Microsoft), CrewAI, кастомна реалізація на LangGraph.

В продакшені агентні системи недетерміновані. Обов'язкові guardrails, ліміти кроків, логування кожного кроку, human‑in‑the‑loop для критичних дій.

Як ми гарантуємо якість LLM рішення?

Ми використовуємо RAGAS для автоматичної оцінки відповідей: faithfulness, answer relevancy, context precision. Система трекінгу експериментів на базі MLflow фіксує всі метрики, датасети та конфіги. Це дозволяє порівнювати різні гіпотези та доводити покращення з цифрами. Гарантію стабільної роботи забезпечує continuous integration з тестами на специфічних сценаріях (prompt injection, edge‑cases).

Як почати LLM розробку: наступні кроки

Ми передаємо:

• Технічну документацію (model card, конфіги, інструкції з розгортання)
• Доступ до інфраструктури (репозиторій з кодом, навчені ваги)
• 1 місяць підтримки після деплою (консультації, виправлення багів)
• Навчання команди замовника (2–3 заняття з експлуатації системи)

Терміни: базовий RAG‑прототип — 1–2 тижні. Fine‑tuning з даними замовника — 3–6 тижнів (з урахуванням підготовки даних). Production‑система з моніторингом та перенавчанням — 2–4 місяці.

Вартість розраховується індивідуально і залежить від обсягу даних, складності моделі та вимог до інфраструктури. Хочете оцінити свій проєкт? Зв'яжіться з нами — ми підготуємо попереднє резюме за 1–2 робочі дні. Або замовте консультацію фахівця з вибору підходу: RAG, fine‑tuning або гібрид — розповімо, що підійде саме вам.