遗传算法与神经网络：用进化训练贪吃蛇

当梯度下降陷入局部最优解时，不妨试试另一条路：让神经网络像生物一样“进化”。本文介绍如何用遗传算法（选择、交叉、变异）训练一个贪吃蛇智能体，附完整 Python 源码与训练反思。

在《初识人工智能——神经网络&强化学习》中，我们介绍了神经网络如何通过梯度下降来训练——但这条路有个常见的问题：优化算法有时会“卡”在一个并非最好的结果上，也就是所谓的局部最优解。

为了解决这个问题，我们可以尝试利用遗传算法（Genetic Algorithm, GA）。遗传算法是一种模拟自然界生物进化过程的搜索优化算法，它借鉴了达尔文的*“物竞天择，适者生存”以及孟德尔的遗传学原理*。它的基本思想是：将待优化的问题（比如神经网络的权重）编码成一个个“个体”（通常称为“染色体”），这些个体组成一个“种群”。算法通过模拟生物进化中的“选择”（保留性能好的个体）、“交叉”（将两个好个体的部分信息进行组合，产生新个体）和“变异”（对个体的某些信息进行随机改变）等操作，让种群一代代地进化，最终朝着越来越优的方向发展。

当遗传算法应用于神经网络时，会将网络的所有权重和偏置串联成一个编码串（“染色体”），每个编码串代表一个特定的神经网络“个体”。算法先随机生成一批个体（初始种群），然后评估每个个体在目标任务上的表现（如贪吃蛇得分），根据表现好坏进行“选择”，选出优秀个体作为“父母”。再通过“交叉”和“变异”生成新的个体（子代），替换旧种群。如此循环，遗传算法能在广阔权重空间中搜索，避开局部最优解，找到性能更优的权重组合。

我们以此为思路，可以构建出一个基于遗传算法训练的神经网络，并将其运用到贪吃蛇游戏智能体的训练中：

snake_ga_nn.py

import pygame
import os
import random
import numpy
import csv
import logging 

# --- 配置日志记录 ---
# 设置日志记录的基本配置，包括日志级别、格式和输出目标（文件和控制台）
logging.basicConfig(
    level=logging.INFO, # 设置日志级别为INFO，记录INFO及以上级别的日志
    format='%(asctime)s - %(levelname)s - %(message)s', # 设置日志格式：时间-级别-消息
    handlers=[
        logging.FileHandler('training_log.log', mode='a', encoding='utf-8'), # 记录到文件（追加模式）
        logging.StreamHandler() # 同时输出到控制台
    ]
)

# --- 输入参数设置 ---
# 定义游戏和遗传算法的基本参数
M = 10  # 游戏网格的行数
N = 10  # 游戏网格的列数
grid_size = 25  # 每个网格的像素大小
population_length = 500  # 种群大小（个体数量）
parants_length = 50  # 每代选择的父代数量
file_name = 'snake_GA.npz'  # 保存种群数据的文件名

# --- CSV 文件参数设置 ---
csv_file_prefix = 'snake_training_data'  # CSV文件前缀
max_rows_per_file = 100000  # 每个CSV文件的最大行数
grids = set((i, j) for i in range(M) for j in range(N))  # 所有可能的网格位置集合

# --- 函数定义 ---

def get_next_csv_filename(base_name, max_rows):
    """根据前一个文件的行数，确定下一个写入的CSV文件名
    
    Args:
        base_name: CSV文件的基础名称
        max_rows: 每个文件的最大行数
        
    Returns:
        下一个可用的CSV文件名
    """
    counter = 1
    while True:
        filename = f"{base_name}_{counter:03d}.csv"
        if not os.path.exists(filename):
            if counter == 1:
                return filename
            else:
                # 检查前一个文件
                prev_filename = f"{base_name}_{counter-1:03d}.csv"
                if os.path.exists(prev_filename):
                    try:
                        with open(prev_filename, 'r', newline='', encoding='utf-8') as f:
                            row_count = sum(1 for row in f)
                        # 如果前一个文件行数达到或超过限制，则开始新文件
                        if row_count >= max_rows + 1: 
                            return filename
                        else:
                            counter += 1
                            continue
                    except IOError as e:
                        logging.warning(f"无法读取前一个CSV文件 {prev_filename} 以检查行数: {e}. 尝试下一个编号。")
                        counter += 1
                        continue
                else:
                    return filename
        else:
            try:
                with open(filename, 'r', newline='', encoding='utf-8') as f:
                    row_count = sum(1 for row in f)
                if row_count < max_rows + 1:
                    # 当前文件未满，返回当前文件
                    return filename
                else:
                    # 当前文件已满，检查下一个编号
                    counter += 1
                    continue
            except IOError as e:
                logging.warning(f"无法读取当前CSV文件 {filename} 以检查行数: {e}. 尝试下一个编号。")
                counter += 1
                continue

def save_to_csv(gen, individual_id, fitness, score, steps, csv_filename):
    file_exists = os.path.isfile(csv_filename)

    
    try:
        with open(csv_filename, 'a', newline='', encoding='utf-8') as csvfile:
            fieldnames = ['Generation', 'Individual_ID', 'Fitness', 'Score', 'Steps']
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

            if not file_exists:
                writer.writeheader()  # 如果是新文件，写入表头

            writer.writerow({
                'Generation': gen,
                'Individual_ID': individual_id,
                'Fitness': fitness,
                'Score': score,
                'Steps': steps
            })
    except PermissionError as e:
        logging.error(f"权限被拒绝，无法写入文件 {csv_filename}. 请确保文件未被其他程序（如Excel）打开，并检查目录权限。错误详情: {e}")
        user_input = input("文件写入失败，是否尝试继续运行程序？(y/n): ")
        if user_input.lower() != 'y':
            logging.info("用户选择停止程序。")
            pygame.quit()
            exit(1)
    except IOError as e:
        logging.error(f"写入CSV文件 {csv_filename} 时发生IO错误: {e}")
        user_input = input("文件写入失败，是否尝试继续运行程序？(y/n): ")
        if user_input.lower() != 'y':
            logging.info("用户选择停止程序。")
            pygame.quit()
            exit(1)
    except Exception as e:
        logging.error(f"写入CSV文件 {csv_filename} 时发生未知错误: {e}")
        user_input = input("文件写入失败，是否尝试继续运行程序？(y/n): ")
        if user_input.lower() != 'y':
            logging.info("用户选择停止程序。")
            pygame.quit()
            exit(1)

def Snake_game():
    """运行一次贪吃蛇游戏，返回适应度、得分和步数"""
    global Snake, screen, loop, mloop, pause_time
    pygame.init()
    screen = pygame.display.set_mode((M * grid_size, N * grid_size))
    
    # 初始化蛇的位置（随机选择两个相邻的网格）
    (x1, y1) = random.choice(list(grids))
    (x2, y2) = random.choice([(x1, y1 + 1), (x1 - 1, y1), (x1, y1 - 1), (x1 + 1, y1)])
    Snake, steps, uniq, pause_time = [(x1, y1), (x2, y2)], 0, set(), 0
    
    food()  # 生成第一个食物
    loop = True
    
    while loop:
        steps = steps + 1
        prediction_from_genetic_weights()  # 使用神经网络预测移动方向
        update_snake()  # 更新蛇的位置
        
        # 检查游戏结束条件
        if len(Snake) == M * N:
            print('Great....Snake get maximum Score')
            loop = False
        elif snake_head == Food:
            food()  # 吃到食物，生成新食物
            Snake.append(snake_tail)  # 蛇身增长
            pause_time = key_sensitive * 5 if len(Snake) == M * N - 10 else pause_time
        elif snake_head not in grids or snake_head in snake_body:
            loop = False  # 撞墙或撞到自己，游戏结束
            
        # 处理事件（如退出游戏或调整速度）
        ev = pygame.event.get()
        for event in ev:
            if event.type == pygame.QUIT:
                pygame.quit()
                mloop, loop = False, False
            elif event.type == pygame.KEYDOWN:
                pause_time = pause_time + key_sensitive if event.key == pygame.K_UP else pause_time - key_sensitive if event.key == pygame.K_DOWN and pause_time >= key_sensitive else pause_time
        
        # 检查状态是否重复（避免循环）
        state_key = (Snake[0], Food)
        if state_key not in uniq:
            uniq.add(state_key)
            if len(uniq) > M * N - 2: 
                 uniq.pop() 
        else:
            loop = False 
    
    score = len(Snake) - 2  # 计算得分（蛇长度-初始长度）
    # 计算适应度函数（考虑得分和步数的平衡）
    return (score + 0.5 + 0.5 * (score - steps / (score + 1)) / (score + steps / (score + 1))) * 1000000, score, steps

def food():
    """在空白位置随机生成食物"""
    global Food
    snake_no_grids = grids - set(Snake)  # 计算所有空白位置
    Food = random.choice(list(snake_no_grids))  # 随机选择一个位置

def prediction_from_genetic_weights():
    """使用神经网络预测蛇的移动方向"""
    global action
    lstop = [[1, 0, 0, 0], [0, 1, 0, 0], [0, 0, 1, 0], [0, 0, 0, 1]]
    lst = [(0, -1), (1, 0), (0, 1), (-1, 0)]  # 四个可能的移动方向
    
    # 确定蛇头的当前方向
    head_diriction = lstop[lst.index((Snake[0][0] - Snake[1][0], Snake[0][1] - Snake[1][1]))]
    (x, y) = Snake[0]

    # 计算八个方向的视野信息（墙壁距离、食物距离、身体距离）
    d1 = [(x, _) for _ in range(y - 1, -1, -1)]  # 上
    d3 = [(_, y) for _ in range(x + 1, M)]  # 右
    d5 = [(x, _) for _ in range(y + 1, N)]  # 下
    d7 = [(_, y) for _ in range(x - 1, -1, -1)]  # 左
    
    # 对角线方向（限制范围避免越界）
    d8 = [(x - _, y - _) for _ in range(1, min(x, y) + 1)]  # 左上
    d4 = [(x + _, y + _) for _ in range(1, min(M - x, N - y) + 1)]  # 右下
    d2 = [(x + _, y - _) for _ in range(1, min(M - x, y + 1)) if (x + _, y - _) in grids]  # 右上
    d6 = [(x - _, y + _) for _ in range(1, min(x + 1, N - y)) if (x - _, y + _) in grids]  # 左下

    d, val = [d1, d2, d3, d4, d5, d6, d7, d8], min(M, N) - 1
    # 计算墙壁距离、食物存在性和身体存在性
    wall_distance = [len(i) / val if val != 0 and i else 0 for i in d]
    food_presence = [(val - j.index(Food)) / val if val != 0 and Food in j else 0 for j in d]
    body_presence = [min([dv.index(v) if v in Snake else val for v in dv]) / val if val != 0 and dv else 0 for dv in d]

    # 组合输入特征
    vision = [j[i] for i in range(8) for j in [wall_distance, body_presence, food_presence]]
    input_layer = vision + head_diriction
    action = neural_network(input_layer)  # 通过神经网络预测动作

def neural_network(ip):
    """前向传播的神经网络
    
    Args:
        ip: 输入特征向量
        
    Returns:
        输出层的最大值对应的动作
    """
    m1, s1 = numpy.reshape(ip, (1, NN[0])), 0
    for _ in range(len(AF)):
        l1, l2 = NN[_], NN[_ + 1]
        s2, s3 = s1 + l1 * l2, s1 + l2 + l1 * l2
        m2, m3 = numpy.reshape(weights[s1:s2], (l1, l2)), numpy.reshape(weights[s2:s3], (1, l2))
        m4 = numpy.matmul(m1, m2) + m3
        m1, s1 = AF[_](m4), s3
    return Actions[numpy.argmax(m1)]

def relu(x):
    """ReLU激活函数"""
    lst = numpy.where(x[0] > 0, x[0], 0)
    return lst

def sigmoid(x):
    """Sigmoid激活函数（带溢出保护）"""
    z = numpy.clip(x[0], -500, 500)  # 防止数值溢出
    return 1 / (1 + numpy.exp(-z))

def update_snake():
    """更新蛇的位置和游戏画面"""
    global snake_tail, snake_head, snake_body
    display()  # 更新显示
    pygame.time.wait(pause_time)  # 控制游戏速度
    
    (x, y) = Snake[0]
    # 根据预测的动作移动蛇
    if action == 'Right':
        Snake.insert(0, (x + 1, y))
    elif action == 'Left':
        Snake.insert(0, (x - 1, y))
    elif action == 'Bottum':
        Snake.insert(0, (x, y + 1))
    else: # Top
        Snake.insert(0, (x, y - 1))
    
    snake_tail = Snake.pop()  # 移除蛇尾（除非吃到食物）
    snake_head, snake_body = Snake[0], Snake[1:]

def display():
    """绘制游戏画面"""
    pygame.draw.rect(screen, (0, 0, 0), (0, 0, M * grid_size, N * grid_size))  # 黑色背景
    # 绘制蛇头（白色实心）
    pygame.draw.rect(screen, (255, 255, 255), (Snake[0][0] * grid_size, Snake[0][1] * grid_size, grid_size, grid_size))
    # 绘制蛇身（白色边框）
    for i in Snake[1:]:
        pygame.draw.rect(screen, (255, 255, 255), (i[0] * grid_size, i[1] * grid_size, grid_size, grid_size), 1)
    # 绘制食物（绿色）
    pygame.draw.rect(screen, (0, 255, 0), (Food[0] * grid_size, Food[1] * grid_size, grid_size, grid_size))
    pygame.display.update()  # 更新显示

def crossover():
    """遗传算法的交叉操作"""
    global offspring
    offspring = []
    for _ in range(population_length - parants_length):
        # 轮盘赌选择父代
        parant1_id = random.choice(Roulette_wheel)
        parant2_id = random.choice(Roulette_wheel)
        while parant2_id == parant1_id:
            parant2_id = random.choice(Roulette_wheel)
        # 单点交叉生成后代
        wts = [parants[parant1_id][i] if random.uniform(0, 1) < 0.5 else parants[parant2_id][i] for i in range(weights_length)]
        offspring.append(wts)

def mutation():
    """遗传算法的变异操作"""
    global offspring
    for i in range(population_length - parants_length):
        # 每个个体有5%的权重发生变异
        for _ in range(int(weights_length * 0.05)):
            plc = random.randint(0, weights_length - 1)
            value = random.choice(numpy.arange(-0.5, 0.5, step=0.001))
            offspring[i][plc] = offspring[i][plc] + value

# --- 主算法参数设置 ---
NN = [28, 8, 4]  # 神经网络结构：输入层28个神经元，隐藏层8个，输出层4个
AF = [relu, sigmoid]  # 激活函数：隐藏层用ReLU，输出层用Sigmoid
pause_time, key_sensitive, generation_length, mloop = 0, 15, 2000, True
Actions = ['Top', 'Right', 'Bottum', 'Left']  # 可能的动作
# 轮盘赌选择概率分布（前20%的个体有更高选择概率）
Roulette_wheel = list(range(0, int(0.2 * parants_length))) * 3 + \
                 list(range(int(0.2 * parants_length), int(0.5 * parants_length))) * 2 + \
                 list(range(int(0.5 * parants_length), parants_length))
weights_length = sum([NN[_] * NN[_ + 1] + NN[_ + 1] for _ in range(len(NN) - 1)])  # 计算总权重数

# --- 初始化或加载种群数据 ---
if file_name not in os.listdir(os.getcwd()):
    # 如果不存在保存文件，初始化随机种群
    population, statis = numpy.random.choice(numpy.arange(-1, 1, step=0.001), size=(population_length, weights_length), replace=True), numpy.array([[0, 0, 0, 0]])
    Generation, High_score = 1, 0
    logging.info("未找到现有种群数据，已初始化新种群。")
else:
    try:
        # 尝试加载现有种群数据
        IP = numpy.load(file_name)
        loaded_statis = IP['STATIS']
        if loaded_statis.ndim != 2 or loaded_statis.shape[1] != 4:
            logging.warning(f"加载的 STATIS 数据格式不正确 (shape: {loaded_statis.shape})，将重新初始化。")
            population, statis = numpy.random.choice(numpy.arange(-1, 1, step=0.001), size=(population_length, weights_length), replace=True), numpy.array([[0, 0, 0, 0]])
            Generation, High_score = 1, 0
        else:
            population, statis = IP['POPULATION'], loaded_statis
            Generation, High_score = statis[-1][0] + 1, statis[-1][-1]
            logging.info(f"成功从 {file_name} 加载种群数据，从第 {Generation} 代开始。")
    except (IOError, ValueError) as e:
        logging.error(f"加载种群数据文件 {file_name} 时出错: {e}. 将初始化新种群。")
        population, statis = numpy.random.choice(numpy.arange(-1, 1, step=0.001), size=(population_length, weights_length), replace=True), numpy.array([[0, 0, 0, 0]])
        Generation, High_score = 1, 0

# --- 主循环 ---
current_csv_file = get_next_csv_filename(csv_file_prefix, max_rows_per_file)

while Generation <= generation_length and mloop:
    print('###################### ', 'Generation ', Generation, ' ######################')
    Fitness, Score, i = [], [], 0
    
    # 评估当前种群的所有个体
    while i < population_length and mloop:
        weights, i = list(population[i, :]), i + 1
        fitness, score, steps = Snake_game()
        print('Chromosome ', "{:03d}".format(i), ' >>> ', 'Score : ', "{:03d}".format(score), ', Steps : ', "{:04d}".format(steps), ', Fitness : ', fitness)
        Fitness.append(fitness)
        Score.append(score)

        # 保存当前个体的数据到CSV
        save_to_csv(Generation, i, fitness, score, steps, current_csv_file)

        # 检查是否需要切换到下一个CSV文件
        try:
            with open(current_csv_file, 'r', newline='', encoding='utf-8') as f:
                current_row_count = sum(1 for row in f)
            if current_row_count >= max_rows_per_file + 1: 
                 current_csv_file = get_next_csv_filename(csv_file_prefix, max_rows_per_file)
                 logging.info(f"切换到新的CSV文件: {current_csv_file}")
        except IOError as e:
            logging.warning(f"无法检查当前CSV文件 {current_csv_file} 的行数: {e}. 继续使用当前文件。")

    # 选择父代（基于适应度）
    parants, max_fitness, avg_score, j = [], max(Fitness), sum(Score) / len(Score), 0
    while j < parants_length and mloop:
        j, parant_id = j + 1, Fitness.index(max(Fitness))
        Fitness[parant_id] = -999  # 标记已选个体
        parants.append(list(population[parant_id, :]))
    
    # 生成新种群
    while mloop and j == parants_length:
        j = j + 1
        High_score = max(Score) if max(Score) > High_score else High_score
        print('Generation high score : ', max(Score), ', Generation Avg score : ', avg_score, ', Overall high score : ', High_score)
        crossover()  # 交叉操作
        mutation()   # 变异操作
        # 记录统计信息
        new_row = numpy.array([[Generation, max(Score), avg_score, High_score]])
        statis = numpy.row_stack((statis, new_row))
        population = numpy.reshape(parants + offspring, (population_length, -1))
    Generation = Generation + 1

# 训练结束，保存数据
pygame.quit()
try:
    numpy.savez(file_name, POPULATION=population, STATIS=statis)
    logging.info(f"训练完成，种群数据已保存到 {file_name}。")
except IOError as e:
    logging.error(f"保存种群数据文件 {file_name} 时出错: {e}")

经过一轮完整的训练后，生成的贪吃蛇智能体确实能够有效地游玩游戏，能吃到大量食物获得高分，甚至填满整个棋盘。然而，这个智能体的行为也暴露出一些特点：即便在游戏初期、蛇身还较短的时候，它也表现得相当谨慎保守，倾向于规避潜在的风险。

这种过度谨慎的行为，一方面可能是因为适应度函数的设计所致。在代码中，该函数同时考虑了得分和步数，这可能导致算法更倾向于奖励那些“求稳”、优先保证存活的策略，而不是鼓励积极寻找食物。另一方面，遗传算法本身的搜索方向性不强，在漫长的进化过程中，可能无意中将这种规避风险的行为模式固化了下来。

此外，整个训练过程耗时巨大，需要经历数千代的遗传迭代，每一代又要评估数百个个体，反复运行游戏模拟，这使得优化过程非常漫长，这也是遗传算法的缺陷之一。这似乎也能从某种层面反映自然进化时间尺度跨度之大。

某一轮训练的评估数据与成果：

完整的项目与数据：Snake_GA-NN
参考：AI-learns-to-play-Snake-using-Genetic-Algorithm-and-Neural-Network