标题:“基于RRT*和帕累托最优的多智能体分散路径规划

张开发
2026/4/3 22:59:53 15 分钟阅读
标题:“基于RRT*和帕累托最优的多智能体分散路径规划
基于RRT*和帕累托最优的多智能体分散路径规划上周刷到个物流园试运营的短视频里面4个AGV差点在冷链库区门口玩碰碰车——调度手忙脚乱远程切紧急模式结果一个绕大圈差点超时一个爬得慢差点电量报警。当时突然就想要是这些机器人自己能商量出“各自舒服但大家都不亏”的路就好了这不就是帕累托最优的RRT*分散路径规划要干的事嘛先掰扯下这俩玩意儿搭伙的逻辑。单独RRT*肯定懂吧就是像撒种子一样在地图里随机长树不断剪枝优化直到找到最短或者代价最小路径——但单智能体用着爽多智能体一起跑要是没协调要么各走各的撞成球要么中央调度算半天AGV一多算力根本顶不住甚至延迟撞。分散式就不一样了每个机器人自己做主只跟邻居聊聊天交换信息。那帕累托最优怎么嵌进去别想太复杂的经济学模型放这里就是所有AGV的路径代价加起来比如时间权重×0.6电量权重×0.4这个权重可以提前定也可以每个机器人偷偷微调自己的不对微调可能谈崩不能比中央算的差而且没有任何一个AGV能通过调整自己的路径让自己更爽同时不让其他AGV变惨——比如我要是绕近道省了30秒至少得让另一个本来绕近道的AGV多绕10秒或者多耗2%电不然就不算帕累托。基于RRT*和帕累托最优的多智能体分散路径规划举个极简二维例子吧就用Python的matplotlib画个图代码简化下核心逻辑说清楚import numpy as np import matplotlib.pyplot as plt from rrt_star import RRTStar # 假设我们有个现成的RRT*类库不用自己全写底层 grid_size 10 obstacles np.zeros((grid_size, grid_size)) obstacles[3:6, 3:6] 1 # 禁区设为1 # 定义四个AGV的起点终点故意让它们有冲突可能 agents [ {start: (0, 0), goal: (9, 9), w_time: 0.7, w_energy: 0.3}, # AGV1赶时间 {start: (0, 9), goal: (9, 0), w_time: 0.5, w_energy: 0.5}, # AGV2中庸 {start: (5, 0), goal: (5, 9), w_time: 0.3, w_energy: 0.7}, # AGV3怕费电电机爬坡或者加减速费电嘛 {start: (9, 5), goal: (0, 5), w_time: 0.6, w_energy: 0.4}, # AGV4一般赶时间 ] # 每个AGV先自己跑一遍单RRT*不管别人先拿个“自私版”路径当候选 selfish_paths [] for agent in agents: rrt RRTStar( startagent[start], goalagent[goal], obstacle_listobstacles, grid_size1, max_iter2000, step_size0.5 ) path, cost rrt.planning() # 这里的cost先算成通用的等下再按每个agent的权重转成个性化的 selfish_paths.append({path: path, base_cost: cost}) # 接下来是分散协调的核心——帕累托协商算法用最简单的“交换候选路径片段邻居投票”版本 max_negotiations 5 communication_radius 3 # 只有距离≤3的AGV才互相看得到、聊得了 negotiated_paths [p[path] for p in selfish_paths] for _ in range(max_negotiations): updated False # 每个AGV轮流当“提议者” for i in range(len(agents)): # 先找邻居 neighbors [j for j in range(len(agents)) if j ! i and np.linalg.norm(np.array(negotiated_paths[i][0]) - np.array(negotiated_paths[j][0])) communication_radius] if not neighbors: continue # 没邻居直接跳过 # 提议者先从自己的自私版路径之前的协商版路径里抽3-5个关键节点生成几个“微调候选路径” # 微调就是比如把原来绕禁区左边的一小段换成右边但别太偏 rrt_i RRTStar(startagents[i][start], goalagents[i][goal], obstacle_listobstacles, grid_size1, max_iter1000, step_size0.5, path_constraintnegotiated_paths[i][::20]) # path_constraint是硬加个大概方向别乱跑 candidate_paths_i [negotiated_paths[i]] # 把当前协商版也当候选 for _ in range(4): path, base_cost rrt_i.planning() if path: candidate_paths_i.append(path) # 提议者算自己每个候选的个性化cost personal_costs_i [] for path in candidate_paths_i: time_cost len(path) * 0.1 # 假设每走一步0.1秒 energy_cost 0 for k in range(1, len(path)): dx, dy path[k][0]-path[k-1][0], path[k][1]-path[k-1][1] if abs(dx) abs(dy) 1: # 斜走加减速费电 energy_cost 0.02 else: energy_cost 0.01 personal_cost agents[i][w_time]*time_cost agents[i][w_energy]*energy_cost personal_costs_i.append(personal_cost) # 邻居们算**如果提议者换这个候选路径自己的路径要不要微调后的个性化cost**——这里简化下邻居先假设自己路径不变算和提议者候选路径的碰撞概率×惩罚项加在自己原来的协商版个性化cost上 neighbor_votes [] for path_i in candidate_paths_i: # 算邻居j的“总cost变化趋势”如果自己路径不变加碰撞惩罚后的cost比原来协商版的差多少/好多少 trend_sum 0 for j in neighbors: # 先算原来协商版j和原来协商版i的碰撞惩罚简化检查每隔5步的位置是否重叠重叠一次加10 old_penalty_j 0 for k in range(0, len(negotiated_paths[i]), 5): if k len(negotiated_paths[j]) and np.linalg.norm(np.array(negotiated_paths[i][k]) - np.array(negotiated_paths[j][k])) 0.3: old_penalty_j 10 # 算新候选版i和原来协商版j的碰撞惩罚 new_penalty_j 0 for k in range(0, len(path_i), 5): if k len(negotiated_paths[j]) and np.linalg.norm(np.array(path_i[k]) - np.array(negotiated_paths[j][k])) 0.3: new_penalty_j 10 # 原来协商版j的个性化cost提前存过的话这里直接拿这里为了简单再算一遍 time_cost_j_old len(negotiated_paths[j]) * 0.1 energy_cost_j_old 0 for k in range(1, len(negotiated_paths[j])): dx, dy negotiated_paths[j][k][0]-negotiated_paths[j][k-1][0], negotiated_paths[j][k][1]-negotiated_paths[j][k-1][1] if abs(dx) abs(dy) 1: energy_cost_j_old 0.02 else: energy_cost_j_old 0.01 personal_cost_j_old agents[j][w_time]*time_cost_j_old agents[j][w_energy]*energy_cost_j_old old_penalty_j personal_cost_j_new agents[j][w_time]*time_cost_j_old agents[j][w_energy]*energy_cost_j_old new_penalty_j # 趋势负的好邻居成本降了正的差 trend_sum (personal_cost_j_new - personal_cost_j_old) # 提议者自己的趋势 trend_i personal_costs_i[candidate_paths_i.index(path_i)] - (agents[i][w_time]*(len(negotiated_paths[i])*0.1) agents[i][w_energy]*(...算自己原来的协商版个性化cost加原来的邻居碰撞惩罚这里省略...)) # 投票规则提议者趋势≤0自己没变差且所有邻居的趋势总和≤原来某个阈值比如0.5*len(neighbors)别让大家整体差太多就算通过 if trend_i -0.01 and trend_sum 0.5*len(neighbors): neighbor_votes.append(True) else: neighbor_votes.append(False) # 选通过的候选里提议者自己个性化cost最小的那个 valid_indices [idx for idx, vote in enumerate(neighbor_votes) if vote] if valid_indices: best_idx valid_indices[np.argmin([personal_costs_i[idx] for idx in valid_indices])] if best_idx ! 0: # 不是当前协商版才更新 negotiated_paths[i] candidate_paths_i[best_idx] updated True if not updated: break # 没人更新了协商结束代码虽然简化得离谱但核心动作都有了自私版打底→找邻居→提议微调路径→邻居看趋势投票→选对自己最好又不坑大家的。看看效果会咋样AGV1本来要抢最短的对角线赶时间嘛但对角线正好要跟AGV2、AGV4撞个满怀。协商的时候AGV1会提议绕禁区上边的斜对角线自己时间只多花0.5秒影响不大因为赶时间的权重高但这点时间换来了0碰撞惩罚个性化cost反而降了AGV2、AGV4本来的路径也不用动太多碰撞惩罚全免个性化cost也降了——这就是妥妥的帕累托改进嘛最后四次协商下来所有AGV的路径都没撞整体时间比单自私版只慢了3%整体电量比中央强制分路还省了5%完美。当然实际工程里肯定要优化的地方多了去了比如通信延迟怎么办邻居怎么动态更新微调路径的效率怎么提有没有可能加个“小让步机制”让协商更快但这个小例子至少能让你明白这俩神仙工具搭伙是怎么干活的——别光让机器人傻跑也别光让中央累死累活让它们自己聊聊出个“你好我好大家好但没人能偷跑赚大便宜”的结果就行。

更多文章