ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать consensus механизм для агентов (Weighted Voting)

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

Разработать и реализовать механизм достижения консенсуса в мультиагентной системе на основе взвешенного голосования. Система должна позволять группе автономных агентов с разными весами (значимостью) согласовывать общее решение при наличии конфликтующих мнений. Ключевой результат работающий прототип, в котором агенты успешно разрешают конфликтные ситуации и приходят к единому решению, используя weighted voting с заданными правилами.

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

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

1. Реализуем агентов как Python-классы в одном процессе, обмен сообщениями — через очереди asyncio.Queue.
2. Для демонстрации конфликтов создаём 4–6 агентов с искусственно заданными весами и предпочтениями.
3. Отсутствие внешнего API (например, LLM) не критично — мнения задаются статически или генерируются простым правилом.

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

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

Этап 1: Проектирование протокола консенсуса (3 ч)

Действия

1. Определить роли: агент (голосует), координатор (собирает голоса, объявляет результат).
2. Задать структуру голоса: {'agent_id': str, 'vote': str/float, 'weight': float}.
3. Выбрать стратегию агрегации:
   • Weighted majority — побеждает вариант с суммарным весом > 50% от общего веса.
   • Weighted average — для числовых вариантов.
4. Определить обработку конфликтов:
   • Если ни один вариант не набирает более 50% — переголосование (максимум 3 раунда), затем автоматический выбор варианта с наибольшим весом.
   • При равенстве весов двух лидеров — запуск аргументации (агент-лидер может изменить вес? нет, фиксированный).
5. Описать протокол сообщений: START_VOTE, VOTE, RESULT, RETRY, ABORT.

Ожидаемый результат этапа документ (в коде — docstring протокола) с описанием фаз, форматов сообщений и правил выхода из конфликта.

Этап 2: Реализация базовых классов агентов и координатора (4 ч)

Действия

1. Создать абстрактный класс Agent:
   • __init__(self, agent_id: str, weight: float, initial_opinion: str)
   • async def vote(self, options: list[str]) -> str — возвращает выбранный вариант (может быть детерминированным).
2. Создать класс Coordinator:
   • хранит список агентов, их веса, общий вес.
   • метод async def run_vote(self, options: list[str]) -> dict:
     • отправляет START_VOTE
     • собирает голоса с таймаутом (например, 5 секунд)
     • вычисляет результат.
3. Реализовать простую симуляцию конфликта: агенты с разными предпочтениями (например, 3 агента за A, 2 за B, 1 за C, веса разные).

Ожидаемый результат этапа работающая пара «агент–координатор», проводящая один раунд голосования с фиксированным набором опций.

Этап 3: Реализация механизма взвешенного голосования и подсчёта (3 ч)

Действия

1. Написать функцию def weighted_vote(tallies: dict[str, float], total_weight: float, threshold: float = 0.5) -> tuple[str | None, bool]:
   • возвращает победителя (или None) и флаг достижения консенсуса.
2. Интегрировать в Coordinator.run_vote:
   • аккумулировать веса по выбранным вариантам,
   • проверять превышение порога.
3. Для случая weighted average (если варианты числовые) — реализовать вторую функцию.

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

Этап 4: Обработка конфликтов и повторные раунды (4 ч)

Действия

1. Модифицировать Coordinator.run_vote для поддержки нескольких раундов (max_rounds=3).
2. Добавить логику retry: если консенсус не достигнут, уведомить агентов и запустить новый раунд.
3. Реализовать стратегию «forced choice»: после исчерпания раундов выбрать вариант с максимальным суммарным весом.
4. Добавить возможность для агента менять свой голос между раундами (опционально — если хотим динамическое поведение). Для простоты разрешим менять.

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

Этап 5: Тестирование и демонстрация (4 ч)

Действия

1. Написать тесты (pytest) для:
   • простого консенсуса с первого раунда,
   • конфликта, требующего 2+ раундов,
   • ситуации, когда консенсус не достигается — проверка forced choice,
   • равных весов у лидеров.
2. Создать сценарий demo.py, который:
   • создаёт 5 агентов с весами [0.3, 0.25, 0.2, 0.15, 0.1] и разными мнениями,
   • запускает голосование по трём вариантам,
   • выводит протокол раундов и финальное решение.
3. Добавить лог-вывод в формате: [Round 1] Agent A voted X (weight 0.3), etc.

Ожидаемый результат этапа успешный прогон тестов и демонстрационный скрипт, показывающий разрешение конфликта.

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

• Протокол голосования описан в README и docstring.
• Реализованы классы Agent и Coordinator с полным интерфейсом.
• Механизм взвешенного голосования корректен: сумма весов за победителя > 50% (если есть консенсус).
• Обработка конфликтов: не более 3 раундов, принудительное завершение с выбором лидера по суммарному весу.
• Тесты покрывают минимум три сценария: быстрый консенсус, конфликт с несколькими раундами, отсутствие консенсуса.
• Код проходит статический анализ (pylint с оценкой ≥8.0/10) и форматирован black.
• Стек технологий использован согласно разделу 3.
• Демонстрация запускается из командной строки и выводит читаемый лог.

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

• Основной артефакт один файл consensus.py (или небольшой пакет) с классами Agent, Coordinator и вспомогательными функциями.
• Содержимое файла импорты, класс Agent (метод vote), класс Coordinator (метод run_vote), функция weighted_vote, асинхронные хендлеры.
• Сопроводительные файлы test_consensus.py (минимум 5 тестовых кейсов), README.md с описанием протокола, demo.py.
• Опционально диаграмма последовательности (например, в Mermaid) в README.

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

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

Примечание: для первого выполнения возможен запас +20% (≈22 ч) на отладку асинхронного взаимодействия.

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

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

• Я проверил, что при консенсусе сумма весов победившего варианта строго превышает 50% от общего веса всех агентов.
• Я убедился, что после конфликта происходит не более 3 раундов, и в случае неудачи применяется forced choice.
• Я написал тесты, которые запускаются одной командой pytest -v и проходят.
• Я просмотрел лог демонстрации и убедился, что он читаем и содержит номера раундов, варианты, веса.
• Я задокументировал протокол сообщений в README и добавил пример вызова.