简介Python基于蒙特卡洛树搜索算法实现多机器人区域覆盖路径规划的完整工程面向机器人路径规划、多智能体协同与AI算法学习者可直接运行观察覆盖效果。压缩包共6个文件含4个Python脚本、1个说明文档与1个License4个脚本分别承担多机器人主算法、单机器人对照、地图绘制与动态结果可视化README介绍快速上手方式项目总大小仅19KB结构紧凑、方便研读。MCTS算法通过节点选择、随机模拟、扩展与回传更新不断优化覆盖路径使多机器人能够有效降低重复覆盖并避免碰撞可视化模块则借助matplotlib等工具将机器人轨迹、覆盖热区与地图叠加展示便于直观判断覆盖质量与算法收敛情况。目前已有1102人学习下载适合想掌握蒙特卡洛树搜索、多机器人路径规划或基于Python实现仿真实验的开发者可作为课程设计、毕业设计或科研项目的参考基线代码简洁易于二次修改。1. 多机器人区域覆盖遇上蒙特卡洛树搜索先别急着贪心多机器人区域覆盖路径规划的目标很直接一组机器人在有限区域内移动用最短时间或最少重复动作把目标区域完整覆盖一遍。实际项目里最常见的翻车不是机器人不会走路而是大家挤在已覆盖区域反复绕圈全局覆盖率迟迟上不去。蒙特卡洛树搜索MCTS解决这个问题时把每一步决策当成对局机器人是棋手联合动作是落子覆盖率是终局得分用几十上百次随机模拟近似出哪个动作更有长期收益。本文用Python拆一个基于MCTS的多机器人覆盖路径规划实现从地图建模、搜索循环讲到参数调优和可视化覆盖新手能跑通的最小闭环也说清楚搜索规模的上界。适合做AGV巡检、室内清扫、消杀与区域测绘路径规划的人。2. 为什么用蒙特卡洛树搜索覆盖问题的奖励要“算总账”2.1 多机器人覆盖的贪心陷阱多机器人覆盖路径规划multi-robot coverage path planning可以建模成栅格地图上的离散决策问题。常见做法是按螺旋或牛耕式直线扫描把区域切成条带然后分给各个机器人。单机器人这么做很管用一旦变成多机器人麻烦就来了区域怎么分、边界怎么协调、两个机器人同时进入窄通道怎么办。先按面积均分任务往往导致某台机器人很快做完另一些还在拥堵区域排队整个系统的完成时间被一台机器人拖高。贪心动作选择也有类似问题。每一步都选“新增覆盖面积最大”的动作会让机器人优先扑向大片空地把通往狭长走廊的入口留到最后。等大区域扫完再回头处理走廊机器人的转弯和回头成本极高覆盖率曲线前期涨得快末期被那几条窄道卡住。MCTS的做法是每次决策都往前模拟若干步把未来新增覆盖与当前动作绑定这样机器人在岔路口会倾向于先走近处的窄通道因为模拟看到了后续收益。这里的关键是奖励的稀疏性和延迟性。已覆盖区域不会带来即时奖励只有跨入未覆盖格子才有反馈。一个动作的价值要等几步甚至十几步后才显现普通贪婪搜索天然短视。MCTS用大量随机对局来估计动作的长期期望收益正好避开这个问题。2.2 从树搜索到多机器人四步流程怎么对应MCTS的核心是四步循环选择、扩展、模拟、回传。在多机器人场景里我把每个时间步所有机器人的动作拼成一个联合动作作为树上的一个分支。状态node当前时间步 t所有机器人坐标与朝向覆盖掩码已经走过的历史动作。动作每个机器人只能从[前、左转、右转、停留]里选一个。N个机器人的原始联合动作组合有4^N种通常要先过滤掉非法动作。模拟从当前节点开始按一个固定策略跑H步累计收益。收益函数由新增覆盖、重复覆盖和步数惩罚构成。回传模拟结束后把总收益沿着刚才expand时走过的路径加回去更新每个节点的累计价值和访问次数。选择阶段用的是UCTUpper Confidence Bound for Trees公式score v/n c * sqrt(log(parent_n)/n)。v/n是该节点的平均收益c是探索常数n是当前节点访问次数。c越大越鼓励探索那些访问少的节点。在多机器人覆盖里这个公式不需要改动只要收益计算得合理就行。为什么不用遗传算法或强化学习遗传算法适合一次性求一个长序列但覆盖过程中状态高度依赖之前的动作交叉算子容易破坏已覆盖区域的结构强化学习需要大量交互样本仿真环境搭建成本高训练不稳定时很难判断是奖励设计问题还是网络问题。MCTS的优势是自带“搜索即训练”不需要梯度也能在运行期持续提升决策质量尤其适合10x10到50x50规模的地图。2.3 状态表示覆盖掩码是整棵树的“内存”实现前最重要的设计是状态怎么保存。我一般用一个二维numpy布尔数组作为覆盖掩码True表示已经被覆盖过机器人坐标用一个(N,2)的整数数组记录朝向用一个(N,)整数数组表示0东1南2西3北。动作历史用tuple存储因为tuple是不可变对象适合作为字典key。下面这个状态类是小项目里最常用的骨架。from dataclasses import dataclass import numpy as np dataclass class State: t: int robot_pos: np.ndarray # shape (N, 2) headings: np.ndarray # shape (N,), 0东 1南 2西 3北 covered: np.ndarray # shape (H, W), dtypebool actions: tuple () # 每一步联合动作的元组 classmethod def from_parent(cls, parent, new_pos, new_headings, new_covered, joint_action): return cls( tparent.t 1, robot_posnp.array(new_pos, copyTrue), headingsnp.array(new_headings, copyTrue), coverednp.array(new_covered, dtypebool, copyTrue), actionsparent.actions (joint_action,), )注意from_parent里必须copy新增掩码。很多第一次做的人直接传引用模拟阶段原地覆盖后父节点和兄弟节点的掩码全被污染搜索树会变成一个能“预知未来”的黑匣子结果却完全不可复现。这个问题我会第5章再展开。这段代码里的State只保存与搜索有关的最小信息。实际项目里想保存路径最好单独记录动作序列而不是把每个状态都挂在树上否则树一深内存就爆。覆盖掩码用bool数组而不是int数组可节省内存numpy做count和非零判断也更方便。动作元组里每个元素也是一元组比如((1,0),(2,1))表示机器人1前进一步、机器人2右转等第3章再详细说。3. Python实现从地图建模到覆盖结果可视化3.1 地图与初始信息建模先写一个最小可用的GridMap。地图我习惯用二维list或numpy数组表示0是可通行空地1是障碍物。为了后续计算边界把障碍物填充成True的布尔mask。机器人初始位置要确保在空地上不能重叠。import numpy as np class GridMap: def __init__(self, grid): self.grid np.array(grid, dtypenp.int8) self.h, self.w self.grid.shape self.obstacle_mask (self.grid 1) self.free_mask ~self.obstacle_mask def in_bounds(self, x, y): return 0 x self.w and 0 y self.h def is_free(self, x, y): return self.in_bounds(x, y) and not self.obstacle_mask[y, x] def neighbors(self, x, y): result [] for dx, dy in [(1, 0), (-1, 0), (0, 1), (0, -1)]: nx, ny x dx, y dy if self.is_free(nx, ny): result.append((nx, ny)) return result这里坐标用(x, y)x是列y是行和数组索引的[y, x]对应。很多新手在这里搞混导致地图和机器人位置看起来歪了。我一般会写一个单元测试放一个5x5的地图机器人从(0,0)往右走应该到(1,0)而不是第二行。这个小代价能省掉后面大量调试图时间。机器人初始位置我保存在数组里robot_pos np.array([[2, 1], [1, 3]], dtypeint)代表两个机器人的起点。注意GPS坐标、像素坐标和栅格坐标在真实系统中往往不是一回事本文直接用栅格坐标。如果要从真实地图转过来需要注意分辨率、原点位置和障碍物膨胀半径这些不在MCTS范围里但对结果影响巨大。3.2 搜索主循环选择、扩展、模拟、回传MCTS的每个节点我保存四个字段当前状态、访问次数n、累计价值v、以及一个字典childrenkey是联合动作元组value是子节点。动作编码用数字0前进1左转2右转3停留。每个机器人的具体转向效果依赖当前朝向因此State里还需要保存朝向数组headings。下面是选择与扩展的代码我按一次搜索迭代来写。class MCTSNode: def __init__(self, state): self.state state self.n 0 self.v 0.0 self.children {} # joint_action - MCTSNode self.parent None def is_fully_expanded(self, valid_actions): return len(self.children) len(valid_actions) def best_child(self, c_puct): log_pn np.log(self.n 1e-6) best None best_score -float(inf) for action, child in self.children.items(): score child.v / (child.n 1e-6) c_puct * np.sqrt(log_pn / (child.n 1e-6)) if score best_score: best_score score best action return best, self.children[best]选择阶段从根节点开始只要当前节点还有未扩展动作就停下否则用best_child往下走一层。这里的关键是每个节点要保存自己的状态而不是共享全局位置。1e-6是防除零不要省。扩展阶段需要生成当前状态下的合法联合动作。我一般会先给每个机器人单独生成候选动作过滤掉会撞墙或走进障碍物的前进一步然后取笛卡尔积。如果某个机器人的所有动作都非法就只留停留动作保证迭代能继续。import itertools def get_valid_actions(state, grid_map): all_actions [] for i in range(state.robot_pos.shape[0]): actions_i [] for action in [0, 1, 2, 3]: if is_action_valid(state, i, action, grid_map): actions_i.append(action) if not actions_i: actions_i [3] all_actions.append(actions_i) return list(itertools.product(*all_actions))这个函数每次都要调用is_action_valid小地图里问题不大地图大了之后应该把每台机器人的可行方向缓存起来因为同一个位置的转向结果是一样的。模拟和回传是整个MCTS最耗时的部分。模拟策略我用“局部贪心随机转向”机器人在每个时间步优先选择能走进未覆盖格子的前进一步如果没有就随机转向。先定义一个根据朝向返回候选格子的辅助函数。def get_step_candidates(x, y, heading, grid_map): # 顺序前方、左转后前方、右转后前方 forward_delta [(1, 0), (0, 1), (-1, 0), (0, -1)] left_delta [(0, -1), (1, 0), (0, 1), (-1, 0)] right_delta [(0, 1), (-1, 0), (0, -1), (1, 0)] result [] for delta, delta_heading in [ (forward_delta[heading], heading), (left_delta[heading], (heading - 1) % 4), (right_delta[heading], (heading 1) % 4), ]: nx, ny x delta[0], y delta[1] if grid_map.is_free(nx, ny): result.append(((nx, ny), delta_heading)) return [p for p, _ in result], [h for _, h in result]这个函数返回两个列表第一个是候选坐标第二个是对应的新朝向。rollout每步选择逻辑就可以直接写。def rollout_policy(state, grid_map): joint [] for i in range(state.robot_pos.shape[0]): x, y state.robot_pos[i] h state.headings[i] candidates, next_heads get_step_candidates(x, y, h, grid_map) chosen None for (nx, ny), nh in zip(candidates, next_heads): if not state.covered[ny, nx]: chosen (nx, ny, nh) break if chosen: nx, ny, nh chosen if nh h: joint.append(0) elif nh (h - 1) % 4: joint.append(1) else: joint.append(2) else: joint.append(3) return tuple(joint) def rollout(state, grid_map, horizon): s State.from_parent(state, state.robot_pos.copy(), state.headings.copy(), state.covered.copy(), ()) for _ in range(horizon): joint_action rollout_policy(s, grid_map) s apply_action(s, joint_action, grid_map, add_coveredTrue) return compute_reward(s, grid_map)apply_action负责把联合动作变成下一个状态包括转向、移动和碰撞检测。我单独拆出来写因为它也是主搜索里扩展子节点时要用的那个函数。def apply_action(state, joint_action, grid_map): new_pos state.robot_pos.copy() new_headings state.headings.copy() new_covered state.covered.copy() for i, action in enumerate(joint_action): x, y state.robot_pos[i] h state.headings[i] if action 0: # 前进 dx, dy [(1, 0), (0, 1), (-1, 0), (0, -1)][h] new_pos[i] [x dx, y dy] elif action 1: # 左转 new_headings[i] (h - 1) % 4 elif action 2: # 右转 new_headings[i] (h 1) % 4 # action 3 停留 # 撞墙、出界、机器人相撞后回退到上一位置 for i in range(state.robot_pos.shape[0]): nx, ny new_pos[i] if not grid_map.is_free(int(nx), int(ny)): new_pos[i] state.robot_pos[i] for i in range(state.robot_pos.shape[0]): for j in range(i 1, state.robot_pos.shape[0]): if np.array_equal(new_pos[i], new_pos[j]): new_pos[i] state.robot_pos[i] new_pos[j] state.robot_pos[j] # 进入未覆盖格子则标记覆盖 for x, y in new_pos: x, y int(x), int(y) if grid_map.is_free(x, y) and not new_covered[y, x]: new_covered[y, x] True return State.from_parent(state, new_pos, new_headings, new_covered, joint_action)这段代码里有一个边界情况我特别提醒如果两台机器人交换位置即A的目标是B的旧位置B的目标是A的旧位置上面的碰撞检测检查的是“目标位置是否相同”无法发现这种对穿。严格实现需要再检测new_pos[i] state.robot_pos[j] and new_pos[j] state.robot_pos[i]的场景否则两机器人可能在窄通道里穿模。小地图上概率不高但要留着。backprop很简单从扩展出的节点沿着树往上更新每个节点的n和v。我常用增量式node.n 1; node.v reward。如果是多线程并行做多个rollout需要加锁但Python的多线程受GIL限制后面会讲替代做法。3.3 主迭代与路径生成把上面的函数拼成一次完整搜索。从根状态开始循环iterations次。每次迭代执行selection、expansion、rollout、backprop。迭代结束后用best_child从根一路取到叶子把记录的联合动作解码成真实轨迹。代码如下def mcts_search(root_state, grid_map, iterations, c_puct, rollout_horizon): root MCTSNode(root_state) for _ in range(iterations): node root # 选择走到一个未完全扩展的节点 while node.children: action, node node.best_child(c_puct) # 扩展生成合法动作创建一个未访问子节点 valid get_valid_actions(node.state, grid_map) if not node.is_fully_expanded(valid): for action in valid: if action not in node.children: child_state apply_action(node.state, action, grid_map) child_node MCTSNode(child_state) child_node.parent node node.children[action] child_node unvisited [a for a in valid if node.children[a].n 0] action unvisited[0] if unvisited else valid[0] node node.children[action] # 模拟 reward rollout(node.state, grid_map, rollout_horizon) # 回传 while node is not None: node.n 1 node.v reward node node.parent return root这里有一个性能取舍valid动作数量很大时全量扩展仍然会吃内存。动作组合作出了3_2节的过滤但如果地图是20x20、4个机器人有效动作还是可能到几十个。我的经验是当valid数量超过20个时只随机抽一个未访问动作扩展别把全部子节点都建出来。原因是MCTS本身要通过访问次数来区分动作好坏一次性全建出来会把大量低质量分支提前压入内存。最后把最优路径画出来。我通常用matplotlib画一个栅格热力图白色未覆盖浅绿色已覆盖深灰色障碍物红色标注机器人位置。这样整张图就能看到覆盖范围。下面是一个最小画图函数。import matplotlib.pyplot as plt def plot_coverage(grid_map, state, save_pathNone): fig, ax plt.subplots(figsize(6, 6)) canvas np.zeros((grid_map.h, grid_map.w, 3), dtypenp.uint8) canvas[grid_map.obstacle_mask] (40, 40, 40) canvas[~grid_map.obstacle_mask] (245, 245, 245) canvas[state.covered grid_map.free_mask] (170, 220, 170) for x, y in state.robot_pos: ax.scatter(x 0.5, y 0.5, colorred, s80, zorder5) ax.imshow(canvas, originupper) ax.set_xticks([]) ax.set_yticks([]) if save_path: fig.savefig(save_path, dpi100, bbox_inchestight)这个图能直接看出覆盖率、重复率和机器人走位。我调参时会把每一步的state都保存下来生成动画逐帧看比只看最终覆盖率可靠得多。如果后续要对接业务系统也可以把这段路径导出成JSON前端用可视化大屏渲染但调试阶段本地热力图是成本最低的验证手段。4. 参数怎么定迭代次数、探索常数、模拟深度与惩罚项4.1 一张参数表解决起步MCTS在覆盖问题里没有一套通用参数但可以先从下面这张表起步再按覆盖率曲线调整。这张表针对10x10到30x30的地图2到4个机器人单轮搜索时间控制在几秒。参数推荐范围作用设置过大设置过小iterations200~2000每轮决策执行多少次模拟耗时线性增长搜索不充分决策近似贪心c_puct0.5~2.0探索与利用平衡过度探索路径跳跃过早锁定局部最优rollout_horizon10~30模拟的展望步数模拟太慢远期奖励噪声大看不到延迟收益step_penalty0.1~0.5每走一步扣分抑制绕圈机器人不敢远行路径冗余、重复覆盖collision_penalty1.0~3.0两机器人相撞扣分过度避让保守机器人扎堆这张表是经验值不是理论最优。地图分辨率提高后先把地图降到适当尺寸再调参是更省时间的做法。代码里的reward函数我习惯写成def compute_reward(state, grid_map): step_cost state.t * step_penalty collision_cost collision_count * collision_penalty reward state.covered.sum() * 10 - step_cost - collision_cost return reward其中collision_count需要在apply_action里统计一次我为了保持示例简短没有放在State里实际项目可以直接加一个字段。新增覆盖的权重10是相对步数惩罚而言的目的是让“覆盖一整片新区域”明显优于“原地磨蹭”。4.2 用覆盖率曲线倒推参数光看最终覆盖率不够我会记录每次搜索迭代后当前最优路径对应的覆盖率。方法是搜索结束后沿最优子节点路径跑一遍统计经过的格子里未覆盖转为已覆盖的数量除以全部空地数量。每次迭代都做一遍太贵可以每迭代50次采样一次。然后画出“覆盖率-迭代次数”曲线。def eval_policy(root, grid_map): node root total_free grid_map.free_mask.sum() while node.children: _, node node.best_child(0.0) covered set() for x, y in node.state.robot_pos: if grid_map.is_free(int(x), int(y)): covered.add((int(x), int(y))) return len(covered) / total_free # 搜索过程中每50次迭代调用一次eval_policy记录到数组里这条曲线接近对数增长是正常的但如果前200次迭代覆盖率纹丝不动大概率是c_puct太大模拟策略一直在随机转如果曲线很快封顶但实际轨迹明显绕远路说明step_penalty太小。我把这个曲线叫“搜索的心电图”它比任何玄学调参都准。4.3 算力与实时性的取舍MCTS最大的问题是每次决策都要重复搜索。如果机器人每0.5秒需要给出下一步动作2000次模拟很可能来不及。常见做法是固定迭代次数而不是固定时间或使用渐进式策略先快速用低粒度地图选目标区域再做局部精细搜索。Python实现里还要注意numpy的cache尽量将整棵树的state做成连续对象而不要散落。我的经验是30x30地图、2个机器人、iterations500单步决策大约0.8到2秒勉强够离线规划用真正实时控制建议把模拟部分搬到C或numba加速Python只负责组装状态和可视化。另外在搜索时不要每次都np.sum(state.covered)可以维护一个增量字段每次新增覆盖时给父节点提交增量减少重复计算。5. 避坑清单我在多机器人MCTS覆盖里踩过的5个坑5.1 坑1rollout策略太随机回传奖励稀疏得没法收敛现象跑第一次搜索后日志里平均回报一直是负数覆盖率曲线像锯齿甚至搜索次数越多效果越差。原因rollout里每个机器人每一步都均匀随机选动作大量模拟落进撞墙、互相堵路、原地转向的无效轨迹偶尔扫到新格子的样本被噪声淹没。解决rollout策略改成“惯性优先”。先让机器人保持当前方向前进如果前方有障碍或已覆盖率高再随机转一个方向。这样模拟轨迹更接近真实可通行路径回报方差明显下降。另外在reward里加一个基础步数惩罚让无效动作收到负反馈而不是期望稀疏的正向奖励。我在第3章的rollout_policy里已经体现这个思想。5.2 坑2联合动作空间指数膨胀树宽爆炸现象3个机器人每个有4个动作一扩展就是64个子节点4个机器人到了4^4256。搜索还没走几层内存就涨到几个GB程序被OOM干掉。原因树的每个节点都存一个State状态里有完整覆盖掩码numpy数组copy一次的成本是O(H*W)。动作组合一旦乘起来节点数量和内存都失控。解决三个手段一起用。第一过滤无效动作比如前方是墙的“前进”直接删掉。第二不采用全量扩展而是每次只随机挑一个未访问动作展开保留足够的探索。第三对等价状态做节点合并比如两台机器人互换位置如果地图对称性强可以视为同一状态。对小规模场景我只做前两个就能把树宽压到20以内。5.3 坑3子节点状态复制不干净搜索树出现“预知”现象模拟完某一条分支后另一条完全不同路径的覆盖率也突然变高生成的路径在未走路段上画满了颜色。原因expand时直接用了父节点的State引用或者对同一numpy数组做了浅拷贝。rollout里修改covered时因为底数组共享把父节点也给改了。整个搜索树变成一个有记忆的怪胎结果看似聪明实际是状态污染。解决所有状态对象遵循“只读父、新建子”的原则。创建一个子节点时强制对covered调用np.array(..., copyTrue)机器人坐标和朝向也copy。不要为省这点性能去掉copy因为copy的代价是O(H*W)而一次错误模拟会污染整棵子树修复成本高得多。5.4 坑4覆盖掩码用原地操作同一份数组在兄弟节点间串数据现象某个分支的机器人明明从没到过右下角最终热力图却显示右下角被覆盖了。原因这个坑比坑3更隐蔽。在State.from_parent里虽然copy了掩码但在rollout中每一步又对new_covered做原地更新比如new_covered[ny, nx] True如果new_covered是上一轮状态的浅拷贝兄弟节点就会受影响。解决我把它归结为一条纪律任何从旧状态衍生新状态的操作都必须走State.from_parent且函数内部再copy一次不直接修改state.covered的任何元素。为了查这类问题我会在每次扩展时校验child.state.covered is not parent.state.covered用断言拦住引用共享。注意调试状态污染时最快的定位方法是打印父节点和子节点numpy数组的__array_interface__[data]地址如果地址相同说明还在共享内存。5.5 坑5奖励只算覆盖率机器人原地打转、重复率爆炸现象覆盖率最终接近100%但总步数比别人多一倍机器人反复碾压同一片区域可视化里能看到轨迹像毛线团。原因reward里没有对步数做惩罚或惩罚太小。MCTS发现原地小范围移动也能通过重复覆盖累积正的收益于是最优解变成了刷步数。解决在reward里加-step_penalty * state.t并且对已经覆盖过的格子再次走过时额外扣overlap_penalty。我会把step_penalty从0.1开始尝试观察最终路径步数是否明显下降。这里有一条经验法则最终路径步数不应超过理论下界空地数量 / 机器人数量的2.5倍超过这个值时优先检查惩罚项的权重。6. 再往前走一步树复用、并行搜索与三个验证指标加速方面我常用的两个手段是树复用和多进程扰动。树复用思路很简单上一轮搜索结束机器人执行了联合动作a那么当前节点就是上一轮根节点的一个子节点直接把那个子树作为新一轮搜索的根保留访问次数和收益。空出来的算力继续往这棵树上加模拟次数决策会越来越稳。这种做法的前提是状态没有外部噪声变化如果地图上有动态障碍就不能直接继承需要额外检查障碍掩码变化。多进程扰动适合离线批量规划开4个worker每个worker用不同的随机种子独立搜索各算500次迭代然后选覆盖率和步数的加权评分最高的路径。这个可以避免单一搜索陷入随机种子带来的偏好。注意worker之间不要共享状态对象进程建完就返回路径和评分最后在主进程合并简单可靠。验证覆盖结果时我会额外算三个指标覆盖率已覆盖空地/全部空地重复率总移动步数中落在已覆盖格子的比例完成时间最后一个机器人停止移动的时间步。它们比单一覆盖率更早暴露路径冗余。指标公式预期覆盖率已覆盖格点数 / 空地总数不低于95%重复率重复步入次 / 总步数尽量低于30%最大完成时间max(每机器人最后一步)接近空地数/机器人数的1.5~2倍把这三个指标配合逐帧可视化热力图基本能定位绝大多数问题。我的习惯是调完一轮参数后把最终路径每一帧存成PNG再用ffmpeg合成视频肉眼扫两遍比看十个指标更直接。这个方案不一定是最优覆盖规划但它是把MCTS落到多机器人场景里最可控的一条路径希望帮到你。本文还有配套的精品资源点击获取