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

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

Представьте: у вас пять AI-агентов. Каждый умеет своё — анализировать финансы, проверять юридические документы, оценивать рынок. Вы даёте команду провести due diligence компании. Без оркестратора начинается хаос: агенты вызываются последовательно, при ошибке всё перезапускается, параллельные задачи не выполняются, а p99 latency уходит за минуту. Мы столкнулись с этим на реальном проекте и построили оркестратор на LangGraph, который сократил время выполнения DD с 4 недель до 3 дней. В оркестраторе всё в 9 раз быстрее, а покрытие аспектов проверки достигает 94%. Это не предел — можно добиться ещё большей эффективности.

Почему без оркестратора агенты неэффективны?

Типичная проблема — каждый агент работает изолированно. Вы последовательно передаёте запрос от исследователя к аналитику, затем к писателю. Ошибка на любом этапе перезапускает всю цепочку. Если один агент завис, весь процесс стопорится. Параллельные задачи отсутствуют. В итоге — низкая утилизация 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%
• Стоимость одного DD снизилась на 71% — экономия составляет десятки тысяч долларов на каждом проекте. Для крупных корпораций экономия может достигать сотен тысяч.

Сравним подходы:

Распределение задач между агентами

Мониторинг и трассировка

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, автоматизации отчётности и генерации контента.

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