ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Multi-agent для планирования

1. Цель задачи

Разработать multi-agent систему, где Planner-агент декомпозирует сложную задачу на последовательность шагов (plan), а Executor-агент выполняет эти шаги через вызовы внешних API. Система должна справляться с задачей, состоящей из 5 логически связанных шагов, корректно передавая контекст между шагами и обрабатывая ошибки.

Ключевой результат Работающий прототип multi-agent pipeline, который принимает на вход задачу на естественном языке, автоматически строит план из 5 шагов, выполняет каждый шаг через реальные (или симулированные) API вызовы и возвращает итоговый результат.

2. Исходные данные

Перед началом необходимо иметь:

Если нет реальных API — симулируем:

1. Создать простой FastAPI/Flask сервер с двумя-тремя mock-ручками (/weather, /search_nearby, /book_table, /send_email).
2. Каждая ручка принимает JSON, ждёт случайную задержку (0.5–1.5 сек) и возвращает фиксированный или параметризированный ответ.
3. Для усложнения добавить случайные ошибки (5% сбоя) — Executor должен обрабатывать retry.

3. Технологический стек

4. Этапы выполнения

Этап 1: Проектирование архитектуры агентов (30 минут)

Действия

1. Определить структуру Agent-a Каждый агент — это класс с методом run(context: dict) -> dict. Context содержит:

   • task: исходная задача пользователя.
   • plan: список шагов (генерируется Planner).
   • state: текущий шаг, собранные результаты.
   • errors: список ошибок.

2. Спроектировать Planner Он должен принимать task и возвращать:

   • список шагов (каждый шаг: {"step_id": 1, "description": "Запросить погоду", "api_endpoint": "weather", "params": {"city": "Moscow"}}).

3. Спроектировать Executor Он принимает шаг, вызывает соответствующий API, обрабатывает ответ и дополняет state.

4. Продумать оркестрацию Цикл: Planner → for each step: Executor → проверка результата → переход к следующему шагу.

5. Нарисовать диаграмму потоков (схематично в README или Mermaid).

Ожидаемый результат этапа Документ (или кодовая заготовка) с описанием классов и интерфейсов агентов.

Этап 2: Реализация Planner-агента (1 час)

Действия

1. Реализовать класс PlannerAgent

   • Инициализация с LLM клиентом.
   • Метод create_plan(task: str) -> List[Dict].
   • Prompt для LLM: "Ты — планировщик. Разбей задачу на ровно 5 шагов. Для каждого шага укажи: step_id, description, api_endpoint (один из: weather, search_nearby, get_hours, book_table, send_email), params (обязательные поля для вызова API). Верни только JSON массив."
   • Парсинг ответа: извлечение JSON, валидация (проверить, что ровно 5 шагов, все поля есть).

2. Написать юнит-тест

   • Передать тестовую задачу.
   • Проверить, что возвращается 5 шагов.
   • Проверить, что каждый шаг содержит step_id, description, api_endpoint, params.

def test_planner_creates_five_steps():
    planner = PlannerAgent(llm_client)
    task = "Organize a morning routine: check weather, find a café, get opening hours, book a table, send confirmation."
    plan = planner.create_plan(task)
    assert len(plan) == 5
    assert all("api_endpoint" in step for step in plan)

Ожидаемый результат этапа Работающий PlannerAgent, который для заданной задачи возвращает корректный план из 5 шагов.

Этап 3: Реализация Executor-агента и симуляция API (1 час 30 минут)

Действия

1. Создать mock-сервер на FastAPI:

   from fastapi import FastAPI
   import random, asyncio
   app = FastAPI()
   
   @app.post("/weather")
   async def weather(params: dict):
       await asyncio.sleep(random.uniform(0.3, 1.0))
       if random.random() < 0.05:
           return {"error": "service_unavailable"}, 503
       return {"temperature": 12, "conditions": "cloudy", "city": params.get("city")}
   
   # Аналогично для /search_nearby, /get_hours, /book_table, /send_email
   

2. Реализовать класс ExecutorAgent

   • Инициализация с базовым URL mock-сервера.
   • Метод execute_step(step: dict, context: dict) -> dict.
   • Внутри: HTTP POST к mock-эндпоинту с params.
   • Обработка ошибок: при 5xx — retry до 3 раз с экспоненциальной задержкой.
   • Запись результата в context["state"][step_id].

3. Добавить трассировку логировать каждый вызов (время, эндпоинт, статус, ответ).

4. Написать интеграционный тест запустить mock-сервер, выполнить один шаг, проверить ответ.

Ожидаемый результат этапа ExecutorAgent, который успешно вызывает mock-сервер и обрабатывает ошибки.

Этап 4: Оркестрация и запуск полного пайплайна (1 час)

Действия

1. Реализовать класс Orchestrator

   • Инициализация с Planner и Executor.
   • Метод run(task: str) -> dict:

     context = {"task": task, "plan": [], "state": {}, "errors": []}
     plan = planner.create_plan(task)
     context["plan"] = plan
     for step in plan:
         result = executor.execute_step(step, context)
         if "error" in result:
             context["errors"].append(result)
             break  # или continue — по решению
         context["state"][step["step_id"]] = result
     return context
     

2. Протестировать на задаче, которая состоит из 5 шагов (например, описанная в Этапе 2). Убедиться, что:

   • Все 5 шагов выполняются последовательно.
   • Результаты предыдущих шагов доступны для следующих (если нужно) через context.
   • При возникновении ошибки Executor делает retry.

3. Добавить финальную агрегацию собрать результаты всех шагов в человекочитаемый ответ (через LLM или просто форматированный текст).

4. Провести ручное тестирование 3–5 различных задач, убедиться, что план всегда содержит именно 5 шагов (можно жёстко зашить в prompt требование 5 шагов).

Ожидаемый результат этапа Полный pipeline, который для задачи «узнать погоду → найти кафе → проверить часы → забронировать → отправить email» проходит все 5 шагов без ошибок.

Этап 5: Обработка ошибок и наблюдаемость (30 минут)

Действия

1. Добавить обработку ошибок на уровне Orchestrator:

   • Если шаг провалился после всех retry — записать в errors и прервать выполнение (опционально перезапустить планер с дополнительным контекстом).
   • Добавить fallback: если шаг book_table не удался, попробовать альтернативный эндпоинт (например, reserve_alternative).

2. Внедрить логирование в JSON-формате для дальнейшего анализа:

   {"timestamp": "...", "step": 1, "action": "execute", "status": "success", "duration_ms": 1200}
   

3. Подготовить простой отчёт (в stdout или файл): сколько шагов выполнено, сколько ошибок, время выполнения.

Ожидаемый результат этапа Система корректно обрабатывает ошибки (retry + fallback) и выводит структурированный лог.

5. Критерии приемки (Definition of Done)

• Planner генерирует ровно 5 шагов для любой подходящей задачи.
• Executor корректно вызывает mock-эндпоинты и обрабатывает ответы.
• При ошибке API Executor делает до 3 retry с задержкой.
• Orchestra передаёт контекст между шагами (результат предыдущего шага доступен следующему).
• При успешном выполнении всех 5 шагов возвращается сводный результат.
• При фатальной ошибке выполнение прерывается и ошибка логируется.
• Весь код покрыт минимум 2 юнит-тестами (Planner, Executor).
• Mock-сервер запускается отдельно и предоставляет все 5 эндпоинтов.
• Система работает с реальной LLM (OpenAI/GPT-4o-mini) или локальной через Ollama.
• Время выполнения полного цикла для 5 шагов не превышает 30 секунд (с учётом сетевых задержек).

6. Ожидаемый результат

После выполнения задачи должен быть создан репозиторий (или папка) со следующей структурой:

multi-agent-planner/
├── agents/
│   ├── __init__.py
│   ├── planner.py          # PlannerAgent класс
│   ├── executor.py         # ExecutorAgent класс
│   └── orchestrator.py     # Orchestrator класс
├── mock_server/
│   ├── __init__.py
│   ├── main.py             # FastAPI приложение с 5 ручками
│   └── requirements.txt
├── tests/
│   ├── test_planner.py
│   ├── test_executor.py
│   └── test_integration.py
├── logs/
│   └── trace.log           # Пример лога
├── requirements.txt
├── README.md               # Архитектура, запуск, примеры
└── config.py               # Настройки (базовый URL, API ключи)

Содержание README.md описание архитектуры, инструкция по запуску, пример выполнения задачи.

Опционально Dockerfile для сборки всего решения (mock + агенты) и docker-compose для одного вызова.

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание Если LLM API используется впервые (настройка ключа, установка библиотеки), добавьте 30–60 минут.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я развернул mock-сервер и убедился, что все 5 эндпоинтов отвечают.
• Я проверил, что Planner всегда генерирует ровно 5 шагов (включая граничные случаи).
• Я протестировал Executor с разными сценариями (успех, ошибка, таймаут).
• Я выполнил end-to-end прогон для задачи с 5 шагами и получил корректный финальный результат.
• Я написал хотя бы один тест, который проверяет обработку сбоя и retry.
• Я подготовил README с инструкцией по запуску.
• Я проверил, что все зависимости перечислены в requirements.txt.