Розробка 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, автоматизації звітності та генерації контенту.

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