ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать fallback-цепь (Агент А → Агент Б → человек)

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

Разработать надёжную fallback-цепь для системы агентов, где при сбое основного агента (А) управление передаётся резервному агенту (Б), а при его неудаче — оператору-человеку. Задача не должна быть потеряна ни на одном из этапов: каждый шаг логируется, и в случае эскалации человек получает полный контекст. Результат — готовый к интеграции модуль fallback-ора.

Ключевой результат Гарантированная обработка каждого запроса (через А → Б → человек) без потери контекста и с явным подтверждением завершения.

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

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

Если нет реального инструмента — симулируем:

1. Написать простой класс Agent(name, function) с методом run(task).
2. Реализовать FallbackChain(agents=[A, B, Human], ...) с управлением состояниями.
3. Вместо настоящего человека использовать input() и локальный файл очереди.

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

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

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

Действия

1. Определить интерфейсы агентов Каждый агент — асинхронная функция, принимающая task: dict и возвращающая result: dict или поднимающая исключение AgentError(message).

   class AgentError(Exception):
       def __init__(self, message: str, context: dict = None):
           self.context = context or {}
           super().__init__(message)
   

2. Спроектировать граф состояний Использовать конечный автомат (State Machine) с состояниями:

   • INIT
   • AGENT_A_WORKING
   • AGENT_B_WORKING
   • HUMAN_WORKING
   • COMPLETED
   • FAILED

   Переходы:

   • INIT → AGENT_A_WORKING (при старте)
   • AGENT_A_WORKING → COMPLETED (успех)
   • AGENT_A_WORKING → AGENT_B_WORKING (ошибка А)
   • AGENT_B_WORKING → COMPLETED (успех)
   • AGENT_B_WORKING → HUMAN_WORKING (ошибка Б)
   • HUMAN_WORKING → COMPLETED (человек обработал)
   • Любое исключение в HUMAN_WORKING → FAILED (критическая ошибка).

3. Определить структуру задачи

   task = {
       "id": "uuid",
       "input": "текст запроса",
       "history": [],          # логи всех попыток
       "state": "INIT",
       "created_at": timestamp,
       "max_attempts": 3       # на одного агента
   }
   

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

Этап 2: Реализация базового оркестратора (1–1,5 часа)

Действия

1. Написать класс FallbackChain

   • Конструктор принимает список агентов (первый — А, второй — Б, третий — Human-заглушка).
   • Метод run(task): цикл по агентам, каждый раз обновляет task["state"], логирует начало и конец.
   • При исключении AgentError — записывает ошибку в task["history"] и переходит к следующему агенту.

   class FallbackChain:
       def __init__(self, agents: list):
           self.agents = agents          # [A, B, Human]
       
       async def run(self, task: dict) -> dict:
           for level, agent in enumerate(self.agents):
               try:
                   task["state"] = f"AGENT_{level}_WORKING"
                   log_start(task, agent.name)
                   result = await agent.run(task)
                   task["result"] = result
                   task["state"] = "COMPLETED"
                   return task
               except AgentError as e:
                   task["history"].append({
                       "agent": agent.name,
                       "error": str(e),
                       "context": e.context,
                       "timestamp": now()
                   })
                   # если последний агент — ошибка критическая
                   if level == len(self.agents) - 1:
                       task["state"] = "FAILED"
                       return task
                   # иначе продолжаем
           return task
   

2. Добавить логирование

   • Использовать logging.getLogger(name) + logging.FileHandler('fallback.log') в JSON-формате.
   • Каждое событие: id задачи, агент, статус, duration, input_truncated.

3. Реализовать mock-агентов

   • AgentA: успех для 70% тестовых запросов, ошибка для 30%.
   • AgentB: успех для 50% оставшихся, ошибка для остальных.
   • HumanAgent: выводит на экран задачу, ждёт ввода пользователя, записывает ответ в JSON-файл escalated_tasks.json.

Ожидаемый результат этапа Рабочий класс FallbackChain с mock-агентами, который корректно переключает обработку.

Этап 3: Интеграция с хранилищем состояний и восстановление после сбоя (45 минут)

Действия

1. Добавить поддержку долговременного хранилища (SQLite или Redis).

   • Перед каждым вызовом агента сохранять состояние task в БД с ключом task["id"].
   • При старте агента загружать состояние из БД, чтобы при падении процесса можно было продолжить.

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

   • Метод recover(task_id) — загружает задачу из БД, проверяет state, и возобновляет с того же уровня (если state AGENT_X_WORKING, запускается соответствующий агент).

3. Написать тест для аварийного восстановления

   • Создать задачу, принудительно прервать выполнение (raise KeyboardInterrupt в агента), затем вызвать recover() и убедиться, что задача продолжается без дублирования.

Ожидаемый результат этапа Конфигурация с SQLite, возможность перезапуска задачи после падения.

Этап 4: Написание тестов и валидация (45 минут – 1 час)

Действия

1. Покрыть unit-тестами

   • test_full_success: задача выполняется первым агентом.
   • test_fallback_to_B: A падает, B успевает.
   • test_escalation_to_human: A и B падают, человек завершает.
   • test_critical_failure: все агенты падают, статус FAILED.
   • test_recovery: прерывание и восстановление.

   @pytest.mark.asyncio
   async def test_fallback_to_B():
       chain = FallbackChain([FailingAgent("A"), SucceedAgent("B"), HumanAgent()])
       task = new_task("test")
       result = await chain.run(task)
       assert result["state"] == "COMPLETED"
       assert "B" in [h["agent"] for h in result["history"]]
   

2. Создать end-to-end сценарий

   • Запустить fallback-цепь на наборе из 10 тестовых задач.
   • Проверить, что:
     • ни одна задача не потеряна (все имеют state == COMPLETED или FAILED с причиной),
     • количество эскалаций человеку соответствует ожидаемому (с учётом вероятностей).

3. Симулировать реальную эскалацию

   • Файл escalated_tasks.json: содержит задачи, которые дошли до человека.
   • Написать скрипт, который читает этот файл и позволяет смоделировать ответ человека через input() в текстовом интерфейсе.

Ожидаемый результат этапа Прогон всех тестов зелёный, JSON-отчёт о прогоне.

Этап 5: Документация и демонстрация (30 минут)

Действия

1. Написать README с архитектурой

   • Схема переходов (ASCII или draw.io).
   • Пример запуска.
   • Инструкция по добавлению нового агента в цепь.

2. Создать скрипт демонстрации

   • python demo.py — запускает 5 задач с разными сценариями, выводит лог на экран.

3. Упаковать код

   • Структура:

     fallback-chain/
     ├── README.md
     ├── requirements.txt
     ├── src/
     │   ├── __init__.py
     │   ├── chain.py
     │   ├── agents.py
     │   ├── storage.py
     │   └── logger.py
     ├── tests/
     │   └── test_chain.py
     └── escalated_tasks.json (игнорируется git)
     

Ожидаемый результат этапа Готовый к просмотру репозиторий с документацией.

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

• Реализована цепь из трёх звеньев: AgentA → AgentB → Human (mock).
• При ошибке AgentA задача автоматически переходит к AgentB с полным контекстом.
• При ошибке AgentB задача эскалируется человеку (mock) с контекстом и историей.
• Ни одна задача не теряется: все имеют статус COMPLETED или FAILED с логом.
• Реализовано долговременное хранение состояния (SQLite) с возможностью восстановления.
• Написаны юнит-тесты (минимум 5) — все проходят.
• Документация содержит архитектуру, пример запуска и инструкцию по доработке.
• Логирование в JSON-формате фиксирует все переходы, ошибки и длительность.
• Восстановление после аварийного останова сохраняет позицию и не дублирует обработку.

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

Основной артефакт Python-пакет fallback_chain с классами FallbackChain, AgentError, хранилищем состояний и демо-скриптом. Включает:

• src/chain.py — оркестратор с переходом по агентам.
• src/agents.py — mock-агенты (A, B, Human).
• tests/test_chain.py — тесты pytest.
• demo.py — демонстрация.
• README.md — описание.
• escalated_tasks.json — пример эскалированных задач для тестирования ввода человека.

Дополнительные результаты

• Логи fallback-цепи в файле fallback.log (JSONL).
• Чек-лист покрытия сценариев (успех, fallback, эскалация, критическая ошибка).
• Возможность легко заменить mock на реальные агенты (соответствие интерфейсу async def run(task)).

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

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

Примечание: Для первого раза закладывайте +50% времени на отладку интеграции с хранилищем и тестирование восстановления.

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

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

• Я реализовал замкнутую цепь: при ошибке А → Б, при ошибке Б → человек.
• Я убедился, что task["history"] передаётся между агентами без потерь.
• Я написал тест для каждого сценария (успех А, успех Б, эскалация, крит-ошибка).
• Я проверил, что восстановление после принудительного останова не дублирует работу.
• Я задокументировал, как добавить нового агента (реализовать интерфейс async def run(task)).