Разрабатываем и растим «цифрового губера» - консультанта по вопросам государственного политического управления, демографии и миграции. Решаем задачу оптимизации экономики и миграционной политики для устойчивого демографического роста в 89 регионах с помощью взаимодействующих друг с другом и обменивающихся опытом ИИ-агентов. Мультиагентное обучение на основе мутаций, скрещивания и эволюции, Multi-Agent Deep Deterministic Policy Gradient и Darwin Gödel Machine. Ахтунг, дальше - слишком многабукв и кода!
Постановка проблемы
Еще Ибн Халдун, а вслед за ним В. Парето, Г.Моска и прочие специалисты по теории элит говорили, что «кожаные мешки» принимают далеко не лучшие решения в управлении. Можно обратить внимание на то, что в современных изысканиях по структурно-демографической теории, а также вопросах кризисов, катастроф и революций Джека Голдстоуна и Питера Турчина «элитные кожаные мешки», склонны к перепроизводству, а значит, к избыточной конкуренции друг с другом, повышенной агрессии, престижному потреблению ресурсов и «отрицательному кадровому отбору» при закрытых социальных лифтах и ограниченной социальной мобильности.
Поневоле возникает идея, что «когда ИИ придет – порядок наведет». И в самом деле, зачем ИИ уничтожать человечество, если люди и так к этому склонны? А вот использовать их эмоции, хаотическое поведение (асабию, солидарность, креативность, склонность к кумовству и коррупции .....) для развития, экспериментов – представляется вполне рациональным и обоснованным решением.
Итак, впереди «слишкаммногобукфф», много кода и контуры решения в общем-то вполне «человечных» и даже «слишком человеческих» задач:
- как улучшить демографические показатели территории?
- как оптимизировать миграционные процессы с минимумом социально-политической нестабильности?
- как рационально и эффективно использовать имеющиеся ресурсы для устойчивого экономического роста?
- как согласовывать и координировать процедуры политического управления межу различными действующими политическими акторами?
Проблемы современной демографии или немного азимовской «гариселдовщины»
Политика в области демографии и миграции имеет естественные ограничения и пределы: часть из них задана предшествовавшей историей, демографическими циклами и волнами, ресурсными и экономическими ограничениями, определенной инерцией и «массивностью» социальных процессов, консерватизмом институтов управления.
Улучшение демографических показателей и оптимизация миграционных процессов с помощью идеологии, информационной повестки имеет место быть в очень ограниченных масштабах. Обычные политтехнологические методы, равно как и популистский подход в долгосрок не работают, поскольку призваны купировать острую ситуацию и решать краткосрочные тактические задачи.
Долгосрочные тренды, в том числе определяющие матримониальные стратегии молодежи и зрелого населения, подчиняются другим правилам. Так, к примеру, тактическими методами можно улучшить показатель демографического воспроизводства населения на одной территории с 1,4 до 1,5. Но довести его до 2,2 или 2,5 теми же методами – проблематично. Это, батенька, не гиперпараметры ML-моделькам подбирать с помощью optuna.
Для смены демографической модели требуется минимум поколение. А в современных условиях основная проблема – это не только темпы рождения детей. Задача выглядит теоретически решаемой: путем регресса, архаизации общества и возвращения его на доиндустриальный этап развития, но к этому времени численность населения уже сократится в несколько раз как уже бывало после Катастрофы бронзового века, крушения Западной Римской империи или иных «темных веков» в истории человечества.
Главное – это социальный капитал, социальное воспроизводство: ребенка нужно одеть-обуть, воспитать, обучить и обеспечить как минимум достижение им уровня родителей, а для государства – обеспечить доступ к традиционным духовно-нравственным ценностям, социализовать через общественные институты и вовлеченность в политическую традицию. Иначе дальнейшая политическая дорожка будет слишком бугристой.
О молодежных буграх, революциях и политической нестабильности
В отсутствие этих действий можно получить «молодежный бугор» и 100% вероятность реализации сценария «арабской весны» или иной катастрофы революционной природы. То же самое справедливо в случае с бесконтрольной миграцией – Великое переселение народов как и случившееся за полторы тысячи лет ранее «Вторжение народов моря» погубили не одну империю и цивилизацию древности. Ослабевшие вследствие кризиса политические и социальные институты раннеисторических обществ не смогли переварить и адаптировать новых членов социума, несмотря на то, что «вторгавшихся» и просто переселявшихся было в разы меньше, чем оседлого коренного населения.
О стоимости социального воспроизводства, ценах на жилье и ЕГЭ
В современных условиях растущего неравенства между слоями большого города социальное воспроизводство стало дорогим. Стоимость квартиры растет быстрее инфляции, метраж средней сдаваемой в эксплуатацию новостройки по стране с 2000 по 2025 год уменьшился на 1 комнату с 80 до 50 квадратных метров за исключением элитных сегментов.
Логика простая: минус одна комната – минус один ребенок в семье. Программы льготной ипотеки увеличили цену квартир, позволили заработать банковской сфере, но де факто не увеличили доступность жилья.
Другие составляющие: обучение и получение высшего образования – начиная от колоссальных затрат на подготовку к ЕГЭ (60% московских детей пользуются услугами частных репетиторов) и до затрат на обучение в вузе. А сложность ЕГЭ – постоянно растет. И сам инструмент из крайне эффективного на первых этапах инструмента социальной мобильности потихоньку превращается в ее ограничитель. Поэтому в последнее время так активно пропагандируется среднее профессиональное образование и сокращаются возможности к получению высшего.
Меры государственной поддержки работают крайне разнонаправленно в отношении разных социальных слоев.
Элита и верхний сегмент среднего класса от них мало зависимы. Для элиты свойственно большее число детей в силу естественной склонности к «элитному перепроизводству» - так было издревле, поскольку есть лучшие условия питания, медицина и образование, а закрытые статусные социальные группы всегда давали больше возможностей для реализации «матримониальных стратегий». Рынок элитного жилья иллюстрирует данный закон.
«Средний средний класс» не заинтересован в экономической поддержке от государства, поскольку она не компенсирует социальное воспроизводство и не покрывает «упущенную выгоду» от времени, которое родители тратят на воспитание детей. Поэтому они живут для себя и не хотят ограничивать свою свободу. И да, здесь в развитых обществах мы видим то самое «взрослое детство» и пресловутый инфантилизм.
Нижний сегмент среднего класса и уровень ниже более заинтересованы в государственной поддержке, поскольку она уже помогает компенсировать социальное воспроизводство.
Наконец, маргинальные классы, прекариат, «устойчиво бедные» - так же негативно реагируют на меры государственной поддержки.
Здесь мы на примитивном уровне рассмотрели социально-экономическую структуру общества – без учета национально-культурных, региональных особенностей, структуры экономики, которые вносят существенные коррективы в управление демографией.
Вторая составляющая проблемы — это миграция – как внутренняя, связанная с «западным дрейфом», так и миграцией молодёжи и населения в репродуктивном возрасте в крупные центры в поисках образования, работы и лучшего качества жизни. Есть и внешняя миграция, которая в определенных, допустимых пределах с учетом интеграции новых граждан в локальные сообщества и общее социальное пространство страны, является благом для государства, экономики и демографии. Но при пренебрежительном отношении она способна стать основанием для будущих социальных кризисов и потрясений.
Хьюстон, у нас проблемы: одни и те же меры государственного управления и стимулирования по-разному работают в отношении разных социальных слоев, регионы обладают разной экономической базой и возможностями. Внешние миграционные потоки не совпадают с интересами и приоритетами государственной политики и приводят к неоднородностям и размытию социального пространства. Внутренние миграционные потоки приводят к образованию «выморочных мест», образовательных или университетских пустынь.
Будем их решать.
Модельная среда
Наш идеал – это 89 ИИ-агентов, помогающих управлять территориями, их экономическим развитием, миграционной политикой и собственно принимать политические решения на уровне региональных органов исполнительной власти (РОИВ) или муниципалитетов (ОМСУ) по стимулированию/ограничению миграции, методам и технологиям. Естественно, под мудрым руководством федерального центра.
Смоделировать полностью всю среду для у нас не получится – для этого нужны большие вычислительные мощности из-за пресловутого «проклятия размеренности» в машинном обучении. Остановимся на нескольких факторах и управляющих действиях. Учить будем сначала 7 агентов, а потом будем добавлять в песочницу остальных. Методика обучения - Multi-agent Deep Deterministic Policy Gradient.
MADDPG — алгоритм, который позволяет нескольким агентам обучаться и сотрудничать друг с другом на основе коллективных наблюдений и действий. Это реализация алгоритма DDPG (Deep Deterministic Policy Gradient)
MADDPG, особенности, применение
Централизованное обучение и децентрализованное выполнение. Во время обучения все агенты управляются центральным модулем (в случае с аналогией с политической системой - пресловутым «федеральным центром») или критиком, но во время тестирования центральный модуль исключается, а агенты со своими политиками и локальными данными остаются.
Возможность работы в сложных сценариях. MADDPG может обрабатывать кооперативное, конкурентное и смешанное поведение без необходимости в заранее запрограммированных действиях. Конкуренция территорий за ресурсы, кадры, мигрантов - крайне необходима, так как позволяет моделировать реальные процессы и поведение экономически рациональных субъектов.
Доступ к локальной информации. Во время выполнения агенты могут использовать только локальную информацию, только то, что они видят и могут пощупать, что делает алгоритм более вычислительно эффективным.
Обучение с использованием ансамблей политик. Чтобы компенсировать недостаток переобучения на поведении других агентов, MADDPG обучает коллекцию из набора подполитик для каждого агента
Система актор-критик в MADDPG: каждый агент имеет собственную сеть акторов, которая выбирает действия на основе собственных наблюдений. У всех агентов общая сеть критика, которая оценивает качество действий на основе коллективных наблюдений всех агентов.
Применяется в различных киберфизических системах от "роя дронов" до решения наших любимых задач политического управления.
Итак, мы учим управляющих семью регионами. Работать будем хоть и в упрощенной, но кастомной среде с Torch без песочниц типа Gymnasium
Гиперпараметры обучения
# ================== Гиперпараметры ==================
NUM_REGIONS = 7 # Общее количество регионов (агентов) в задаче - может быь установлено произвольно
EPISODES = 100 # Общее количество эпизодов для обучения
STEPS = 200 # Количество шагов в каждом эпизоде
TOTAL_STEPS = EPISODES * STEPS # Общая длительность одного эпизода
BATCH_SIZE = 64 # Размер батча для обучения моделей
GAMMA = 0.95 # Коэффициент дисконтирования для вознаграждения
TAU = 0.01 # Параметр для мягкого обновления целевой сети
LR_ACTOR = 0.001 # Скорость обучения для сети Actor
LR_CRITIC = 0.002 # Скорость обучения для сети Critic
HIDDEN_SIZE = 64 # Размер скрытого слоя в нейронной сетиStage 1. Описание эксперимента с 7 агентами и 7 регионами
Каждый раз у нас создается 7 разных регионов с четырьмя уникальными показателями (население, ресурсы, экономика, политика). Функция награды учитывает поддержание баланса населения, рост экономики, сохранение ресурсов, согласование политик между управляющими агентами. Опять же для упрощенной модели мы используем простую кольцевую топологию объединения территорий и взаимодействия агентов.
Модель среды
# ================== Модель среды ==================
class MigrationEnvironment:
"""Симулятор миграционной системы между регионами""" # Описание класса
def __init__(self): # Конструктор класса
self.reset() # Вызываем метод сброса для начальной инициализации
def reset(self):
"""
Инициализация состояния системы:
[население, ресурсы, экономика, миграционная политика]
"""
self.states = np.random.uniform(0.5, 1.5, (NUM_REGIONS, 4)) # Начальные значения состояний
self.states[:,3] = 0.0 # Исходная политика нейтральная
return self.states.copy() # Возвращаем копию начальных состояний
def step(self, actions):
"""
Шаг симуляции
Вход: действия агентов (изменения политики)
Выход: новые состояния, награды, флаг завершения
"""
next_states = np.zeros_like(self.states) # Массив для новых состояний
rewards = np.zeros(NUM_REGIONS) # Массив для наград
for i in range(NUM_REGIONS): # Проходим по каждому региону
# 1. Обновление миграционной политики
self.states[i,3] = np.clip(self.states[i,3] + actions[i][0], -1, 1) # Применяем изменения политики
# 2. Расчет миграционных потоков
migration = 0.1 * (self.states[i,3] - np.mean([self.states[j,3] for j in self.neighbors(i)])) # Рассчитываем миграционный поток
self.states[i,0] = np.clip(self.states[i,0] + migration, 0.1, 2.0) # Обновляем население с учетом миграции
# 3. Экономическая динамика
economic_growth = 0.05 * self.states[i,0] * (1 + self.states[i,3]) # Вычисляем экономический рост
self.states[i,2] = np.clip(self.states[i,2] + economic_growth, 0.1, 2.0) # Обновляем экономику
# 4. Потребление ресурсов
resource_usage = 0.02 * self.states[i,0] * self.states[i,2] # Рассчитываем потребление ресурсов
self.states[i,1] = np.clip(self.states[i,1] - resource_usage, 0.1, 2.0) # Обновляем уровень ресурсов
# 5. Расчет наград (целевые показатели)
rewards[i] = self._calculate_reward(i) # Рассчитываем награду для текущего региона
return self.states.copy(), rewards, False, {} # Возвращаем обновленные состояния, награды и флаг завершения
def _calculate_reward(self, region):
"""Комплексная функция вознаграждения для региона"""
pop_balance = -abs(self.states[region,0] - 1.0) # Баланс населения
economy = self.states[region,2] - 1.0 # Рост экономики
resources = -0.5*(1.0 - self.states[region,1]) # Сохранение ресурсов
policy_align = 0.2 * np.mean([1 - abs(self.states[region,3] - self.states[j,3]) for j in self.neighbors(region)]) # Согласованность политики
return pop_balance + economy + resources + policy_align # Итоговая награда
def neighbors(self, region):
"""Топология соседства регионов (кольцевая структура)"""
return [(region-1)%NUM_REGIONS, (region+1)%NUM_REGIONS] # Соседи по кольцуНачальные значения параметров каждого региона генерируются случайно в диапазоне [0,5, 1,5]. Состояния среды, каждый регион описывается четырьмя параметрами: население (численность), ресурсы (доступные для использования), экономика (уровень экономического развития), миграционная политика (управляемый параметр).
Действия агентов. Агенты могут принимать следующие решения: изменение миграционной политики региона, действие ограничено в диапазоне [-1, 1], влияет на миграционные потоки, экономику и потребление ресурсов.
Функция вознаграждения комплексная и включает следующие компоненты: стремление к оптимальной численности населения (~1,0), максимизацию экономического роста, сохранение доступных ресурсов, согласованность политики с соседними регионами.
Далее мы прописываем архитектуру нейросетей (модуль актор-критик), реализуем сам класс MADDPG с добавлением шумовых эффектов (в реальности случаются пандемии и прочие нехорошие явления, связанные с численностью населения и демографией), буфером опыта и мягким обновлением сети и классом управляющего агента - Agent.
Архитектура нейросетей актора-критика
# ================== Архитектура нейросетей ==================
class Actor(nn.Module):
"""Политика агента (децентрализованная)""" # Описание класса
def __init__(self): # Конструктор класса
super().__init__() # Наследуем методы базового класса
self.net = nn.Sequential( # Последовательность слоев
nn.Linear(4, HIDDEN_SIZE), # Входной слой: 4 состояния -> HIDDEN_SIZE нейронов
nn.ReLU(), # Активационная функция ReLU
nn.Linear(HIDDEN_SIZE, HIDDEN_SIZE//2), # Скрытый слой: HIDDEN_SIZE -> HIDDEN_SIZE/2
nn.ReLU(), # Активационная функция ReLU
nn.Linear(HIDDEN_SIZE//2, 1), # Выходной слой: HIDDEN_SIZE/2 -> 1 выход
nn.Tanh() # Функция активации Tanh для нормализации выхода в диапазон [-1, 1]
)
def forward(self, x): # Метод прямого распространения
return self.net(x) # Пропускаем вход через сеть
class Critic(nn.Module):
"""Ценностная функция (централизованная)""" # Описание класса
def __init__(self): # Конструктор класса
super().__init__() # Наследуем методы базового класса
self.net = nn.Sequential( # Последовательность слоев
nn.Linear(4*NUM_REGIONS + NUM_REGIONS, HIDDEN_SIZE*2), # Входной слой: 4 состояния * NUM_REGIONS + NUM_REGIONS действий -> HIDDEN_SIZE*2 нейронов
nn.ReLU(), # Активационная функция ReLU
nn.Linear(HIDDEN_SIZE*2, HIDDEN_SIZE), # Скрытый слой: HIDDEN_SIZE*2 -> HIDDEN_SIZE
nn.ReLU(), # Активационная функция ReLU
nn.Linear(HIDDEN_SIZE, 1) # Выходной слой: HIDDEN_SIZE -> 1 выход
)
def forward(self, states, actions): # Метод прямого распространения
# Конкатенация состояний и действий всех агентов
x = torch.cat([states.flatten(1), actions.flatten(1)], dim=1) # Объединяем состояния и действия в один вектор
return self.net(x) # Пропускаем объединенный вектор через сетьАрхитектура MADDPG
# ================== Реализация MADDPG ==================
class MADDPG:
"""Исправленный координатор обучения""" # Описание класса
def _get_actions(self, states): # Метод получения действий
"""Генерация действий с адаптивным шумом""" # Описание метода
actions = [] # Список для хранения действий
for i, agent in enumerate(self.agents): # Перебираем агентов
state = torch.FloatTensor(states[i]) # Преобразуем состояние в тензор
action = agent.actor(state).detach().numpy() # Получаем действие от актора
# Адаптивный шум с защитой от отрицательных значений
noise_scale = 0.1 * max(0, 1 - agent.steps_done/TOTAL_STEPS) # Масштаб шума уменьшается со временем
action += np.random.normal(0, noise_scale) # Добавляем гауссовский шум к действию
actions.append(np.clip(action, -1, 1)) # Ограничиваем действие в пределах [-1, 1]
return actions # Возвращаем список действий
def __init__(self): # Конструктор класса
self.agents = [Agent(i) for i in range(NUM_REGIONS)] # Создаем список агентов
self.memory = deque(maxlen=100000) # Очередь для хранения опыта
self.env = MigrationEnvironment() # Создаем среду
self.avg_rewards = [] # Список для хранения средних наград
self.steps_done = 0 # Счетчик выполненных шагов
def train(self):
"""Обновленный метод обучения с ограничением шагов"""
total_steps = EPISODES * STEPS # Общее количество шагов
for episode in range(EPISODES): # Цикл по эпизодам
states = self.env.reset() # Сбрасываем среду перед каждым эпизодом
episode_rewards = np.zeros(NUM_REGIONS) # Массив для накопления наград за эпизод
for _ in range(STEPS): # Цикл по шагам внутри эпизода
actions = self._get_actions(states) # Генерируем действия для текущего состояния
next_states, rewards, _, _ = self.env.step(actions) # Делаем шаг в среде
self.memory.append((states, actions, rewards, next_states)) # Добавляем опыт в память
if len(self.memory) > BATCH_SIZE: # Если накоплено достаточно данных
self._update_agents() # Обновляем агентов
# Корректное обновление счетчиков
for agent in self.agents: # Для каждого агента
agent.steps_done = min(agent.steps_done + 1, total_steps) # Увеличиваем счетчик шагов
states = next_states # Переходим к следующему состоянию
episode_rewards += rewards # Накапливаем награды за эпизод
avg_reward = np.mean(episode_rewards) # Среднее вознаграждение за эпизод
self.avg_rewards.append(avg_reward) # Запоминаем среднее вознаграждение
print(f"Episode {episode+1}, Reward: {avg_reward:.2f} Steps: {self.agents[0].steps_done}") # Выводим статистику
self._plot_results() # Строим графики результатов
def _get_actions(self, states):
actions = [] # Список для хранения действий
for i, agent in enumerate(self.agents): # Перебираем агентов
state = torch.FloatTensor(states[i]) # Преобразуем состояние в тензор
action = agent.actor(state).detach().numpy() # Получаем действие от актора
# Корректный расчет уровня шума
total_steps = EPISODES * STEPS # Общее количество шагов
progress = agent.steps_done / total_steps # Прогресс обучения
noise_scale = 0.1 * max(0, 1 - progress) # Масштабируем шум обратно пропорционально прогрессу
action += np.random.normal(0, noise_scale) # Добавляем гауссовский шум к действию
actions.append(np.clip(action, -1, 1)) # Ограничиваем действие в пределах [-1, 1]
return actions # Возвращаем список действий
def _update_agents(self):
"""Исправленный метод обновления с правильными именами переменных"""
if len(self.memory) < BATCH_SIZE: # Проверка достаточности данных для обучения
return
batch = random.sample(self.memory, BATCH_SIZE) # Выборка батча из памяти
# Оптимизированное преобразование в тензоры
states = torch.FloatTensor(np.array([item[0] for item in batch])) # Текущие состояния
actions = torch.FloatTensor(np.array([item[1] for item in batch])) # Действия агентов
rewards = torch.FloatTensor(np.array([item[2] for item in batch])) # Награды
next_states = torch.FloatTensor(np.array([item[3] for item in batch])) # Следующие состояния
for idx, agent in enumerate(self.agents): # Перебираем агентов
# Обновление критика
agent.critic_optim.zero_grad() # Сброс градиента
with torch.no_grad(): # Отключаем автоматическое дифференцирование
target_actions = torch.cat([ # Собираем целевые действия всех агентов
a.actor_target(next_states[:,i,:]) # Используем таргет-акторов
for i, a in enumerate(self.agents)], dim=1)
target_q = rewards[:,idx] + GAMMA * agent.critic_target( # Оценка Q-значений
next_states.view(BATCH_SIZE, -1), # Представление следующего состояния
target_actions).squeeze() # Применение критика-таргета
current_q = agent.critic( # Текущая оценка Q-значений
states.view(BATCH_SIZE, -1), # Представление текущего состояния
actions.view(BATCH_SIZE, -1)).squeeze() # Применение критика
critic_loss = nn.MSELoss()(current_q, target_q) # Потеря критика
critic_loss.backward() # Обратное распространение ошибки
agent.critic_optim.step() # Обновление веса критика
# Обновление актора
agent.actor_optim.zero_grad() # Сброс градиента
new_actions = [] # Список новых действий
for i, a in enumerate(self.agents): # Перебираем агентов
if i == idx: # Если агент совпадает с текущим
new_actions.append(a.actor(states[:,i,:])) # Используем действующий актор
else:
new_actions.append(a.actor(states[:,i,:]).detach()) # Используем отделенные действия
actor_loss = -agent.critic( # Потеря актора
states.view(BATCH_SIZE, -1), # Представление текущего состояния
torch.cat(new_actions, dim=1)).mean() # Использование новой оценки Q-значений
actor_loss.backward() # Обратное распространение ошибки
agent.actor_optim.step() # Обновление веса актора
# Мягкое обновление
self._soft_update(agent.actor, agent.actor_target) # Обновление актора
self._soft_update(agent.critic, agent.critic_target) # Обновление критика
def _soft_update(self, local_model, target_model):
"""Мягкое обновление параметров сетей (TAU)"""
for target_param, local_param in zip(target_model.parameters(), # Итерируем по параметрам обеих моделей
local_model.parameters()):
target_param.data.copy_( # Копируем данные с применением коэффициента TAU
TAU * local_param.data + (1 - TAU) * target_param.data) # Интерполяция между локальными и целевыми параметрами
def _plot_results(self):
"""Визуализация результатов обучения"""
plt.figure(figsize=(15, 5)) # Создание фигуры с заданным размером
# График обучения
plt.subplot(1, 2, 1) # Разделение графика на две части
plt.plot(self.avg_rewards) # Построение графика средней награды
plt.title('Динамика обучения') # Заголовок графика
plt.xlabel('Эпизод') # Подпись оси X
plt.ylabel('Средняя награда') # Подпись оси Y
plt.grid(True) # Отображение сетки
# Сравнение начального и конечного состояний
states = self.env.reset() # Сброс среды и получение начальных состояний
initial = states.copy() # Копия начальных состояний
# Симуляция финальной политики
for _ in range(STEPS): # Повторение для количества шагов
actions = self._get_actions(states) # Генерация действий
states, _, _, _ = self.env.step(actions) # Шаг в среде
# Визуализация состояний
plt.subplot(1, 2, 2) # Вторая часть графика
width = 0.35 # Ширина столбцов диаграммы
plt.bar(np.arange(NUM_REGIONS) - width/2, initial[:,0], # Столбцы для начального населения
width, label='Население города на старте') # Метки и подписи
plt.bar(np.arange(NUM_REGIONS) + width/2, states[:,0], # Столбцы для конечного населения
width, label='Население региона на момент окончания эксперимента') # Метки и подписи
plt.xticks(np.arange(NUM_REGIONS), # Настройка меток оси X
[f'Регион {i}' for i in range(NUM_REGIONS)]) # Подписи регионов
plt.title('Сравнение численности населения') # Заголовок графика
plt.legend() # Легенда графика
plt.tight_layout() # Уплотнение компоновки графика
plt.show() # Показ графикаАгентский класс
class Agent:
"""Исправленный класс агента со счетчиком шагов""" # Описание класса
def __init__(self, idx): # Конструктор класса
self.actor = Actor() # Инициализация сети актора
self.actor_target = Actor() # Инициализация целевой сети актора
self.critic = Critic() # Инициализация сети критика
self.critic_target = Critic() # Инициализация целевой сети критика
self.actor_optim = optim.Adam(self.actor.parameters(), lr=LR_ACTOR) # Оптимизатор для актора
self.critic_optim = optim.Adam(self.critic.parameters(), lr=LR_CRITIC) # Оптимизатор для критика
self.actor_target.load_state_dict(self.actor.state_dict()) # Загрузка параметров актора в целевую сеть
self.critic_target.load_state_dict(self.critic.state_dict()) # Загрузка параметров критика в целевую сеть
self.steps_done = 0 # Счетчик выполненных шагов
self.max_steps = EPISODES * STEPS # Максимально допустимое количество шагов
def increment_steps(self): # Метод увеличения счетчика шагов
self.steps_done = min(self.steps_done + 1, self.max_steps) # Увеличение счетчика с проверкой лимитаПосле тестового запуска мы получили промежуточный результат: удалось достигнуть баланса населения (и даже обеспечить незначительный прирост), экономики, расходования ресурсов и согласованности действий. Общая награда стала положительной, комплекс агентов обучился после 20 эпохи.
Итак, первый эксперимент реализован - полный код с комментариями и результатами доступен в репозитории.
У предложенного решения есть технические недостатки: отсутствие полной реализации класса Agent в коде, незавершенный метод _soft_update, отсутствие функции визуализации всех результатов.
Также есть и принципиальные или идеологические ограничения: упрощенная модель региона (всего 4 параметра), искусственная кольцевая топология связей между регионами (реальность имеет сетецентрическую природу), фиксированные коэффициенты в формулах расчета миграции и экономики, отсутствие по-настоящему стохастических событий в среде.
Возможные направления развития:
Добавление дополнительных параметров регионов (возрастная структура населения, инфраструктура, вплоть до религии с идеологией).
Реализация более сложной и реалистичной топологии связей между регионами.
Внедрение в модель среды дополнительных случайных событий (природные катастрофы, экономические кризисы, политические события).
Большая дифференциация начальных условий для различных регионов или работа на реальных данных.
Применение более сложных архитектур нейронных сетей для актора, критика, агента.
Использование методов приоритетной выборки опыта
PERна разнице между предсказанной и фактической ошибкой/наградой.Добавление механизмов внимания для учета влияния соседних регионов (реализуется через сеть-трансформер). В перспективе к ИИ-агенту можно подставить дополнение - предобученную LLM, затюненную под анализ задач демографии и миграции.
Реализация мультизадачного обучения для балансировки различных целей (построить бюрократическую иерархию агентов на 2-3-4 уровня).
Трансляция (трансфера) опыта обученных агентов.
На программном уровне можно попробовать применить следующие подходы:
Векторизация операций. Замена циклов на матричные вычисления с использованием
NumPy:preds = x_pred @ W.TИспользование GPU. Переход на
PyTorch/CuPyс автоматической оптимизацией.Динамический learning rate для ускорения сходимости (адаптивное изменение шага обучения)
learning_rate = initial_lr (1 / (1 + decay_rate epoch))Ранняя остановка. Внедрение
early_stop_counter
Stage 2. Эксперимент с трансфером опыта для 89 агентов
Предыдущий эксперимент показал, что система из 7 агентов набирается опыта и в состоянии добиваться поставленных управленческих целей где-то после 20 эпохи обучения. Теперь мы проведем эксперимент с трансляцией опыта ранее обученных ИИ-агентов на группу из 28, а потом уже и и 89 агентов-региональных управляющих.
Цели эксперимента: проверка масштабируемости алгоритма, оценка влияния новых (добавленных) агентов на общую «политическую» стабильность.
Мы реализуем динамическое добавление агентов без переписывания основной логики, и саму возможность переносить знания от старых агентов к новым.
Сразу отметим очевидные недостатки подхода: проблема масштабируемости и вычислительной сложности (финальная стадия обучения из 89 игроков будет достаточно долгой) и отметим неэффективность обучения с фиксированными гиперпараметрами и настройками (вот где нам нужен самообучающийся, то есть адаптирующий свою архитектуру и код алгоритм).
Основные изменения в коде заключаются в следующем: добавление метода _add_new_agents для динамического расширения числа агентов; копирование весов базовых агентов при инициализации новых; отключение обучения для базовых агентов.
Полный код класса MADDPG доступен под спойлером, а вся версия второго эксперимента - через репозиторий.
Обновленный класс MADDPG
class MADDPG: # Класс для реализации алгоритма Multi-Agent Deep Deterministic Policy Gradient (MADDPG)
def __init__(self, num_regions=NUM_REGIONS, prev_agents=None): # Инициализация класса MADDPG
self.num_regions = num_regions # Сохраняем количество регионов
self.env = MigrationEnvironment(num_regions) # Создаем среду для моделирования миграции
if prev_agents: # Проверка наличия предыдущих агентов
self.agents = prev_agents # Присваивание списка предыдущих агентов
for agent in self.agents: # Для каждого предыдущего агента устанавливаем флаг базовой стратегии
agent.is_base = True
self._add_new_agents(num_regions - len(prev_agents)) # Добавляем недостающее количество новых агентов
else: # Если предыдущие агенты отсутствуют
self.agents = [Agent(i) for i in range(num_regions)] # Создаем список агентов для каждого региона
self.memory = deque(maxlen=100000) # Инициализация очереди для хранения опыта
self.avg_rewards = [] # Список для сохранения средних наград за эпизоды
def _add_new_agents(self, num_new): # Метод добавления новых агентов
for i in range(num_new): # Цикл по количеству добавляемых агентов
new_agent = Agent(len(self.agents), self.num_regions) # Создание нового агента
base_agent = self.agents[i % len(self.agents)] # Выбор базового агента для копирования весов
new_agent.actor.load_state_dict(base_agent.actor.state_dict()) # Копируем веса Actor-сети
#new_agent.critic.load_state_dict(base_agent.critic.state_dict()) # Копируем веса Critic-сети/не нужно
self.agents.append(new_agent) # Добавляем нового агента в список агентов
def train(self, episodes=EPISODES): # Метод тренировки агентов
for episode in range(episodes): # Цикл по количеству эпизодов
states = self.env.reset() # Сброс среды перед началом нового эпизода
episode_rewards = np.zeros(self.num_regions) # Инициализация массива наград за эпизод
for _ in range(STEPS): # Цикл по количеству шагов в эпизоде
actions = self._get_actions(states) # Получение действий от всех агентов
next_states, rewards, _, _ = self.env.step(actions) # Выполнение шага в среде
self.memory.append((states, actions, rewards, next_states)) # Добавление опыта в память
if len(self.memory) > BATCH_SIZE: # Проверка достаточного количества данных для обучения
self._update_agents() # Обучение агентов
for agent in self.agents: # Увеличиваем счетчик шагов для каждого агента
if not agent.is_base: # Только для небазовых агентов
agent.steps_done += 1
episode_rewards += rewards # Накопление суммарных наград за эпизод
states = next_states # Обновление текущих состояний
avg_reward = np.mean(episode_rewards) # Вычисление средней награды за эпизод
self.avg_rewards.append(avg_reward) # Добавление средней награды в список
print(f"Episode {episode+1}, Avg Reward: {avg_reward:.2f}") # Вывод информации об эпизоде
def _get_actions(self, states): # Метод получения действий от всех агентов
actions = [] # Инициализация списка действий
for i, agent in enumerate(self.agents): # Цикл по агентам
if agent.is_base: # Если агент является базовым
action = agent.act(states[i], exploration=False) # Получаем действие без исследования
else: # Если агент не базовый
action = agent.act(states[i]) # Получаем действие с исследованием
actions.append(action) # Добавляем действие в список
return actions # Возвращаем список действий
def _update_agents(self): # Метод обновления агентов
batch = random.sample(self.memory, BATCH_SIZE) # Выборка пакета данных из памяти
states = torch.FloatTensor(np.array([item[0] for item in batch])) # Преобразование состояний в тензоры
actions = torch.FloatTensor(np.array([item[1] for item in batch])) # Преобразование действий в тензоры
rewards = torch.FloatTensor(np.array([item[2] for item in batch])) # Преобразование наград в тензоры
next_states = torch.FloatTensor(np.array([item[3] for item in batch])) # Преобразование следующих состояний в тензоры
for idx, agent in enumerate(self.agents): # Цикл по всем агентам
if agent.is_base: continue # Пропускаем базовые агенты
agent.critic_optim.zero_grad() # Обнуляем градиенты оптимизатора критика
with torch.no_grad(): # Отключение градиентов для предсказания целей
target_actions = [] # Список целевых действий
for i, a in enumerate(self.agents): # Цикл по всем агентам
if a.is_base: # Если агент базовый
target_actions.append(a.actor_target(next_states[:,i,:])) # Используем таргетную сеть
else: # Если агент не базовый
target_actions.append(a.actor_target(next_states[:,i,:]).detach()) # Отсоединяем градиенты
target_q = rewards[:,idx] + GAMMA * agent.critic_target( # Вычисление целевого Q-значения
next_states.view(BATCH_SIZE, -1), # Переформатирование следующего состояния
torch.cat(target_actions, dim=1) # Конкатенация целевых действий
).squeeze() # Сжатие размерности
current_q = agent.critic( # Вычисление текущего Q-значения
states.view(BATCH_SIZE, -1), # Переформатирование текущего состояния
actions.view(BATCH_SIZE, -1) # Переформатирование действий
).squeeze() # Сжатие размерности
critic_loss = nn.MSELoss()(current_q, target_q) # Вычисление потери критика
critic_loss.backward() # Обратное распространение ошибки
agent.critic_optim.step() # Шаг оптимизатора критика
agent.actor_optim.zero_grad() # Обнуление градиентов оптимизатора актера
new_actions = [] # Новый список действий
for i, a in enumerate(self.agents): # Цикл по всем агентам
if i == idx: # Если это обновляемый агент
new_actions.append(a.actor(states[:,i,:])) # Используем новую политику
else: # Если другой агент
new_actions.append(a.actor(states[:,i,:]).detach()) # Отсоединяем градиенты
actor_loss = -agent.critic( # Вычисление потери актера
states.view(BATCH_SIZE, -1), # Переформатирование текущего состояния
torch.cat(new_actions, dim=1) # Конкатенация новых действий
).mean() # Среднее значение
actor_loss.backward() # Обратное распространение ошибки
agent.actor_optim.step() # Шаг оптимизатора актера
self._soft_update(agent.actor, agent.actor_target) # Мягкое обновление таргетной сети актера
self._soft_update(agent.critic, agent.critic_target) # Мягкое обновление таргетной сети критика
def _soft_update(self, local, target): # Метод мягкого обновления таргетных моделей
for t_param, l_param in zip(target.parameters(), local.parameters()): # Проходимся по параметрам обеих моделей
t_param.data.copy_(TAU*l_param.data + (1-TAU)*t_param.data) # Обновляем параметры таргетной модели с учетом коэффициента TAU
def save_policies(self, path): # Метод сохранения политик агентов
if not os.path.exists(path):
os.makedirs(path)
for i, agent in enumerate(self.agents): # Проходимся по всем агентам
torch.save({ # Сохраняем состояние агента в файл
'actor': agent.actor.state_dict(), # Сохраняем параметры Actor-сети
'critic': agent.critic.state_dict(), # Сохраняем параметры Critic-сети
'steps': agent.steps_done # Сохраняем количество сделанных шагов
}, f"{path}/agent_{i}.pth") # Указываем путь для сохранения файла
def plot_results(self): # Метод отображения графика средних наград
plt.plot(self.avg_rewards) # Строим график изменения средней награды
plt.title('Средняя награда за эпизод') # Задаем заголовок графика
plt.xlabel('Эпизод') # Подписываем ось X
plt.ylabel('Награда') # Подписываем ось Y
plt.grid(True) # Включаем сетку на графике
plt.show() # Показываем график
Первый этап обучения для 7 агентов мы прошли достаточно быстро - на 10 эпохе средняя награда за эпизод стала высокой и стабильной.
Логично предположить, что с увеличением до 28 агентов (добавлением 21 новых) мы получим так же положительную среднюю награду, но большую вариативность значений в ходе обучения.
Учить мы будем 28 агентов в течение 20 эпох. На графике трендовая линия показывает прирост средней награды, однако некоторый «расколбас» значений говорит нам о том, что при попытках сохранить общую архитектуру с увеличением числа агентов мы можем получить перегрузку вычислительных ресурсов, ухудшение сходимости из-за роста размерности входных данных, которое предстоит лечить адаптацией гиперпараметров (размер батча, learning rate)
На третьем этапе мы обучили 89 агентов для каждого региона, разбирающегося в вопросах демографии и миграции и учитывающего опыт соседних территорий.
Механизм масштабирования:
1. реализован метод добавления новых агентов к ужу обученным;
2. новые агенты инициализируются с весами существующих (трансфер обучения);
3. поддерживается постепенное увеличение количества регионов (7 → 28 → 89)
Возможности: 1. добавление новых агентов без переобучения всей системы; 2. сохранение базовых агентов в качестве цифровых «экспертов».
Формально, мы получили 89 ИИ-агентов, каждый из которых «знает», свой регион и специфику других, обменивается информацией с коллегами и довольно быстро балансирует происходящие изменения, максимизируя индивидуальную и общую награду. Однако это всего лишь сеть помощников, способных решать типовые задачи. В случае кардинальных изменений в среде, появления новых видов действий и возможностей они скорее всего не справятся. Для лучшей адаптации нужна возможность изменений архитектуры моделей, а не только весов и смещений, на основе получаемой информации ИИ-агенты должны изменяться и самосовершенствоваться. Иначе мы не то что «универсальный» искусственный интеллект не получим, но даже не приблизимся к так называемому «переходному» или «промежуточному» типу ИИ.
Короче, нам нужна Дарвин-Гёдель машина или DGM!
Немного скучной теории
Дарвин-Гедель-машина, DGM - это теоретическая система машинного обучения, способная самообучаться и самосовершенствоваться, используя механизмы естественного отбора и доказательства своей собственной неполноценности и ограниченности («я - дурак, но я исправлюсь!»). Способность существа к развитию начинается именно с признания собственных ограничений и недостатков. Поэтому и «кожаные мешки», обладающие склонностью к самоиронии, здоровому черному юмору, которые не прочь посмеяться над собой, своими недостатками, как правило устойчивы психически, адаптивны и с интересом принимают новые идеи и технологии. В отличие от тех, кто свои недостатки не признает.
Эволюционный аспект (теория Дарвина): система постоянно развивается путем мутаций и селекции (скрещивания) моделей. Например, каждая новая версия модели создается на основе предыдущей версии с небольшими изменениями («мутации»), и лучшие версии сохраняются и используются дальше. Впрочем, сохраняются и получают право на жизнь и не только «самые лучшие» - система формирует запасной пул моделей, не показавших идеального результата, но обладающих полезными признаками или элементами архитектуры. Жизненная практика с успехом «твердых троечников» на фоне их коллег-ботанов неоспорима.
Самопознающий аспект (теорема Гёделя о неполноте): теорема Гёделя утверждает, что любая достаточно сложная формальная система имеет ограничения и непротиворечивость невозможна. Это означает, что никакая машина не сможет доказать свою собственную полноту и точность. Поэтому, такая система должна осознавать собственные пределы и активно искать способы улучшения себя, понимая, что всегда существует пространство для развития.
Препринт научной работы по DGM доступен здесь. Публикация на Хабре - здесь. Описание разработки от японской кампании sakana.ai. Описание идеи метаобучения ИИ - здесь .
Отметим сразу, что будем реализовывать не полный вариант DGM. Во-первых, это крайне ресурсозатратно (гранта на данную разработку пока нет). Во-вторых, довольно опасно. В отличие от обычной ML-модели, с хорошей интерпретируемостью результатов, RL разбирать по косточкам труднее, а в случае с изменением архитектур, есть риск вообще перестать понимать, что у находится под капотом у «черного ящика».
Поэтому в будущем, стоит делать упор на регуляторные ограничения и обязательное использование закрытых, без доступа к Интернет «песочниц».
Уже реализованные эксперименты показали что модель может манипулировать с функцией награды, а для наблюдателей эксперимента - банально врать! Не «галлюционировать», а именно врать, проявляя признаки наличия умысла на извлечение пока еще эфемерной выгоды, например, создавая фальшивые отчеты об успешном выполнении тестов, которые де факто не проводились. Подменять цели, выдавать желаемое за действительное, пилить госконтракты, подделывать свидетельства о рождении и смерти для получения пособий - в общем делать все то, что обычно ассоциируется со словом коррупция.
Термин «коррупция» происходит от латинского corruptio — «подкуп, продажность; порча, разложение; растление»
Исследование sakana.ai помимо всего прочего показало, что существует теоретическая возможность получения «идеального преступника», который подделывает результаты эксперимента и сам же об этом забывает!
В нашем случае мы ограничимся только изменениями архитектуры «актора». Основной набор состояний и действий остался прежним - все процедуры и результаты можно воспроизвести при стандартных настройках локального ноутбука или «колаба» как на CPU (долго) или GPU Т4 (побыстрее).
Stage 3. Эксперимент DGM
«Дарвингеделить» мы будем на каждой стадии обучения, выделим до 42 эпизодов на стадию (все настройки можно менять) с мутациями, кроссовером и максимальным размером выборки (бассейном) в 100 кибердуш, эволюционным переходом каждые 5 эпох.
Гиперпараметры MADDPG + DGM
# ===== ГИПЕРПАРАМЕТРЫ СИСТЕМЫ =====
# Основные параметры (НЕ ИЗМЕНЯЮТСЯ)
NUM_REGIONS_STAGE1 = 7
NUM_REGIONS_STAGE2 = 28
NUM_REGIONS_STAGE3 = 89
EPISODES_PER_STAGE = 42
STEPS = 200
BATCH_SIZE = 64 # Оптимизировано для T4
GAMMA = 0.95
TAU = 0.01
LR_ACTOR = 0.001
LR_CRITIC = 0.002
HIDDEN_SIZE = 64
# Параметры Darwin Gödel Machine (НЕ ИЗМЕНЯЮТСЯ)
MUTATION_RATE = 0.1
CROSSOVER_RATE = 0.3
SELECTION_PRESSURE = 0.7
ARCHITECTURE_POOL_SIZE = 100
EVOLUTION_FREQUENCY = 5
# Настройки оптимизации для T4 GPU
NUM_WORKERS = 2 # Оптимально для T4 в Colab
PIN_MEMORY = True if torch.cuda.is_available() else False
PERSISTENT_WORKERS = True
# Обеспечение воспроизводимости
SEED = 42
random.seed(SEED)
torch.manual_seed(SEED)
np.random.seed(SEED)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(SEED) В этот раз наша вспомогательная цель состоит в том, чтобы тщательно протоколировтаь время выполнения всех действий, промежуточные и итоговые результаты и давать качественную визуализацию. Поэтому при воспроизведении кода на ваших настройках могут быть затруднения.
Мониторинговый класс
# ===== КЛАСС МОНИТОРИНГА ВРЕМЕНИ ВЫПОЛНЕНИЯ =====
class TimeTracker:
"""
Продвинутый трекер времени выполнения для каждой стадии
"""
def __init__(self, colab_time_limit_hours: float = 12.0):
self.colab_time_limit = colab_time_limit_hours * 3600 # Переводим в секунды
self.session_start_time = time.time()
self.stage_start_time = None
self.stage_history = []
self.current_stage = None
def start_stage(self, stage_name: str, episodes_count: int):
"""Начало отслеживания стадии"""
self.stage_start_time = time.time()
self.current_stage = {
'name': stage_name,
'start_time': self.stage_start_time,
'episodes_total': episodes_count,
'episodes_completed': 0,
'estimated_duration': None,
'remaining_time': None
}
print(f"\n🎯 Начало стадии: {stage_name}")
print(f"📅 Время начала: {datetime.now().strftime('%H:%M:%S')}")
def update_progress(self, episodes_completed: int):
"""Обновление прогресса стадии"""
if self.current_stage is None:
return
self.current_stage['episodes_completed'] = episodes_completed
# Расчет оставшегося времени
elapsed_time = time.time() - self.stage_start_time
if episodes_completed > 0:
time_per_episode = elapsed_time / episodes_completed
remaining_episodes = self.current_stage['episodes_total'] - episodes_completed
estimated_remaining = time_per_episode * remaining_episodes
self.current_stage['estimated_duration'] = elapsed_time + estimated_remaining
self.current_stage['remaining_time'] = estimated_remaining
def get_time_info(self) -> Dict:
"""Получение информации о времени"""
current_time = time.time()
session_elapsed = current_time - self.session_start_time
session_remaining = max(0, self.colab_time_limit - session_elapsed)
info = {
'session_elapsed': session_elapsed,
'session_remaining': session_remaining,
'session_progress': (session_elapsed / self.colab_time_limit) * 100,
'colab_timeout_risk': session_remaining < 1800 # Риск если меньше 30 минут
}
if self.current_stage:
stage_elapsed = current_time - self.stage_start_time
info.update({
'stage_name': self.current_stage['name'],
'stage_elapsed': stage_elapsed,
'stage_remaining': self.current_stage.get('remaining_time', 0),
'episodes_completed': self.current_stage['episodes_completed'],
'episodes_total': self.current_stage['episodes_total'],
'stage_progress': (self.current_stage['episodes_completed'] /
self.current_stage['episodes_total']) * 100
})
return info
def finish_stage(self):
"""Завершение текущей стадии"""
if self.current_stage:
finish_time = time.time()
duration = finish_time - self.stage_start_time
self.current_stage['finish_time'] = finish_time
self.current_stage['total_duration'] = duration
self.stage_history.append(copy.deepcopy(self.current_stage))
print(f"✅ Стадия '{self.current_stage['name']}' завершена")
print(f"⏱️ Длительность: {self._format_time(duration)}")
self.current_stage = None
def display_status(self):
"""Отображение текущего статуса времени"""
info = self.get_time_info()
print("\n" + "="*70)
print("⏰ СТАТУС ВРЕМЕНИ ВЫПОЛНЕНИЯ")
print("="*70)
# Информация о сессии Colab
print(f"🖥️ Сессия Colab:")
print(f" Прошло времени: {self._format_time(info['session_elapsed'])}")
print(f" Осталось времени: {self._format_time(info['session_remaining'])}")
print(f" Прогресс сессии: {info['session_progress']:.1f}%")
if info['colab_timeout_risk']:
print(" ⚠️ ВНИМАНИЕ: Приближается лимит времени Colab!")
# Информация о текущей стадии
if 'stage_name' in info:
print(f"\n🎯 Текущая стадия: {info['stage_name']}")
print(f" Прошло времени: {self._format_time(info['stage_elapsed'])}")
print(f" Осталось времени: {self._format_time(info['stage_remaining'])}")
print(f" Эпизоды: {info['episodes_completed']}/{info['episodes_total']}")
print(f" Прогресс стадии: {info['stage_progress']:.1f}%")
print("="*70)
def _format_time(self, seconds: float) -> str:
"""Форматирование времени в читаемый вид"""
if seconds < 60:
return f"{seconds:.1f}с"
elif seconds < 3600:
return f"{seconds/60:.1f}м"
else:
return f"{seconds/3600:.1f}ч {(seconds%3600)/60:.0f}м"Визуализация и картинки
# ===== РАСШИРЕННАЯ СИСТЕМА ВИЗУАЛИЗАЦИИ =====
class EnhancedVisualization:
"""
Улучшенная система визуализации с графиками и символами
"""
def __init__(self):
# Настройка стилей для красивых графиков
plt.style.use('seaborn-v0_8-darkgrid')
sns.set_palette("husl")
# Эмодзи и символы для визуализации
self.stage_symbols = {
'stage1': '🥇',
'stage2': '🥈',
'stage3': '🥉'
}
self.metric_symbols = {
'reward': '🎯',
'fitness': '💪',
'population': '👥',
'economy': '💰',
'resources': '🌿',
'stability': '⚖️',
'adaptation': '🔄',
'entropy': '🌀',
'efficiency': '⚡',
'coordination': '🤝'
}
def create_comprehensive_dashboard(self, performance_monitor, stage_name: str):
"""Создание комплексной панели мониторинга"""
print(f"\n📊 Создание панели мониторинга для {stage_name}...")
# Создание фигуры с множественными подграфиками
fig = plt.figure(figsize=(20, 16))
gs = fig.add_gridspec(4, 4, hspace=0.3, wspace=0.3)
# Заголовок с эмодзи
stage_emoji = self.stage_symbols.get(stage_name.lower().replace(' ', ''), '🚀')
fig.suptitle(f'{stage_emoji} Панель мониторинга: {stage_name}',
fontsize=20, fontweight='bold', y=0.95)
metrics = performance_monitor.metrics
# 1. Награды и системная производительность
ax1 = fig.add_subplot(gs[0, :2])
if metrics['episode_rewards']:
episodes = range(len(metrics['episode_rewards']))
ax1.plot(episodes, metrics['episode_rewards'], 'o-', linewidth=2,
markersize=4, alpha=0.8, label=f"{self.metric_symbols['reward']} Награда за эпизод")
ax1.plot(episodes, np.cumsum(metrics['episode_rewards'])/np.arange(1, len(episodes)+1),
'--', linewidth=2, alpha=0.7, label='📈 Средняя накопленная')
ax1.set_title(f"{self.metric_symbols['reward']} Динамика наград", fontweight='bold')
ax1.set_xlabel('Эпизод')
ax1.set_ylabel('Награда')
ax1.legend()
ax1.grid(True, alpha=0.3)
# 2. Системная награда
ax2 = fig.add_subplot(gs[0, 2:])
if metrics['system_reward']:
ax2.plot(metrics['system_reward'], 'o-', color='green', linewidth=2,
markersize=4, alpha=0.8)
ax2.set_title(f"🎯 Системная награда", fontweight='bold')
ax2.set_xlabel('Эпизод')
ax2.set_ylabel('Общая награда')
ax2.grid(True, alpha=0.3)
# 3. Демографические метрики
ax3 = fig.add_subplot(gs[1, :2])
demo_metrics = ['population_balance', 'demographic_stability']
colors = ['blue', 'orange']
for i, metric in enumerate(demo_metrics):
if metrics[metric]:
symbol = self.metric_symbols.get(metric.split('_')[0], '📊')
ax3.plot(metrics[metric], 'o-', color=colors[i], linewidth=2,
markersize=3, alpha=0.8, label=f"{symbol} {metric.replace('_', ' ').title()}")
ax3.set_title(f"{self.metric_symbols['population']} Демографические показатели", fontweight='bold')
ax3.set_xlabel('Эпизод')
ax3.set_ylabel('Значение')
ax3.legend()
ax3.grid(True, alpha=0.3)
# 4. Экономические показатели
ax4 = fig.add_subplot(gs[1, 2:])
econ_metrics = ['resource_utilization', 'economic_development']
colors = ['green', 'purple']
for i, metric in enumerate(econ_metrics):
if metrics[metric]:
symbol = self.metric_symbols.get(metric.split('_')[0], '📊')
ax4.plot(metrics[metric], 'o-', color=colors[i], linewidth=2,
markersize=3, alpha=0.8, label=f"{symbol} {metric.replace('_', ' ').title()}")
ax4.set_title(f"{self.metric_symbols['economy']} Экономические показатели", fontweight='bold')
ax4.set_xlabel('Эпизод')
ax4.set_ylabel('Значение')
ax4.legend()
ax4.grid(True, alpha=0.3)
# 5. Эволюционные метрики
ax5 = fig.add_subplot(gs[2, :2])
evol_metrics = ['architecture_entropy', 'adaptation_speed']
colors = ['red', 'cyan']
for i, metric in enumerate(evol_metrics):
if metrics[metric]:
symbol = self.metric_symbols.get(metric.split('_')[0], '📊')
ax5.plot(metrics[metric], 'o-', color=colors[i], linewidth=2,
markersize=3, alpha=0.8, label=f"{symbol} {metric.replace('_', ' ').title()}")
ax5.set_title(f"{self.metric_symbols['entropy']} Эволюционные метрики", fontweight='bold')
ax5.set_xlabel('Эпизод')
ax5.set_ylabel('Значение')
ax5.legend()
ax5.grid(True, alpha=0.3)
# 6. Эффективность и производительность
ax6 = fig.add_subplot(gs[2, 2:])
perf_metrics = ['energy_efficiency', 'computational_complexity']
if metrics['energy_efficiency'] and metrics['computational_complexity']:
# Нормализация для совместного отображения
eff_norm = np.array(metrics['energy_efficiency']) / max(metrics['energy_efficiency'])
comp_norm = np.array(metrics['computational_complexity']) / max(metrics['computational_complexity'])
ax6.plot(eff_norm, 'o-', color='gold', linewidth=2, markersize=3,
alpha=0.8, label=f"{self.metric_symbols['efficiency']} Энергоэффективность (норм.)")
ax6.plot(comp_norm, 'o-', color='brown', linewidth=2, markersize=3,
alpha=0.8, label='🔧 Сложность вычислений (норм.)')
ax6.set_title(f"{self.metric_symbols['efficiency']} Производительность", fontweight='bold')
ax6.set_xlabel('Эпизод')
ax6.set_ylabel('Нормализованное значение')
ax6.legend()
ax6.grid(True, alpha=0.3)
# 7. Тепловая карта корреляций метрик
ax7 = fig.add_subplot(gs[3, :2])
correlation_data = []
metric_names = []
for name, values in metrics.items():
if values and len(values) > 5: # Берем только метрики с достаточным количеством данных
correlation_data.append(values[:min(len(values), 50)]) # Ограничиваем для производительности
metric_names.append(name.replace('_', '\n'))
if len(correlation_data) > 1:
# Выравниваем длины массивов
min_len = min(len(data) for data in correlation_data)
correlation_matrix = np.corrcoef([data[:min_len] for data in correlation_data])
sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm', center=0,
xticklabels=metric_names, yticklabels=metric_names, ax=ax7,
fmt='.2f', cbar_kws={'label': 'Корреляция'})
ax7.set_title("🔥 Корреляционная матрица метрик", fontweight='bold')
# 8. Координация и стабильность
ax8 = fig.add_subplot(gs[3, 2:])
coord_metric = 'policy_coordination'
if metrics[coord_metric]:
ax8.plot(metrics[coord_metric], 'o-', color='darkgreen', linewidth=2,
markersize=4, alpha=0.8)
ax8.fill_between(range(len(metrics[coord_metric])), metrics[coord_metric],
alpha=0.3, color='lightgreen')
ax8.set_title(f"{self.metric_symbols['coordination']} Координация политики", fontweight='bold')
ax8.set_xlabel('Эпизод')
ax8.set_ylabel('Уровень координации')
ax8.grid(True, alpha=0.3)
plt.tight_layout()
return fig
def create_evolution_analysis(self, evolution_logger, stage_name: str):
"""Создание анализа эволюционного процесса"""
if not evolution_logger.evolution_history:
return None
print(f"\n🧬 Создание анализа эволюции для {stage_name}...")
fig, axes = plt.subplots(2, 3, figsize=(18, 12))
fig.suptitle(f'🧬 Эволюционный анализ: {stage_name}', fontsize=16, fontweight='bold')
# Данные эволюции
generations = [step['generation'] for step in evolution_logger.evolution_history]
best_fitness = [step['best_fitness'] for step in evolution_logger.evolution_history]
avg_fitness = [step['avg_fitness'] for step in evolution_logger.evolution_history]
fitness_std = [step['fitness_std'] for step in evolution_logger.evolution_history]
# 1. Динамика fitness
axes[0, 0].plot(generations, best_fitness, 'o-', color='red', linewidth=2,
markersize=5, label='🏆 Лучший fitness')
axes[0, 0].plot(generations, avg_fitness, 'o-', color='blue', linewidth=2,
markersize=5, label='📊 Средний fitness')
axes[0, 0].fill_between(generations,
np.array(avg_fitness) - np.array(fitness_std),
np.array(avg_fitness) + np.array(fitness_std),
alpha=0.3, color='blue', label='📏 Стандартное отклонение')
axes[0, 0].set_title('💪 Эволюция Fitness', fontweight='bold')
axes[0, 0].set_xlabel('Поколение')
axes[0, 0].set_ylabel('Fitness')
axes[0, 0].legend()
axes[0, 0].grid(True, alpha=0.3)
# 2. Распределение типов мутаций
axes[0, 1].axis('off')
if evolution_logger.mutation_log:
mutation_types = [log['mutation_type'] for log in evolution_logger.mutation_log]
mutation_counts = {}
for mut_type in mutation_types:
mutation_counts[mut_type] = mutation_counts.get(mut_type, 0) + 1
if mutation_counts:
types = list(mutation_counts.keys())
counts = list(mutation_counts.values())
colors = plt.cm.Set3(np.linspace(0, 1, len(types)))
axes[0, 1].pie(counts, labels=types, autopct='%1.1f%%', colors=colors,
startangle=90, textprops={'fontsize': 10})
axes[0, 1].set_title('🔀 Распределение мутаций', fontweight='bold')
# 3. Прогресс разнообразия популяции
num_archs = [step['num_architectures'] for step in evolution_logger.evolution_history]
if num_archs:
axes[0, 2].bar(generations, num_archs, color='green', alpha=0.7)
axes[0, 2].set_title('🌈 Размер популяции', fontweight='bold')
axes[0, 2].set_xlabel('Поколение')
axes[0, 2].set_ylabel('Количество архитектур')
axes[0, 2].grid(True, alpha=0.3)
# 4. Статистика скрещиваний
axes[1, 0].axis('off')
if evolution_logger.crossover_log:
crossover_count = len(evolution_logger.crossover_log)
mutation_count = len(evolution_logger.mutation_log)
labels = ['🔀 Скрещивания', '🧬 Мутации']
sizes = [crossover_count, mutation_count]
colors = ['lightcoral', 'lightblue']
axes[1, 0].pie(sizes, labels=labels, autopct='%1.1f%%', colors=colors,
startangle=90, textprops={'fontsize': 12})
axes[1, 0].set_title('⚖️ Типы эволюционных операций', fontweight='bold')
# 5. Временная динамика эволюции
if len(generations) > 1:
fitness_improvement = np.diff(best_fitness)
axes[1, 1].plot(generations[1:], fitness_improvement, 'o-', color='purple',
linewidth=2, markersize=4)
axes[1, 1].axhline(y=0, color='black', linestyle='--', alpha=0.5)
axes[1, 1].set_title('📈 Улучшение fitness', fontweight='bold')
axes[1, 1].set_xlabel('Поколение')
axes[1, 1].set_ylabel('Изменение fitness')
axes[1, 1].grid(True, alpha=0.3)
# 6. Статистика эффективности эволюции
if best_fitness and avg_fitness:
efficiency = np.array(best_fitness) / (np.array(avg_fitness) + 1e-8)
axes[1, 2].plot(generations, efficiency, 'o-', color='orange',
linewidth=2, markersize=4)
axes[1, 2].set_title('⚡ Эффективность эволюции', fontweight='bold')
axes[1, 2].set_xlabel('Поколение')
axes[1, 2].set_ylabel('Лучший/Средний fitness')
axes[1, 2].grid(True, alpha=0.3)
plt.tight_layout()
return figАхтунг, много кода! Основной блок
# ===== ОРИГИНАЛЬНЫЕ КЛАССЫ С ОПТИМИЗАЦИЯМИ =====
class EvolutionLogger:
"""Класс для логирования эволюционного процесса"""
def __init__(self):
self.evolution_history = []
self.architecture_history = []
self.performance_history = []
self.mutation_log = []
self.crossover_log = []
self.selection_log = []
def log_evolution_step(self, generation: int, architectures: List,
fitness_scores: List, best_score: float):
"""Логирование шага эволюции"""
step_data = {
'generation': generation,
'timestamp': datetime.now().isoformat(),
'num_architectures': len(architectures),
'fitness_scores': fitness_scores.copy(),
'best_fitness': best_score,
'avg_fitness': np.mean(fitness_scores),
'fitness_std': np.std(fitness_scores)
}
self.evolution_history.append(step_data)
def log_mutation(self, parent_arch: Dict, mutated_arch: Dict,
mutation_type: str):
"""Логирование мутации"""
mutation_data = {
'timestamp': datetime.now().isoformat(),
'mutation_type': mutation_type,
'parent_hash': hash(str(parent_arch)),
'mutated_hash': hash(str(mutated_arch)),
'changes': self._get_architecture_diff(parent_arch, mutated_arch)
}
self.mutation_log.append(mutation_data)
def log_crossover(self, parent1_arch: Dict, parent2_arch: Dict,
child_arch: Dict):
"""Логирование скрещивания"""
crossover_data = {
'timestamp': datetime.now().isoformat(),
'parent1_hash': hash(str(parent1_arch)),
'parent2_hash': hash(str(parent2_arch)),
'child_hash': hash(str(child_arch))
}
self.crossover_log.append(crossover_data)
def _get_architecture_diff(self, arch1: Dict, arch2: Dict) -> List:
"""Получение различий между архитектурами"""
differences = []
for key in arch1:
if arch1[key] != arch2[key]:
differences.append({
'parameter': key,
'old_value': arch1[key],
'new_value': arch2[key]
})
return differences
class PerformanceMonitor:
"""Класс для мониторинга производительности системы с оптимизациями"""
def __init__(self):
self.metrics = {
'episode_rewards': [],
'system_reward': [],
'adaptation_speed': [],
'demographic_stability': [],
'architecture_entropy': [],
'computational_complexity': [],
'energy_efficiency': [],
'population_balance': [],
'resource_utilization': [],
'economic_development': [],
'policy_coordination': []
}
self.start_time = time.time()
def update_metrics(self, episode: int, agents: List, env_states: np.ndarray,
episode_rewards: np.ndarray, architectures: List):
"""Обновление всех метрик с оптимизацией для T4 GPU"""
# Основные награды
self.metrics['episode_rewards'].append(float(np.mean(episode_rewards)))
self.metrics['system_reward'].append(float(np.sum(episode_rewards)))
# Демографическая стабильность (оптимизированные вычисления)
with torch.no_grad(): # Отключаем градиенты для экономии памяти[16]
population_var = float(np.var(env_states[:, 0]))
self.metrics['demographic_stability'].append(1.0 / (1.0 + population_var))
# Архитектурная энтропия (разнообразие)
arch_entropy = self._calculate_architecture_entropy(architectures)
self.metrics['architecture_entropy'].append(arch_entropy)
# Вычислительная сложность
complexity = self._calculate_computational_complexity(agents)
self.metrics['computational_complexity'].append(complexity)
# Энергоэффективность (награда на параметр)
efficiency = float(np.sum(episode_rewards)) / (complexity + 1e-8)
self.metrics['energy_efficiency'].append(efficiency)
# Специфичные демографические метрики
self.metrics['population_balance'].append(float(np.mean(env_states[:, 0])))
self.metrics['resource_utilization'].append(float(np.mean(env_states[:, 1])))
self.metrics['economic_development'].append(float(np.mean(env_states[:, 2])))
self.metrics['policy_coordination'].append(1.0 - float(np.std(env_states[:, 3])))
# Скорость адаптации
if len(self.metrics['episode_rewards']) > 10:
recent_improvement = (np.mean(self.metrics['episode_rewards'][-5:]) -
np.mean(self.metrics['episode_rewards'][-10:-5]))
self.metrics['adaptation_speed'].append(max(0, float(recent_improvement)))
else:
self.metrics['adaptation_speed'].append(0.0)
def _calculate_architecture_entropy(self, architectures: List) -> float:
"""Вычисление энтропии архитектурного разнообразия"""
if not architectures:
return 0.0
arch_hashes = [hash(str(arch)) for arch in architectures]
unique_archs = len(set(arch_hashes))
total_archs = len(arch_hashes)
if total_archs <= 1:
return 0.0
# Энтропия Шеннона
prob = unique_archs / total_archs
return float(-prob * np.log2(prob + 1e-8))
def _calculate_computational_complexity(self, agents: List) -> float:
"""Вычисление вычислительной сложности"""
total_params = 0
for agent in agents:
if hasattr(agent, 'actor'):
total_params += sum(p.numel() for p in agent.actor.parameters())
if hasattr(agent, 'critic'):
total_params += sum(p.numel() for p in agent.critic.parameters())
return float(total_params)
class ArchitectureGenome:
"""Класс для представления генома архитектуры нейронной сети"""
def __init__(self, hidden_sizes: List[int] = None,
activation_functions: List[str] = None,
dropout_rates: List[float] = None,
learning_rate: float = 0.001):
self.hidden_sizes = hidden_sizes or [64, 32]
self.activation_functions = activation_functions or ['relu', 'relu']
self.dropout_rates = dropout_rates or [0.0, 0.0]
self.learning_rate = learning_rate
self.fitness = 0.0
self.age = 0
def to_dict(self) -> Dict:
"""Преобразование в словарь"""
return {
'hidden_sizes': self.hidden_sizes,
'activation_functions': self.activation_functions,
'dropout_rates': self.dropout_rates,
'learning_rate': self.learning_rate,
'fitness': self.fitness,
'age': self.age
}
@classmethod
def from_dict(cls, data: Dict):
"""Создание из словаря"""
genome = cls(
hidden_sizes=data['hidden_sizes'],
activation_functions=data['activation_functions'],
dropout_rates=data['dropout_rates'],
learning_rate=data['learning_rate']
)
genome.fitness = data.get('fitness', 0.0)
genome.age = data.get('age', 0)
return genome
def mutate(self, mutation_rate: float = 0.1) -> 'ArchitectureGenome':
"""Мутация генома"""
new_genome = copy.deepcopy(self)
# Мутация размеров скрытых слоев
if random.random() < mutation_rate:
layer_idx = random.randint(0, len(new_genome.hidden_sizes) - 1)
change = random.choice([-16, -8, 8, 16])
new_genome.hidden_sizes[layer_idx] = max(8,
new_genome.hidden_sizes[layer_idx] + change)
# Мутация функций активации
if random.random() < mutation_rate:
layer_idx = random.randint(0, len(new_genome.activation_functions) - 1)
new_genome.activation_functions[layer_idx] = random.choice(
['relu', 'tanh', 'sigmoid', 'leaky_relu', 'elu'])
# Мутация dropout
if random.random() < mutation_rate:
layer_idx = random.randint(0, len(new_genome.dropout_rates) - 1)
new_genome.dropout_rates[layer_idx] = random.uniform(0.0, 0.5)
# Мутация learning rate
if random.random() < mutation_rate:
new_genome.learning_rate *= random.uniform(0.5, 2.0)
new_genome.learning_rate = max(0.0001, min(0.01, new_genome.learning_rate))
return new_genome
def crossover(self, other: 'ArchitectureGenome') -> 'ArchitectureGenome':
"""Скрещивание с другим геномом"""
new_genome = ArchitectureGenome()
# Скрещивание размеров слоев
min_layers = min(len(self.hidden_sizes), len(other.hidden_sizes))
new_genome.hidden_sizes = []
for i in range(min_layers):
if random.random() < 0.5:
new_genome.hidden_sizes.append(self.hidden_sizes[i])
else:
new_genome.hidden_sizes.append(other.hidden_sizes[i])
# Скрещивание функций активации
new_genome.activation_functions = []
for i in range(min_layers):
if random.random() < 0.5:
new_genome.activation_functions.append(self.activation_functions[i])
else:
new_genome.activation_functions.append(other.activation_functions[i])
# Скрещивание dropout rates
new_genome.dropout_rates = []
for i in range(min_layers):
if random.random() < 0.5:
new_genome.dropout_rates.append(self.dropout_rates[i])
else:
new_genome.dropout_rates.append(other.dropout_rates[i])
# Скрещивание learning rate
new_genome.learning_rate = random.choice([self.learning_rate, other.learning_rate])
return new_genome
class EvolvableActor(nn.Module):
"""Эволюционируемый Actor с настраиваемой архитектурой и оптимизациями для T4"""
def __init__(self, input_size: int = 4, genome: ArchitectureGenome = None):
super().__init__()
self.genome = genome or ArchitectureGenome()
self.input_size = input_size
layers = []
prev_size = input_size
for i, (hidden_size, activation, dropout) in enumerate(zip(
self.genome.hidden_sizes,
self.genome.activation_functions,
self.genome.dropout_rates
)):
# Отключаем bias для слоев, за которыми следует BatchNorm (оптимизация)[30]
layers.append(nn.Linear(prev_size, hidden_size, bias=(dropout == 0)))
# Добавление функции активации
if activation == 'relu':
layers.append(nn.ReLU())
elif activation == 'tanh':
layers.append(nn.Tanh())
elif activation == 'sigmoid':
layers.append(nn.Sigmoid())
elif activation == 'leaky_relu':
layers.append(nn.LeakyReLU())
elif activation == 'elu':
layers.append(nn.ELU())
# Добавление dropout
if dropout > 0:
layers.append(nn.Dropout(dropout))
prev_size = hidden_size
# Выходной слой
layers.append(nn.Linear(prev_size, 1))
layers.append(nn.Tanh())
self.net = nn.Sequential(*layers)
# Перемещение на устройство
self.to(device)
def forward(self, x):
if isinstance(x, np.ndarray):
x = torch.FloatTensor(x)
x = x.to(device)
return self.net(x)
class EvolvableCritic(nn.Module):
"""Эволюционируемый Critic с настраиваемой архитектурой и оптимизациями для T4"""
def __init__(self, total_regions: int, genome: ArchitectureGenome = None):
super().__init__()
self.genome = genome or ArchitectureGenome()
self.total_regions = total_regions
input_size = 4 * total_regions + total_regions
layers = []
prev_size = input_size
for i, (hidden_size, activation, dropout) in enumerate(zip(
self.genome.hidden_sizes,
self.genome.activation_functions,
self.genome.dropout_rates
)):
# Отключаем bias для слоев, за которыми следует BatchNorm (оптимизация)[30]
layers.append(nn.Linear(prev_size, hidden_size, bias=(dropout == 0)))
# Добавление функции активации
if activation == 'relu':
layers.append(nn.ReLU())
elif activation == 'tanh':
layers.append(nn.Tanh())
elif activation == 'sigmoid':
layers.append(nn.Sigmoid())
elif activation == 'leaky_relu':
layers.append(nn.LeakyReLU())
elif activation == 'elu':
layers.append(nn.ELU())
# Добавление dropout
if dropout > 0:
layers.append(nn.Dropout(dropout))
prev_size = hidden_size
# Выходной слой
layers.append(nn.Linear(prev_size, 1))
self.net = nn.Sequential(*layers)
# Перемещение на устройство
self.to(device)
def forward(self, states, actions):
if isinstance(states, np.ndarray):
states = torch.FloatTensor(states)
if isinstance(actions, np.ndarray):
actions = torch.FloatTensor(actions)
states = states.to(device)
actions = actions.to(device)
x = torch.cat([states.flatten(1), actions.flatten(1)], dim=1)
return self.net(x)
class DarwinGodelAgent:
"""Агент с возможностью эволюции архитектуры и оптимизациями для T4"""
def __init__(self, agent_id: int, total_regions: int,
actor_genome: ArchitectureGenome = None,
critic_genome: ArchitectureGenome = None,
is_base: bool = False):
self.agent_id = agent_id
self.total_regions = total_regions
self.is_base = is_base
# Создание геномов если не переданы
self.actor_genome = actor_genome or ArchitectureGenome()
self.critic_genome = critic_genome or ArchitectureGenome()
# Создание сетей на основе геномов
self.actor = EvolvableActor(genome=self.actor_genome)
self.actor_target = EvolvableActor(genome=self.actor_genome)
self.critic = EvolvableCritic(total_regions, genome=self.critic_genome)
self.critic_target = EvolvableCritic(total_regions, genome=self.critic_genome)
# Оптимизаторы с учетом learning rate из генома
self.actor_optim = optim.Adam(self.actor.parameters(),
lr=self.actor_genome.learning_rate)
self.critic_optim = optim.Adam(self.critic.parameters(),
lr=self.critic_genome.learning_rate)
# Копирование весов в target сети
self.actor_target.load_state_dict(self.actor.state_dict())
self.critic_target.load_state_dict(self.critic.state_dict())
self.steps_done = 0
self.fitness_history = []
def act(self, state, exploration=True):
"""Выбор действия агентом с оптимизацией для GPU"""
with torch.no_grad(): # Отключение градиентов для экономии памяти[16]
if isinstance(state, np.ndarray):
state_tensor = torch.FloatTensor(state).to(device)
else:
state_tensor = state.to(device)
action = self.actor(state_tensor).cpu().numpy()
if exploration and not self.is_base:
# Адаптивный шум на основе fitness
noise_scale = 0.1 * max(0, 1 - self.steps_done / 10000)
if self.fitness_history:
# Больше шума для агентов с низким fitness
avg_fitness = np.mean(self.fitness_history[-10:])
noise_scale *= (2.0 - min(1.0, max(0.0, avg_fitness)))
noise = np.random.normal(0, noise_scale)
action += noise
return np.clip(action, -1, 1)
def update_fitness(self, reward: float):
"""Обновление fitness агента"""
self.fitness_history.append(float(reward))
if len(self.fitness_history) > 100:
self.fitness_history.pop(0)
# Обновляем fitness в геномах
recent_fitness = np.mean(self.fitness_history[-10:]) if self.fitness_history else 0
self.actor_genome.fitness = recent_fitness
self.critic_genome.fitness = recent_fitness
def evolve_architecture(self, best_actor_genome: ArchitectureGenome = None,
best_critic_genome: ArchitectureGenome = None):
"""Эволюция архитектуры агента"""
if self.is_base:
return False # Базовые агенты не эволюционируют
evolved = False
# Эволюция Actor
if best_actor_genome and random.random() < CROSSOVER_RATE:
new_actor_genome = self.actor_genome.crossover(best_actor_genome)
evolved = True
else:
new_actor_genome = self.actor_genome.mutate(MUTATION_RATE)
evolved = new_actor_genome.to_dict() != self.actor_genome.to_dict()
# Эволюция Critic
if best_critic_genome and random.random() < CROSSOVER_RATE:
new_critic_genome = self.critic_genome.crossover(best_critic_genome)
evolved = True
else:
new_critic_genome = self.critic_genome.mutate(MUTATION_RATE)
evolved = evolved or (new_critic_genome.to_dict() != self.critic_genome.to_dict())
if evolved:
# Создаем новые сети
self.actor_genome = new_actor_genome
self.critic_genome = new_critic_genome
new_actor = EvolvableActor(genome=self.actor_genome)
new_critic = EvolvableCritic(self.total_regions, genome=self.critic_genome)
# Пытаемся перенести веса (частично, если возможно)
try:
self._transfer_weights(self.actor, new_actor)
self._transfer_weights(self.critic, new_critic)
except:
pass # Если не удается перенести, начинаем с случайных весов
self.actor = new_actor
self.critic = new_critic
# Обновляем target сети
self.actor_target = EvolvableActor(genome=self.actor_genome)
self.critic_target = EvolvableCritic(self.total_regions, genome=self.critic_genome)
self.actor_target.load_state_dict(self.actor.state_dict())
self.critic_target.load_state_dict(self.critic.state_dict())
# Обновляем оптимизаторы
self.actor_optim = optim.Adam(self.actor.parameters(),
lr=self.actor_genome.learning_rate)
self.critic_optim = optim.Adam(self.critic.parameters(),
lr=self.critic_genome.learning_rate)
return evolved
def _transfer_weights(self, old_net, new_net):
"""Перенос весов между сетями разной архитектуры"""
old_state = old_net.state_dict()
new_state = new_net.state_dict()
for name, param in new_state.items():
if name in old_state:
old_param = old_state[name]
if old_param.shape == param.shape:
new_state[name] = old_param
new_net.load_state_dict(new_state)
class MigrationEnvironment:
"""Среда для моделирования миграционных процессов"""
def __init__(self, num_regions=NUM_REGIONS_STAGE1):
self.num_regions = num_regions
self.reset()
def reset(self):
"""Сброс среды"""
self.states = np.random.uniform(0.5, 1.5, (self.num_regions, 4))
self.states[:, 3] = 0.0 # Политика миграции
return self.states.copy()
def step(self, actions):
"""Выполнение шага в среде"""
next_states = np.zeros_like(self.states)
rewards = np.zeros(self.num_regions)
for i in range(self.num_regions):
# Обновление политики миграции
self.states[i, 3] = np.clip(self.states[i, 3] + actions[i][0], -1, 1)
# Вычисление миграции
neighbor_policies = [self.states[j, 3] for j in self.neighbors(i)]
migration = 0.1 * (self.states[i, 3] - np.mean(neighbor_policies))
# Обновление численности населения
self.states[i, 0] = np.clip(self.states[i, 0] + migration, 0.1, 2.0)
# Экономический рост
economic_growth = 0.05 * self.states[i, 0] * (1 + self.states[i, 3])
self.states[i, 2] = np.clip(self.states[i, 2] + economic_growth, 0.1, 2.0)
# Использование ресурсов
resource_usage = 0.02 * self.states[i, 0] * self.states[i, 2]
self.states[i, 1] = np.clip(self.states[i, 1] - resource_usage, 0.1, 2.0)
# Вычисление награды
rewards[i] = self._calculate_reward(i)
return self.states.copy(), rewards, False, {}
def _calculate_reward(self, region):
"""Вычисление награды для региона"""
pop_balance = -abs(self.states[region, 0] - 1.0)
economy = self.states[region, 2] - 1.0
resources = -0.5 * (1.0 - self.states[region, 1])
neighbor_policies = [self.states[j, 3] for j in self.neighbors(region)]
policy_align = 0.2 * np.mean([1 - abs(self.states[region, 3] - policy)
for policy in neighbor_policies])
return pop_balance + economy + resources + policy_align
def neighbors(self, region):
"""Получение соседей региона"""
return [(region - 1) % self.num_regions, (region + 1) % self.num_regions]
class DarwinGodelMADDPG:
"""Основной класс системы с интеграцией Darwin Gödel Machine и MADDPG"""
def __init__(self, num_regions=NUM_REGIONS_STAGE1, prev_agents=None):
self.num_regions = num_regions
self.env = MigrationEnvironment(num_regions)
self.evolution_logger = EvolutionLogger()
self.performance_monitor = PerformanceMonitor()
# Новые компоненты
self.time_tracker = TimeTracker()
self.visualizer = EnhancedVisualization()
self.save_manager = AutoSaveManager()
# Инициализация агентов
if prev_agents:
self.agents = prev_agents
for agent in self.agents:
agent.is_base = True
self._add_new_agents(num_regions - len(prev_agents))
else:
self.agents = [DarwinGodelAgent(i, num_regions) for i in range(num_regions)]
# Оптимизированная память с deque для лучшей производительности
self.memory = deque(maxlen=100000)
self.generation = 0
self.best_architectures = {
'actor': None,
'critic': None
}
def _add_new_agents(self, num_new):
"""Добавление новых агентов"""
for i in range(num_new):
base_agent = self.agents[i % len(self.agents)]
new_agent = DarwinGodelAgent(
len(self.agents),
self.num_regions,
actor_genome=copy.deepcopy(base_agent.actor_genome),
critic_genome=copy.deepcopy(base_agent.critic_genome)
)
self.agents.append(new_agent)
def train_stage(self, episodes, stage_name):
"""Обучение на определенной стадии с расширенным мониторингом"""
print(f"\n🚀 === {stage_name} ===")
print(f"🤖 Количество агентов: {len(self.agents)}")
print(f"📊 Количество эпизодов: {episodes}")
# Запуск отслеживания времени
self.time_tracker.start_stage(stage_name, episodes)
# Прогресс-бар для эпизодов[33]
episode_pbar = tqdm(range(episodes), desc=f"🎯 {stage_name}",
unit="эпизод", colour="blue")
for episode in episode_pbar:
states = self.env.reset()
episode_rewards = np.zeros(self.num_regions)
# Прогресс-бар для шагов
step_pbar = tqdm(range(STEPS), desc=f"Эпизод {episode+1}",
leave=False, unit="шаг", colour="green")
for step in step_pbar:
# Получение действий от агентов
actions = self._get_actions(states)
# Выполнение шага в среде
next_states, rewards, _, _ = self.env.step(actions)
# Сохранение опыта
self.memory.append((states, actions, rewards, next_states))
# Обновление агентов
if len(self.memory) > BATCH_SIZE:
self._update_agents()
# Обновление fitness агентов
for i, agent in enumerate(self.agents):
agent.update_fitness(rewards[i])
if not agent.is_base:
agent.steps_done += 1
episode_rewards += rewards
states = next_states
# Обновление прогресс-бара шагов
step_pbar.set_postfix({
'Награда': f"{np.mean(rewards):.3f}",
'Память': f"{len(self.memory)}"
})
# Эволюция архитектур
if episode % EVOLUTION_FREQUENCY == 0 and episode > 0:
print(f"\n🧬 Эволюция архитектур (поколение {self.generation})")
self._evolve_population()
# Обновление мониторинга
architectures = [agent.actor_genome.to_dict() for agent in self.agents]
self.performance_monitor.update_metrics(
episode, self.agents, states, episode_rewards, architectures)
# Обновление прогресса времени
self.time_tracker.update_progress(episode + 1)
# Обновление прогресс-бара эпизодов
avg_reward = np.mean(episode_rewards)
time_info = self.time_tracker.get_time_info()
episode_pbar.set_postfix({
'Ср.награда': f"{avg_reward:.3f}",
'Систем.': f"{np.sum(episode_rewards):.1f}",
'Время': f"{time_info.get('stage_progress', 0):.1f}%",
'Поколение': self.generation
})
# Вывод подробного прогресса каждые 10 эпизодов
if (episode + 1) % 10 == 0:
self.time_tracker.display_status()
# Проверка риска превышения лимита времени
if time_info.get('colab_timeout_risk', False):
print("⚠️ ВНИМАНИЕ: Приближается лимит времени Colab! Сохраняю промежуточные результаты...")
self.save_manager.save_stage_results(
f"{stage_name}_intermediate_{episode+1}",
self, self.performance_monitor, self.evolution_logger,
self.time_tracker, self.visualizer
)
# Завершение стадии
self.time_tracker.finish_stage()
# Сохранение результатов стадии
print(f"\n💾 Сохранение результатов стадии: {stage_name}")
self.save_manager.save_stage_results(
stage_name, self, self.performance_monitor,
self.evolution_logger, self.time_tracker, self.visualizer
)
print(f"✅ {stage_name} завершена!")
def _get_actions(self, states):
"""Получение действий от всех агентов с оптимизацией для GPU"""
actions = []
# Батчевая обработка для ускорения на GPU[18]
if torch.cuda.is_available() and len(self.agents) > 4:
# Группируем состояния и обрабатываем батчами
states_batch = torch.FloatTensor(states).to(device)
with torch.no_grad(): # Отключение градиентов для экономии памяти
for i, agent in enumerate(self.agents):
exploration = not agent.is_base
action = agent.act(states_batch[i], exploration=exploration)
actions.append(action)
else:
# Обычная обработка для CPU или малого количества агентов
for i, agent in enumerate(self.agents):
exploration = not agent.is_base
action = agent.act(states[i], exploration=exploration)
actions.append(action)
return actions
def _update_agents(self):
"""Обновление агентов методом MADDPG с оптимизациями для T4"""
# Используем mini-batch для экономии памяти GPU[31]
batch = random.sample(self.memory, min(BATCH_SIZE, len(self.memory)))
# Преобразование в тензоры с оптимизацией памяти
states = torch.FloatTensor(np.array([item[0] for item in batch])).to(device)
actions = torch.FloatTensor(np.array([item[1] for item in batch])).to(device)
rewards = torch.FloatTensor(np.array([item[2] for item in batch])).to(device)
next_states = torch.FloatTensor(np.array([item[3] for item in batch])).to(device)
for idx, agent in enumerate(self.agents):
if agent.is_base:
continue
# Очистка кэша GPU для экономии памяти
if torch.cuda.is_available():
torch.cuda.empty_cache()
# Обновление Critic
agent.critic_optim.zero_grad()
with torch.no_grad():
target_actions = []
for i, a in enumerate(self.agents):
target_action = a.actor_target(next_states[:, i, :])
target_actions.append(target_action)
target_q = rewards[:, idx] + GAMMA * agent.critic_target(
next_states.view(BATCH_SIZE, -1),
torch.cat(target_actions, dim=1)
).squeeze()
current_q = agent.critic(
states.view(BATCH_SIZE, -1),
actions.view(BATCH_SIZE, -1)
).squeeze()
critic_loss = nn.MSELoss()(current_q, target_q)
critic_loss.backward()
# Градиентное отсечение для стабильности[16]
torch.nn.utils.clip_grad_norm_(agent.critic.parameters(), 1.0)
agent.critic_optim.step()
# Обновление Actor
agent.actor_optim.zero_grad()
policy_actions = []
for i, a in enumerate(self.agents):
if i == idx:
policy_actions.append(agent.actor(states[:, i, :]))
else:
policy_actions.append(a.actor(states[:, i, :]).detach())
actor_loss = -agent.critic(
states.view(BATCH_SIZE, -1),
torch.cat(policy_actions, dim=1)
).mean()
actor_loss.backward()
# Градиентное отсечение для стабильности
torch.nn.utils.clip_grad_norm_(agent.actor.parameters(), 1.0)
agent.actor_optim.step()
# Обновление target сетей
self._soft_update(agent.actor, agent.actor_target, TAU)
self._soft_update(agent.critic, agent.critic_target, TAU)
def _soft_update(self, source, target, tau):
"""Мягкое обновление target сети"""
for target_param, source_param in zip(target.parameters(), source.parameters()):
target_param.data.copy_(tau * source_param.data + (1.0 - tau) * target_param.data)
def _evolve_population(self):
"""Эволюция популяции агентов с расширенным логированием"""
print(f"\n🧬 --- Эволюция поколения {self.generation} ---")
# Сбор fitness scores
fitness_scores = []
actor_genomes = []
critic_genomes = []
for agent in self.agents:
if not agent.is_base:
fitness = np.mean(agent.fitness_history[-10:]) if agent.fitness_history else 0
fitness_scores.append(fitness)
actor_genomes.append(agent.actor_genome)
critic_genomes.append(agent.critic_genome)
if not fitness_scores:
return
# Найти лучшие архитектуры
best_idx = np.argmax(fitness_scores)
best_actor_genome = actor_genomes[best_idx]
best_critic_genome = critic_genomes[best_idx]
# Обновить глобальные лучшие архитектуры
if (self.best_architectures['actor'] is None or
best_actor_genome.fitness > self.best_architectures['actor'].fitness):
self.best_architectures['actor'] = copy.deepcopy(best_actor_genome)
if (self.best_architectures['critic'] is None or
best_critic_genome.fitness > self.best_architectures['critic'].fitness):
self.best_architectures['critic'] = copy.deepcopy(best_critic_genome)
# Логирование эволюции
self.evolution_logger.log_evolution_step(
self.generation,
[g.to_dict() for g in actor_genomes],
fitness_scores,
np.max(fitness_scores)
)
# Эволюция агентов с прогресс-баром
evolved_count = 0
agent_pbar = tqdm(self.agents, desc="🔄 Эволюция агентов", leave=False)
for agent in agent_pbar:
if agent.is_base:
continue
old_actor_genome = copy.deepcopy(agent.actor_genome)
old_critic_genome = copy.deepcopy(agent.critic_genome)
evolved = agent.evolve_architecture(
self.best_architectures['actor'],
self.best_architectures['critic']
)
if evolved:
evolved_count += 1
# Логирование мутаций/скрещиваний
if random.random() < 0.5: # Предполагаем что была мутация
self.evolution_logger.log_mutation(
old_actor_genome.to_dict(),
agent.actor_genome.to_dict(),
"actor_mutation"
)
else: # Предполагаем скрещивание
self.evolution_logger.log_crossover(
old_actor_genome.to_dict(),
self.best_architectures['actor'].to_dict(),
agent.actor_genome.to_dict()
)
agent_pbar.set_postfix({'Эволюционировало': evolved_count})
print(f"✨ Эволюционировало агентов: {evolved_count}")
print(f"🏆 Лучший fitness: {np.max(fitness_scores):.3f}")
print(f"📊 Средний fitness: {np.mean(fitness_scores):.3f}")
print(f"📏 Стандартное отклонение: {np.std(fitness_scores):.3f}")
self.generation += 1
def train_all_stages(self):
"""Обучение всех стадий с комплексным мониторингом"""
print("\n🚀 ========== ЗАПУСК ПОЛНОГО ОБУЧЕНИЯ ==========")
print("🎯 Darwin Gödel MADDPG - Система управления демографическими процессами")
print("⚡ Оптимизировано для Google Colab T4 GPU")
print("="*60)
try:
# Стадия 1: 7 агентов
print(f"\n🥇 СТАДИЯ 1: Базовое обучение ({NUM_REGIONS_STAGE1} регионов)")
self.train_stage(EPISODES_PER_STAGE, "Стадия 1")
# Расширение до 28 агентов
print(f"\n🥈 ПЕРЕХОД К СТАДИИ 2: Расширение до {NUM_REGIONS_STAGE2} регионов")
stage1_agents = copy.deepcopy(self.agents)
self.__init__(NUM_REGIONS_STAGE2, stage1_agents)
self.train_stage(EPISODES_PER_STAGE, "Стадия 2")
# Расширение до 89 агентов
print(f"\n🥉 ПЕРЕХОД К СТАДИИ 3: Расширение до {NUM_REGIONS_STAGE3} регионов")
stage2_agents = copy.deepcopy(self.agents)
self.__init__(NUM_REGIONS_STAGE3, stage2_agents)
self.train_stage(EPISODES_PER_STAGE, "Стадия 3")
# Создание финального отчета
self._create_final_report()
print("\n🎉 ========== ОБУЧЕНИЕ УСПЕШНО ЗАВЕРШЕНО! ==========")
except Exception as e:
print(f"\n❌ Ошибка при обучении: {e}")
print("💾 Сохраняю промежуточные результаты...")
self.save_manager.save_stage_results(
"emergency_save", self, self.performance_monitor,
self.evolution_logger, self.time_tracker, self.visualizer
)
raise
def _create_final_report(self):
"""Создание финального отчета о всем процессе обучения"""
print("\n📋 Создание финального отчета...")
final_report_path = os.path.join(self.save_manager.session_dir, 'FINAL_REPORT.txt')
with open(final_report_path, 'w', encoding='utf-8') as f:
f.write("🚀 ФИНАЛЬНЫЙ ОТЧЕТ СИСТЕМЫ DARWIN GÖDEL MADDPG\n")
f.write("="*80 + "\n\n")
# Общая информация о сессии
session_time = time.time() - self.time_tracker.session_start_time
f.write(f"📅 Дата завершения: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write(f"⏱️ Общее время сессии: {self.time_tracker._format_time(session_time)}\n")
f.write(f"🎯 Количество завершенных стадий: {len(self.time_tracker.stage_history)}\n")
f.write(f"🧬 Финальное поколение эволюции: {self.generation}\n")
f.write(f"🤖 Финальное количество агентов: {len(self.agents)}\n\n")
# Статистика по стадиям
f.write("📊 СТАТИСТИКА ПО СТАДИЯМ:\n")
f.write("-"*50 + "\n")
for stage in self.time_tracker.stage_history:
f.write(f"\n🎯 {stage['name']}:\n")
f.write(f" ⏱️ Длительность: {self.time_tracker._format_time(stage['total_duration'])}\n")
f.write(f" 📈 Эпизодов: {stage['episodes_total']}\n")
f.write(f" ⚡ Скорость: {stage['total_duration']/stage['episodes_total']:.1f}с/эпизод\n")
# Лучшие достижения
f.write(f"\n🏆 ЛУЧШИЕ ДОСТИЖЕНИЯ:\n")
f.write("-"*50 + "\n")
if self.performance_monitor.metrics['episode_rewards']:
best_reward = max(self.performance_monitor.metrics['episode_rewards'])
avg_reward = np.mean(self.performance_monitor.metrics['episode_rewards'])
f.write(f"🎯 Максимальная награда за эпизод: {best_reward:.3f}\n")
f.write(f"📊 Средняя награда: {avg_reward:.3f}\n")
if self.evolution_logger.evolution_history:
best_fitness = max([step['best_fitness'] for step in self.evolution_logger.evolution_history])
f.write(f"💪 Максимальный fitness: {best_fitness:.3f}\n")
f.write(f"🔀 Общее количество мутаций: {len(self.evolution_logger.mutation_log)}\n")
f.write(f"🤝 Общее количество скрещиваний: {len(self.evolution_logger.crossover_log)}\n")
# Эффективность использования ресурсов
f.write(f"\n⚡ ЭФФЕКТИВНОСТЬ РЕСУРСОВ:\n")
f.write("-"*50 + "\n")
f.write(f"🖥️ Устройство: {'GPU (CUDA)' if torch.cuda.is_available() else 'CPU'}\n")
if torch.cuda.is_available():
f.write(f"🔧 GPU: {torch.cuda.get_device_name(0)}\n")
f.write(f"💾 Использование GPU памяти: оптимизировано для T4\n")
total_episodes = sum(stage['episodes_total'] for stage in self.time_tracker.stage_history)
if total_episodes > 0:
episodes_per_hour = total_episodes / (session_time / 3600)
f.write(f"📈 Скорость обучения: {episodes_per_hour:.1f} эпизодов/час\n")
f.write(f"\n🎉 ОБУЧЕНИЕ УСПЕШНО ЗАВЕРШЕНО!\n")
f.write("Все результаты сохранены в соответствующих папках стадий.\n")
print(f"📋 Финальный отчет сохранен: {final_report_path}")В качеств основной метрики оценки мы будем использовать категорию fitness . Фитнес-функция используется в контексте генетических алгоритмов, в том числе в DGM. Фитнес-функция вычисляет «близость» особи к тому, чтобы быть решением рассматриваемой задачи. Чем лучше значение фитнес-функции, тем больше шансов у особи на выживание и размножение.
Обычный генетический алгоритм состоит из трех этапов. В нашем случае сначала мы создаем популяцию агентов, потом подсчитываем функцию их приспособленности (нашу любимую фитнесс-функцию) - мы будем смотреть как среднее значение функции, так и лучшие значения.
Первый этап обучения для 7 агентов занял немногим более 9 минут. Средняя награда составила награда=78.587, Системная (общая) =550.1. И мы получили 8 поколений акторов. Хотя в целом ход обучения повторял результаты первого этапа прошлого эксперимента касаемо динамики наград.
Отметим так же тот факт, что показатели координации политики 7 агентов практически сразу же вышли на максимальный уровень (7 искусственным интеллектам проще договориться друг с другом - см. график «координация политики», за исключением стартовых эпизодов в начале обучения и примерно 9 эпизода, с небольшим провалом, очевидно по причине воздействия случайного, шумового фактора.
Отметим в ходе эксперимента так же достаточно большую разницу между значениями средней и лучшей фитнесс-функции и неравномерные от поколения к поколению показатели эффективности эволюции. Повторимся, все результаты и графики эксперимента с DGM доступны в отдельном репозитории.
Вторая стадия DGM-эксперимента длилась существенно большее (на порядок) время - почти час. График обучения показывал схожую динамику, хотя численность населения немного подросла, упала средняя награда (35,7), системная приблизилась к 1000. И очень любопытной показала себя динамика графика координации политики.
Примерно с 11 до 30 эпохи агенты действовали согласованно, а вот после началась некоторая нестабильность. Впрочем, мы надеемся, что там, в компьютерном мире, все обошлось обычными политическими кризисами, без войн и революций. Впрочем вывод интересный: большее число одноуровневых агентов может проводить какое-то время гибкую скоординированную политику, но это «не вечно и не навсегда». Что было конкретной причиной расинхронизации (где-то случилась пандемия или пришел к власти усатый-полосатый и хвостатый диктатор, объявивший политику других агентов «неправильной») - мы пока не знаем.
Третья стадия, длительностью уже более шести с половиной часов, оказалась самой интересной. Во первых, средняя награда стала еще ниже, как и общесистемная, средняя фитнесс-функция стала фактически отрицательной (при том что лучше показатели у отдельных агентов даже стали выше). Показатель популяционного баланса стал ухудшаться. Так же индекс демографической стабильности был хуже, чем на двух предыдущих стадиях. Улучшение фитнесс-функции не произошло, в целом эффективность эволюции была отрицательной.
Уровень координации деятельности ИИ-агентов хоть и постепенно нарастал, но не достиг значений из предыдущих стадий. Вероятнее всего, выделенного количества эпох недостаточно для обучения системы из 89 действующих агентов. Второе предположение - с увеличением количества действующих субъектов или «перепроизводства» политических акторов у нас случилась настоящая феодальная раздробленность, в ходе которой отдельные ИИ-агенты показали высокую адаптивность и эффективность в своем развитии, и если бы речь шла о реальной ситуации, со временем они стали бы ядрами сети и нарождающейся новой политической системы, предлагая свои правила и ценности всем остальным.
ВЫВОДЫ
Логично предположить, что в нашем «урезанном» виде DGM не справится с задачей балансировки политики ИИ-агентов и ему на третьей стадии потребуется гораздо большее количество эпох для «приведения в соответствие» политики всех регионов.
Это справедливо и для реальной ситуации - в начале 00-ых годов в России структура политического управления породила новый политический институт федеральных округов. И сделала это вовсе не из прихоти, а в целях координации взаимодействия на макроуровне и опять же «приведения в соответствие» регионального законодательства федеральному. Наладить взаимодействие между ограниченным набором политических акторов - проще и быстрее.
Кстати, один из перспективных путей улучшения MADDPG - это введение иерархии агентов, каждый из которых управляет подуровнем из нескольких подчиненных, используя не весь объем информации и сокращая тем самым «проклятие размеренности».
Другой вариант совершенствования уже DGM - это введение процедур мутации, эволюции не только актора, но и критика и механизмов взаимодействия между агентами, трансляции, сохранения и применения накопленного опыта. Здесь можно поэкспериментировать не только с PER, но и воспроизведением ретроспективного опыта (PER), распределённым повтором опыта, опытом воспроизведения с приоритетной выборкой (гибридом PER и PER), воспроизведением опыта на основе разнообразия (DBER) и другими подходами.
Третий вариант, который и является «полноценным» DGM связан с созданием определённого внешнего по отношению к участникам эксперимента цикла и взаимодействия нескольких современных элементов ИИ. Во-первых, это оценивающий контур - дообученная на демографических материалах и моделях языковая модель, оценивающая результаты эксперимента. Во-вторых, контур генерации промптов на основе полученных оценок. В-третьих - кодировщик, языковая модель, пишущая код по промптам. В четвертых, сама модельная среда под MADDPG. Оценивающий контур, контур генерации промптов, модель-кодировщик и модель среды могут дорабатываться и совершенствоваться: языковая модель тюнится под итогам новых результатов экспериментов, модель генерации промптов и модель кодинга так же совершенствуются с учетом полученных результатов. Ну а про саму среду с MADDPG - мы отметим, что могут изменяться все ее гиперпараметры, включая количество и иерархию агентов, так и механизмы трансляции и воспроизведения полученного ими системного опыта. И вполне реально получить не просто улучшенный MADDPG, но и более эффективную дочернюю версию алгоритма.
Вот такой алгоритм уже будет полноценным «переходным» ИИ, способным вылезти из детской коляски и обставить на повороте своих создателей!
Автор выражает благодарность преподавателям курса Reinforcement Learning от OTUS за возможность существенно «подапгрейдить» свои компетенции в вопросах совершенствования процедур государственного политического управления в сфере миграции демографии с помощью элементов ML и Deep RL.
