简介:这份资源围绕基于 Python 实现多 AGV 路径规划算法展开,面向希望入门或进阶学习路径规划技术的学生与开发者,可作为毕业设计、课程设计、大作业或工程实训的参考项目,帮助理解多台自动导引车在共享地图中的路径搜索与协同调度思路。压缩包共 3 个文件,均为 py 源码,整体约 3KB,体量轻便,涵盖地图构建、点位定义与算法核心等模块,便于直接阅读与二次修改。目前已有 790 人学习下载,说明该方向具备一定关注度。读者可从中获取多 AGV 场景下的算法实现框架、随机地图生成方式以及节点与路径的数据组织方法,并据此扩展冲突避让、任务分配等策略,快速搭建自己的实验原型,适合作为算法学习与项目起步的参考素材。
1. 多 AGV 路径规划:从单机 A* 到多机协同,Python 能扛到什么程度
仓库里三台 AGV 同时接到任务,两台在十字路口顶头相遇,第三台堵在它们后面——这不是段子,是很多做多 AGV 调度的人第一次跑仿真时看到的画面。单台 AGV 用 A* 跑一条最短路径,代码不到一百行就能出结果;但一旦 AGV 数量变成三台、五台、十台,问题就从「找一条路」变成了「多台车怎么分路、怎么避让、怎么不死锁」。这个标题要解决的核心,就是用 Python 把多 AGV 的路径规划从算法到可运行代码完整落地,覆盖环境建模、单机规划、多机冲突消解、仿真验证这条链路。
适合谁看:做过单机 A* 或 Dijkstra、现在要扩展到多车场景的工程师;用 Python 做调度仿真、需要快速验证算法思路的研究者;以及被「多车死锁」折磨过、想搞清楚时间窗和优先级到底怎么配合的从业者。下面按「环境怎么建 → 单机怎么跑 → 多机怎么协调 → 坑在哪 → 怎么验证」的顺序推。
2. 环境建模与单机 A*:先把一张能跑的地图搭出来
2.1 栅格地图还是拓扑图,选错了后面全白干
多 AGV 路径规划的第一步不是写算法,是决定地图怎么表示。常见两种做法:栅格地图(把仓库地面切成 N×M 的格子,每格标记可通行/障碍)和拓扑图(把通道抽象成节点和边,用邻接矩阵或邻接表存储)。
栅格地图的优点是直观、容易和激光雷达/摄像头数据对接,A*、Dijkstra、JPS 都能直接跑;缺点是格子一多,搜索空间爆炸,十台 AGV 同时搜索时内存和耗时都上去了。拓扑图的优点是搜索快、天然适合多车调度(节点就是路口和工位),缺点是需要提前人工或半自动建图,对动态障碍的适应差。
我一般会这样选:仓库通道规整、路口明确的场景用拓扑图,因为多 AGV 的冲突基本发生在路口和窄通道,拓扑图能直接把「谁先过路口」变成节点占用问题;环境复杂、障碍不规则、需要和感知模块对接的场景用栅格地图,先保证单机能跑通,再考虑多机。
下面用栅格地图做示例,因为它更容易复现,不依赖额外建图工具。
import numpy as np import heapq # 0 = 可通行, 1 = 障碍 grid = np.array([ [0, 0, 0, 0, 0, 0], [0, 1, 1, 0, 1, 0], [0, 0, 0, 0, 1, 0], [0, 1, 0, 1, 1, 0], [0, 0, 0, 0, 0, 0], ]) def neighbors(node, grid): """返回四邻域可通行节点""" rows, cols = grid.shape r, c = node for dr, dc in [(-1,0),(1,0),(0,-1),(0,1)]: nr, nc = r + dr, c + dc if 0 <= nr < rows and 0 <= nc < cols and grid[nr][nc] == 0: yield (nr, nc)这段代码定义了一个 5×6 的栅格地图和四邻域扩展函数。grid里 0 表示可走、1 表示障碍,neighbors只返回上下左右四个方向的可通行格子。参数上,如果你想让 AGV 走八方向,把方向列表改成八个偏移量即可,但要注意八方向下对角线穿墙问题——两个相邻障碍的对角不能直接走,需要额外判断。
2.2 用 heapq 写一个能复用的 A* 核心
A* 的关键是估价函数f = g + h,g是起点到当前点的实际代价,h是当前点到终点的启发式估计。栅格地图上h常用曼哈顿距离(四邻域)或欧几里得距离(八邻域)。Python 标准库的heapq就是现成的最小堆,不需要自己写优先队列。
def astar(grid, start, goal): """单 AGV A* 路径规划,返回路径列表""" open_set = [(0, start)] came_from = {} g_score = {start: 0} h = lambda a, b: abs(a[0]-b[0]) + abs(a[1]-b[1]) # 曼哈顿距离 while open_set: _, current = heapq.heappop(open_set) if current == goal: path = [current] while current in came_from: current = came_from[current] path.append(current) return path[::-1] for nxt in neighbors(current, grid): tentative_g = g_score[current] + 1 if nxt not in g_score or tentative_g < g_score[nxt]: came_from[nxt] = current g_score[nxt] = tentative_g f = tentative_g + h(nxt, goal) heapq.heappush(open_set, (f, nxt)) return [] # 无路径open_set存的是(f值, 节点)元组,heapq按第一个元素排序。came_from记录每个节点的前驱,用于回溯路径。g_score存起点到各节点的最小代价。注意heapq在 f 值相同时会比较第二个元素(节点元组),所以节点必须是可比较的,这里用(r, c)元组没问题。
调用方式:
path = astar(grid, (0, 0), (4, 5)) print(path) # [(0,0), (0,1), ..., (4,5)]单机跑通之后,你会得到一条最短路径。但多 AGV 场景下,每台车都跑一次 A*,路径之间必然有重叠。下一步就是处理这些重叠。
3. 多 AGV 冲突消解:时间窗、优先级和预约表怎么配合
3.1 先分清三类冲突:顶点冲突、边冲突、跟随冲突
多 AGV 路径冲突不是一种,常见分三类:
| 冲突类型 | 表现 | 例子 |
|---|---|---|
| 顶点冲突 | 两台车同一时刻到达同一节点 | A 和 B 都要在 t=5 经过路口 (2,3) |
| 边冲突 | 两台车同一时刻交换位置 | A 从 (1,2) 到 (1,3),B 从 (1,3) 到 (1,2) |
| 跟随冲突 | 后车追上前车但无法超车 | A 在 B 后面,B 停在工位装卸货 |
顶点冲突和边冲突会导致碰撞,跟随冲突会导致死锁或效率下降。很多教程只讲顶点冲突,实际跑起来边冲突和跟随冲突才是血泪经验的来源。
3.2 时间窗预约表:把「谁先过」变成可计算的占用
时间窗法的核心思想:每台 AGV 的路径上每个节点都有一个到达时间,把「节点 + 时间」作为预约项存进一张全局表。规划新路径时,检查候选节点的预约情况,如果该节点在目标时间已被占用,就等待或绕行。
class ReservationTable: def __init__(self): self.table = {} # {(r,c): {time: agv_id}} def is_free(self, node, t, agv_id): """检查 node 在 t 时刻是否可被 agv_id 占用""" if node not in self.table: return True occupants = self.table[node] # 同一时刻只允许一台车 if t in occupants and occupants[t] != agv_id: return False return True def reserve(self, node, t, agv_id): self.table.setdefault(node, {})[t] = agv_id def release(self, node, t, agv_id): if node in self.table and self.table[node].get(t) == agv_id: del self.table[node][t]ReservationTable用嵌套字典存节点 -> 时间 -> AGV编号。is_free检查某时刻是否被别的车占用,reserve和release负责登记和释放。参数上,时间粒度要和 AGV 运动模型匹配:如果 AGV 每步耗时 1 秒,时间粒度就是 1 秒;如果速度可变,需要按实际到达时间插值。
3.3 带时间窗的 A*:在搜索时就把冲突考虑进去
把预约表嵌进 A* 的扩展逻辑:每次生成邻居节点时,计算到达该节点的时间,如果预约表里该时刻已被占用,就尝试等待一个时间步或跳过该节点。
def astar_with_reservation(grid, start, goal, start_time, agv_id, table): open_set = [(0, start, start_time)] came_from = {} g_score = {(start, start_time): 0} h = lambda a, b: abs(a[0]-b[0]) + abs(a[1]-b[1]) while open_set: _, current, t = heapq.heappop(open_set) if current == goal: path = [(current, t)] state = (current, t) while state in came_from: state = came_from[state] path.append(state) return path[::-1] for nxt in neighbors(current, grid): nt = t + 1 if not table.is_free(nxt, nt, agv_id): # 尝试等待一步 nt_wait = nt + 1 if not table.is_free(nxt, nt_wait, agv_id): continue nt = nt_wait state = (current, t) nxt_state = (nxt, nt) tentative_g = g_score[state] + (nt - t) if nxt_state not in g_score or tentative_g < g_score[nxt_state]: came_from[nxt_state] = state g_score[nxt_state] = tentative_g f = tentative_g + h(nxt, goal) heapq.heappush(open_set, (f, nxt, nt)) return []这里状态从节点变成了(节点, 时间),因为同一节点在不同时间到达的代价不同。is_free检查失败时先尝试等待一步,还不行就跳过。参数上,等待步长可以设成 1 个时间单位,也可以根据 AGV 加减速模型设成动态值。
多台 AGV 按优先级依次规划:优先级高的先规划并占用预约表,优先级低的后规划、避让。优先级可以按任务紧急程度、AGV 编号、剩余电量等设定。常见做法是按任务截止时间排序,紧急任务优先。
4. 避坑与排查:多 AGV 跑不起来时先看这几条
4.1 现象:仿真跑几分钟后所有 AGV 都不动了
原因:死锁。两台或多台 AGV 互相等待对方释放节点,形成循环等待。时间窗法只检查单节点占用,不检查「等待链」是否成环。
解决:加死锁检测。定期检查等待图,如果发现环,强制让优先级最低的 AGV重新规划或后退一步。简单做法是给每台 AGV 设一个等待超时,超过 N 步没动就触发重规划。
4.2 现象:路径明明存在,A* 返回空列表
原因:预约表把唯一通道全占了,或者is_free判断过严,把未来所有时间都锁死。
解决:检查预约表的释放逻辑。AGV 离开节点后要及时release,否则预约表只增不减。另外等待策略不要无限等待,设最大等待步数,超了就绕行或重新规划。
4.3 现象:AGV 在路口反复启停,效率极低
原因:时间窗粒度过细或过粗。粒度过细导致 AGV 频繁等待微小时间差;粒度过粗导致路口利用率低。
解决:根据 AGV 实际运动速度调整时间粒度。一般取 AGV 通过一个栅格所需时间的 1/2 到 1 倍作为粒度。如果 AGV 速度是 1 m/s、栅格 0.5 m,那粒度取 0.25~0.5 秒比较合适。
4.4 现象:多台 AGV 路径长度差异巨大,有的跑死有的闲着
原因:优先级静态分配,高优先级 AGV 总是优先占用最短路径,低优先级 AGV 被迫绕远。
解决:引入动态优先级或轮换机制。比如每完成 N 个任务轮换一次优先级,或者按「当前任务剩余时间 / 预计完成时间」动态调整。也可以给每台 AGV 设路径长度上限,超过就触发重新分配任务。
4.5 现象:Python 跑 10 台 AGV 就卡到没法看
原因:每台 AGV 每步都重新跑完整 A*,搜索空间随 AGV 数量线性增长,预约表查询也是 O(1) 但常数不小。
解决:路径缓存 + 增量规划。AGV 路径不变时不重算;只在冲突发生时对受影响的两三台车局部重规划。另外用numpy存栅格、用set存预约节点、避免在热循环里做字典嵌套查询,都能明显提速。
5. 验证与进阶:怎么确认你的多 AGV 方案真的能用
5.1 用三个指标判断方案是否及格
跑通不等于能用。我一般看三个指标:
| 指标 | 含义 | 及格线参考 |
|---|---|---|
| 任务完成率 | 规定时间内完成的任务数 / 总任务数 | > 95% |
| 平均等待时间 | AGV 因避让等待的总时间 / 任务数 | 单任务 < 20% 行驶时间 |
| 死锁次数 | 仿真全程触发死锁检测的次数 | 0(或可恢复) |
这三个指标比「路径最短」更贴近实际。很多方案单看路径长度很漂亮,一跑多车就死锁或等待爆炸。
5.2 用 matplotlib 把轨迹画出来,肉眼排查冲突
数字指标之外,可视化是最快的排查手段。把每台 AGV 的(节点, 时间)轨迹画成时空图,冲突点一目了然。
import matplotlib.pyplot as plt def plot_trajectories(trajectories): """trajectories: {agv_id: [(r,c,t), ...]}""" for agv_id, traj in trajectories.items(): times = [t for _, _, t in traj] nodes = [r * 10 + c for r, c, _ in traj] # 简单编码节点 plt.plot(times, nodes, marker='o', label=f'AGV {agv_id}') plt.xlabel('Time step') plt.ylabel('Node (r*10+c)') plt.legend() plt.title('AGV Trajectories') plt.show()横轴是时间步,纵轴是节点编码,每条线是一台 AGV 的轨迹。两条线交叉说明同一时刻同一节点被两台车占用,需要检查预约表逻辑。如果线长时间水平,说明 AGV 在等待,检查等待策略是否过保守。
5.3 从时间窗到 CBS:什么时候该换算法
时间窗法实现简单、跑得快,适合 AGV 数量少(< 10 台)、通道规整的场景。但如果 AGV 数量多、冲突密集,时间窗法容易陷入「等待-重规划」循环。这时候可以看CBS(Conflict-Based Search):上层搜索冲突树,下层对每台 AGV 单独跑 A*,冲突时给相关 AGV 加约束再搜。CBS 能找到最优解,但计算量比时间窗大一个量级。
我的习惯是:先用时间窗法跑通,确认业务逻辑没问题;如果指标不达标,再考虑 CBS 或基于强化学习的调度。不要一上来就上复杂算法,多 AGV 的瓶颈往往在环境建模和冲突定义,不在搜索算法本身。
最后说一个我踩过的坑:早期做多 AGV 时,我花了两周优化 A* 的启发式函数,结果发现真正的瓶颈是预约表在每步都做全表扫描。后来把预约表改成按节点索引的字典,速度直接翻了五倍。先定位瓶颈,再优化算法,这个顺序在多 AGV 场景里尤其重要。希望帮到你。
本文还有配套的精品资源,点击获取