简介这份资源面向计算机、人工智能、自动化等专业的在校学生与教研人员提供一套基于深度强化学习的目的楼层预约调度算法与多智能体电梯群控系统设计实现可用于毕业设计、课程设计或项目立项演示。压缩包共30个文件以13个Python源码文件为核心涵盖DQN、Sarsa、Sarsa(λ)、Q-Learning等多种强化学习算法的对比实现并配有14个pyc编译文件、1份doc方案说明、1份md项目报告及1个实验结果压缩包整体约1.94MB目录按算法版本分模块组织便于横向对照。资源已有532人学习下载读者可获取完整可运行代码、算法对比实验数据与项目报告文档快速理解多智能体电梯调度中状态建模、奖励设计与策略收敛过程并在此基础上修改扩展功能。1. 从一部电梯到六部电梯多智能体电梯群控到底难在哪早高峰写字楼里六部电梯同时响应几十层楼的呼叫传统先来先服务调度算法会让某部电梯层层停、某部空跑平均候梯时间轻松突破三分钟。Python基于深度强化学习的目的楼层预约调度算法的多智能体电梯群控系统设计要解决的就是这个场景把每部电梯看作一个独立智能体用深度强化学习让它们学会协商分配任务而不是靠固定规则硬扛。适合有Python基础、想入门多智能体强化学习又缺真实落地场景的开发者也适合做调度算法优化、想对比传统启发式方法边界的工程师。读完你能拿到一套可复现的训练与评估路径知道状态怎么设计、奖励怎么给、参数怎么调、哪里最容易翻车。2. 电梯群控的状态、动作与奖励怎么设计2.1 为什么把每部电梯当成独立智能体电梯群控的本质是一个分布式决策问题。六部电梯各自有位置、方向、载重、门状态乘客呼叫分布在各个楼层任何一部电梯的决策都会改变其他电梯面临的环境。如果用一个中心智能体统一控制所有电梯动作空间会随电梯数量指数膨胀六部电梯每部可选动作按十个算联合动作空间就是十的六次方训练时采样效率极低。常见做法是采用多智能体强化学习里的集中训练分散执行框架。每部电梯是一个智能体训练时允许共享全局状态信息来稳定价值函数执行时每部电梯只依赖自身观测和有限的邻居信息做决策。这样动作空间从联合空间降回单梯空间六部电梯各自输出自己的目标楼层或方向选择整体调度效果通过奖励函数耦合。我一般会把智能体数量控制在四到八部之间做实验。少于四部时多智能体协商带来的收益不明显传统调度算法已经够用超过八部时训练不稳定性和通信开销会快速上升需要引入注意力机制或图神经网络来压缩观测那是另一个量级的工程复杂度。2.2 状态空间每部电梯该看到什么状态设计直接决定策略能不能学到有效调度。每部电梯的观测至少包含四类信息自身状态、呼叫状态、邻居状态、时间上下文。自身状态包括当前楼层、运行方向、门开关状态、当前载客数、已分配但未完成的呼叫列表。呼叫状态包括各楼层上行和下行呼叫的等待时间等待时间是关键没有它智能体学不会优先响应久等乘客。邻居状态包括其他电梯的楼层和方向这是实现协商的基础但不需要全量共享通常只给最近的三部电梯信息就够。时间上下文包括当前时段标识早高峰、午间、晚高峰的客流模式差异很大把时段作为离散特征输入能让同一套策略适应不同场景。import numpy as np def build_observation(elevator, all_elevators, hall_calls, current_step): 构建单部电梯的观测向量 elevator: 当前电梯对象含 floor, direction, load, door_state all_elevators: 所有电梯列表 hall_calls: 各楼层呼叫记录含楼层、方向、等待步数 current_step: 当前时间步 obs [] # 自身状态楼层归一化、方向、载重比例、门状态 obs.append(elevator.floor / 20.0) obs.append(elevator.direction) # -1 下行, 0 静止, 1 上行 obs.append(elevator.load / elevator.capacity) obs.append(1.0 if elevator.door_open else 0.0) # 呼叫状态最近 10 个未响应呼叫的楼层和等待时间 pending [c for c in hall_calls if not c.assigned] pending.sort(keylambda c: c.wait_steps, reverseTrue) for i in range(10): if i len(pending): obs.append(pending[i].floor / 20.0) obs.append(min(pending[i].wait_steps / 100.0, 1.0)) else: obs.append(0.0) obs.append(0.0) # 邻居状态最近 3 部电梯的楼层和方向 neighbors sorted(all_elevators, keylambda e: abs(e.floor - elevator.floor)) neighbors [e for e in neighbors if e.id ! elevator.id][:3] for i in range(3): if i len(neighbors): obs.append(neighbors[i].floor / 20.0) obs.append(neighbors[i].direction) else: obs.append(0.0) obs.append(0.0) # 时间上下文用正弦编码表示一天中的时段 obs.append(np.sin(2 * np.pi * current_step / 1440)) obs.append(np.cos(2 * np.pi * current_step / 1440)) return np.array(obs, dtypenp.float32)这段代码把观测向量长度固定在 4 20 6 2 32 维。楼层除以 20 是归一化假设建筑 20 层实际项目按楼层数调整。等待步数除以 100 再截断到 1是防止极端长等待把数值撑爆。邻居只取最近三部是为了控制观测维度同时保留局部协商能力。时间用正弦余弦编码而不是直接给小时数是因为 23 点和 0 点在物理上连续直接给数值会让网络误以为它们相距很远。注意呼叫列表排序后只取前 10 个意味着超过 10 个未响应呼叫时后面的呼叫不会进入观测。如果建筑楼层多、客流大这个截断会导致智能体忽略部分乘客需要把 10 调大或改用注意力池化。2.3 动作空间离散选层还是连续给方向动作空间有两种常见设计。第一种是离散动作每部电梯从「上行、下行、停靠、空闲」四个动作里选一个具体停靠楼层由底层逻辑根据方向决定。第二种是离散选层每部电梯从所有楼层里选一个作为下一个目标楼层动作空间大小等于楼层数。我倾向用离散选层因为目的楼层预约调度的核心就是分配目标楼层让智能体直接选层比让它选方向再由规则推导楼层更直接。20 层建筑就是 20 个动作加上一个「待命」动作共 21 个。动作空间不大普通全连接网络就能处理。如果楼层超过 40 层动作空间会偏大可以改成两级决策先选方向再在方向内选楼层。或者用动作嵌入把楼层编号映射到低维向量网络输出每个楼层的打分。后者实现复杂一些但扩展性更好。2.4 奖励函数让电梯学会「等人」而不是「跑得快」奖励函数是电梯群控里最容易翻车的地方。如果只奖励运送乘客数量智能体会学会让电梯一直满速运行、拒绝停靠因为停靠浪费时间。如果只惩罚等待时间智能体会让所有电梯都往呼叫最多的楼层挤造成空跑和拥堵。我一般用组合奖励包含四项乘客平均等待时间变化量、乘客平均乘梯时间变化量、电梯空跑惩罚、任务分配均衡度。等待时间变化量是核心每步计算所有未响应呼叫的平均等待步数如果这一步比上一步降低给正奖励升高给负奖励。乘梯时间惩罚防止电梯绕远路接人。空跑惩罚针对没有载客也没有目标楼层的电梯移动。均衡度奖励鼓励多部电梯分担呼叫避免一部电梯被分配过多任务。def compute_reward(prev_wait, curr_wait, prev_ride, curr_ride, empty_moves, assignment_std, weights): 组合奖励计算 prev_wait/curr_wait: 上一步和当前步的平均等待时间 prev_ride/curr_ride: 上一步和当前步的平均乘梯时间 empty_moves: 本步空跑电梯数量 assignment_std: 各电梯已分配任务数的标准差 weights: 各项权重字典 reward 0.0 # 等待时间降低给正奖励升高给负奖励 reward weights[wait] * (prev_wait - curr_wait) # 乘梯时间同样 reward weights[ride] * (prev_ride - curr_ride) # 空跑惩罚 reward - weights[empty] * empty_moves # 任务分配越均衡标准差越小奖励越高 reward - weights[balance] * assignment_std return reward权重设置没有万能值需要根据建筑客流特征调。我常用的初始权重是 wait 1.0、ride 0.3、empty 0.5、balance 0.2。等待时间权重最大因为候梯体验是群控核心指标。空跑惩罚不能太大否则电梯会拒绝去远处接人导致部分楼层永远等不到电梯。均衡度权重最小它起微调作用太大反而会让电梯互相推诿任务。提示奖励尺度要统一。等待时间变化量通常在 0 到 1 之间空跑数量是整数如果直接相加空跑惩罚会主导梯度。建议把空跑数量也归一化比如除以电梯总数。3. 用 Python 搭一套可训练的多智能体电梯环境3.1 环境类的核心接口与仿真步进多智能体环境需要实现标准接口reset 返回初始观测step 接收所有智能体的动作返回下一观测、奖励、完成标志和信息字典。电梯仿真还要处理乘客生成、电梯运动、开关门、上下客这些物理逻辑。import gym from gym import spaces import numpy as np class ElevatorEnv(gym.Env): def __init__(self, num_elevators6, num_floors20, capacity10, arrival_rate0.05, max_steps1440): super().__init__() self.num_elevators num_elevators self.num_floors num_floors self.capacity capacity self.arrival_rate arrival_rate # 每步每层新乘客到达概率 self.max_steps max_steps self.current_step 0 # 每部电梯的观测维度 32动作维度 楼层数1 self.observation_space spaces.Box( low-1, high1, shape(num_elevators, 32), dtypenp.float32) self.action_space spaces.MultiDiscrete( [num_floors 1] * num_elevators) self.elevators [] self.hall_calls [] self.passengers [] def reset(self): self.current_step 0 self.elevators [Elevator(i, capacityself.capacity) for i in range(self.num_elevators)] self.hall_calls [] self.passengers [] return self._get_obs() def step(self, actions): # actions: 长度为 num_elevators 的数组每个值 0..num_floors self.current_step 1 self._generate_passengers() self._apply_actions(actions) self._move_elevators() self._handle_boarding() rewards self._compute_rewards() done self.current_step self.max_steps obs self._get_obs() info {avg_wait: self._avg_wait(), avg_ride: self._avg_ride()} return obs, rewards, done, inforeset 里初始化电梯和呼叫列表。step 的顺序很重要先生成新乘客再应用动作设置目标楼层然后移动电梯最后处理上下客。如果顺序颠倒会出现电梯已经移动但新乘客还没生成的逻辑漏洞。arrival_rate 控制客流强度0.05 表示每层每步有百分之五概率出现新乘客早高峰可以调到 0.15 甚至更高。3.2 电梯运动与开关门的仿真细节电梯运动不是瞬移需要按步移动。每步电梯最多移动一层到达目标楼层后开门停留若干步让乘客上下然后关门继续。这些细节影响等待时间和乘梯时间的计算精度。class Elevator: def __init__(self, eid, capacity10): self.id eid self.floor 1 self.direction 0 self.load 0 self.capacity capacity self.door_open False self.door_timer 0 self.target_floor None self.assigned_calls [] def set_target(self, floor): if floor self.capacity: # 动作值等于楼层数表示待命 self.target_floor None else: self.target_floor floor 1 # 动作 0 对应 1 层 def move(self): if self.door_open: self.door_timer - 1 if self.door_timer 0: self.door_open False return if self.target_floor is None: self.direction 0 return if self.floor self.target_floor: self.floor 1 self.direction 1 elif self.floor self.target_floor: self.floor - 1 self.direction -1 else: self.direction 0 self.door_open True self.door_timer 3 # 开门停留 3 步开门停留 3 步是经验值对应现实中开关门加乘客进出大约 6 到 9 秒。如果仿真步长是 2 秒3 步就是 6 秒。这个值太小会导致乘客来不及上下太大会让电梯运力下降。实际项目里可以根据载客数动态调整载客多时停留久一点。3.3 用 QMIX 还是 MAPPO训练框架选型多智能体强化学习框架常见的有 QMIX、MAPPO、MADDPG。电梯群控是离散动作、合作型任务、智能体数量中等QMIX 和 MAPPO 都合适。QMIX 属于值分解方法训练一个联合 Q 网络再分解成每个智能体的 Q 值。优点是样本效率高适合仿真环境采样慢的场景。缺点是只能处理离散动作且对奖励函数设计敏感。MAPPO 属于策略梯度方法每个智能体有独立策略网络训练时用集中 critic。优点是稳定、支持连续和离散动作缺点是样本需求大。我一般先用 QMIX 快速验证环境设计和奖励函数是否合理因为 QMIX 训练快一两个小时就能看出策略有没有学到东西。如果 QMIX 效果不理想再换 MAPPO 调。如果项目要求更高可以上 MAPPO 加 GAE 和裁剪但调参工作量会明显增加。# QMIX 训练循环核心片段 for episode in range(num_episodes): obs env.reset() episode_reward 0 while True: # 每个智能体根据自身观测选动作epsilon 贪心 actions [] for i in range(num_elevators): if np.random.rand() epsilon: actions.append(env.action_space.sample()[i]) else: q_values agent_q[i](obs[i]) actions.append(q_values.argmax().item()) next_obs, rewards, done, info env.step(actions) # 存入经验回放 buffer.push(obs, actions, rewards, next_obs, done) # 每 100 步从回放采样训练一次 if len(buffer) batch_size and step % 100 0: batch buffer.sample(batch_size) loss qmix_loss(batch, agent_q, mixer, target_mixer) optimizer.zero_grad() loss.backward() optimizer.step() obs next_obs episode_reward sum(rewards) if done: break # epsilon 衰减 epsilon max(0.05, epsilon * 0.995)epsilon 从 1.0 开始衰减到 0.05前期多探索后期多利用。衰减率 0.995 是每 episode 衰减如果 episode 很长可以改成每步衰减。经验回放容量建议至少十万条太小会导致样本相关性太强训练震荡。注意多智能体经验回放里每条经验要存所有智能体的观测和动作不能只存单个智能体的。因为 QMIX 的联合 Q 网络需要全局信息。4. 训练不收敛、电梯扎堆避坑与排查清单4.1 奖励震荡不收敛现象训练曲线剧烈波动平均等待时间时好时坏看不出下降趋势。原因奖励尺度不统一等待时间变化量和空跑惩罚数量级差异大梯度被大项主导。另外多智能体环境本身非平稳每个智能体策略更新都会改变其他智能体的最优策略。解决先把所有奖励项归一化到相近范围等待时间除以最大等待步数空跑数量除以电梯总数。然后降低学习率从 1e-3 降到 3e-4 或 1e-4。如果还震荡检查经验回放里是否混入了不同策略产生的旧数据可以缩短回放容量或提高采样频率。4.2 所有电梯往同一层跑现象训练一段时间后多部电梯同时响应同一个呼叫其他楼层呼叫长时间无人响应。原因观测里邻居信息不足或奖励缺少均衡项每个智能体只看到自己的局部信息不知道其他电梯已经响应了该呼叫。解决在观测里加入「该呼叫是否已被分配」的标志位让智能体知道哪些呼叫已经被其他电梯认领。同时在奖励里加均衡度惩罚各电梯任务数标准差越大惩罚越大。如果还不行可以在动作选择时加一层规则过滤已被分配的目标楼层不允许其他电梯重复选择。4.3 训练后期策略退化现象训练前期平均等待时间下降到某个点后突然反弹并持续恶化。原因epsilon 衰减太快探索不足策略陷入局部最优。或者学习率没有衰减后期梯度太大破坏了已学到的策略。解决把 epsilon 衰减率从 0.995 调到 0.999让探索持续更久。学习率加衰减每 1000 episode 乘以 0.9。另外可以加策略熵正则鼓励策略保持一定随机性。4.4 仿真速度太慢现象一个 episode 跑几分钟训练几万 episode 要几天。原因Python 循环逐部电梯逐层移动没有向量化。乘客生成和上下客逻辑用了大量 Python 对象操作。解决把电梯状态用 numpy 数组批量表示所有电梯同时移动用布尔掩码处理开关门和上下客。乘客生成用 numpy 随机数一次性生成整批。如果还慢可以用 numba 加速核心循环或者把环境用 C 写然后 Python 调用。我一般先用 numpy 向量化通常能提速五到十倍够用了。4.5 评估指标和训练奖励不一致现象训练奖励很高但实际评估时平均等待时间没改善。原因训练奖励是组合奖励包含空跑惩罚和均衡度而评估只看等待时间。智能体可能学会了刷空跑惩罚的负分比如让电梯少动但等待时间没降。解决评估时同时看多个指标平均等待时间、平均乘梯时间、最长等待时间、电梯运行总距离。如果训练奖励高但等待时间没降检查奖励权重是不是空跑惩罚太大导致智能体学会了「不动就不扣分」。把空跑惩罚调小或者改成只有空跑且没有待命任务时才扣分。5. 从训练到上线策略导出与在线调度的衔接技巧训练完的策略要落地到真实或半真实的调度系统中间还有一段工程距离。我一般会先把训练好的策略网络导出成 ONNX 或 TorchScript然后用一个轻量推理服务加载每步接收电梯状态和呼叫状态输出每部电梯的目标楼层。推理延迟要控制在 50 毫秒以内否则会影响调度实时性。导出时注意观测预处理要和训练时完全一致。归一化参数、特征顺序、缺失值填充方式任何一处不一致都会导致策略输出异常。我习惯把预处理逻辑也一起导出或者用同一个 Python 函数封装训练和推理共用。在线调度还要加一层安全兜底。如果策略输出的目标楼层和当前电梯方向冲突比如电梯正在上行但目标楼层在下方不要直接反转方向而是先完成当前方向上的已分配任务再响应。另外设置最大等待时间阈值任何呼叫等待超过阈值时强制分配最近的空闲电梯不依赖策略输出。这层规则兜底能防止策略在极端场景下做出反直觉决策。验证策略效果时不要只看平均等待时间。我一般会跑三组对比传统先来先服务、最近电梯优先、训练策略。每组跑至少一百个 episode统计平均等待时间、百分之九十等待时间、最长等待时间。如果训练策略只在平均等待时间上略好但长尾等待时间差很多说明策略对极端情况处理不好需要调整奖励里的长等待惩罚。最后说一个我踩过的坑训练环境里的乘客到达分布和真实场景不一致。仿真里用均匀分布生成乘客真实早高峰是集中爆发。策略在仿真里表现很好上线后早高峰直接崩。后来我在训练时把到达率做成动态的早高峰时段到达率是平峰的十倍策略才学会应对突发客流。如果你要做落地训练环境的客流模型一定要尽量贴近真实否则训练再久也是白搭。希望帮到你。本文还有配套的精品资源点击获取