如何在类架构的国际象棋引擎中用Python multiprocessing实现并行搜索？

我正在开发一款国际象棋引擎，想通过并行化优化搜索合法走法以缩短找到最佳走法的耗时。我找到了Python的multiprocessing模块，但它需要在if __name__ == "__main__"环境下执行，不知道怎么在现有的类模型中使用。我的代码大致如下：

class ourEngine:
    def __init__(self, board, engine_color, thinking_time=0):
        self.board = board
        self.thinking_time = thinking_time
        self.engine_color = engine_color

    def evaluate(self, board):
        # 评估计算逻辑
    
    def search(self, board, depth, engine_color, alpha=float("-inf"), beta=float("inf")):
        # 带有alpha-beta剪枝的Minimax算法，遍历python-chess的board.legal_moves

我该如何在不破坏现有类结构的前提下，利用multiprocessing.Pool在所有CPU上同时处理legal_moves？我知道需要设置队列，但网上找不到相关示例。曾尝试通过单独函数调用search来实现，但没能成功。

1. 核心思路 #

解决类中使用multiprocessing的关键是：

• 子进程无法共享父进程的实例内存，因此需要将搜索逻辑所需的参数全部序列化传递
• 必须将进程池的创建逻辑放在if __name__ == "__main__"判断内，避免子进程重复初始化资源

2. 实现代码 #

import multiprocessing
import chess

class ourEngine:
    def __init__(self, board, engine_color, thinking_time=0):
        self.board = board
        self.thinking_time = thinking_time
        self.engine_color = engine_color

    def evaluate(self, board):
        # 示例评估逻辑：基于子力价值计算得分
        score = 0
        piece_values = {chess.PAWN: 1, chess.KNIGHT: 3, chess.BISHOP: 3, chess.ROOK: 5, chess.QUEEN: 9}
        for square in chess.SQUARES:
            piece = board.piece_at(square)
            if piece:
                value = piece_values.get(piece.piece_type, 0)
                score += value if piece.color == self.engine_color else -value
        return score

    def search(self, board, depth, engine_color, alpha=float("-inf"), beta=float("inf")):
        # Minimax+alpha-beta剪枝核心逻辑
        if depth == 0 or board.is_game_over():
            return self.evaluate(board)
        
        current_max = float("-inf") if engine_color == board.turn else float("inf")
        for move in board.legal_moves:
            board.push(move)
            eval_result = self.search(board, depth-1, engine_color, alpha, beta)
            board.pop()
            
            if engine_color == board.turn:
                current_max = max(current_max, eval_result)
                alpha = max(alpha, eval_result)
            else:
                current_max = min(current_max, eval_result)
                beta = min(beta, eval_result)
                
            if beta <= alpha:
                break
        return current_max

# 定义并行任务函数：子进程中独立初始化引擎并执行搜索
def parallel_search_task(args):
    board_fen, depth, engine_color, alpha, beta = args
    board = chess.Board(board_fen)
    engine = ourEngine(board, engine_color)
    return engine.search(board, depth, engine_color, alpha, beta)

if __name__ == "__main__":
    # 主程序初始化
    initial_board = chess.Board()
    engine_color = chess.WHITE
    search_depth = 3

    # 生成并行任务列表：将每个走法后的棋盘序列化为FEN字符串
    task_list = []
    legal_moves = list(initial_board.legal_moves)
    for move in legal_moves:
        temp_board = initial_board.copy()
        temp_board.push(move)
        task_list.append((temp_board.fen(), search_depth-1, engine_color, float("-inf"), float("inf")))

    # 使用进程池并行处理所有走法的搜索
    with multiprocessing.Pool() as pool:
        search_results = pool.map(parallel_search_task, task_list)

    # 筛选最佳走法
    best_eval = max(search_results) if engine_color == initial_board.turn else min(search_results)
    best_move = legal_moves[search_results.index(best_eval)]

    print(f"最佳走法: {best_move}, 评估值: {best_eval}")

3. 关键注意事项 #

• 序列化处理：chess.Board对象无法直接在进程间传递，因此用FEN字符串序列化，子进程中再还原为棋盘对象
• 子进程独立实例化：每个子进程必须单独创建ourEngine实例，因为进程间内存完全隔离，父进程的实例状态无法被子进程访问
• 队列替代方案：如果需要异步处理任务，可以使用multiprocessing.Queue手动提交和获取结果，但Pool.map更适合这种批量同步处理的场景，代码更简洁
• 避免重复初始化：所有涉及进程池创建和主逻辑的代码必须放在if __name__ == "__main__"内，防止子进程重复执行主代码引发错误

内容的提问来源于stack exchange，提问作者anon