简介面向RoboCup Rescue仿真救援竞赛的Java代码包适合参赛队员、机器人方向学生及希望了解多智能体决策与救援仿真的开发者用于搭建可重复的灾害救援实验环境研究搜索、路径规划、目标识别与避障等核心流程。包内共43个文件以42个Java源码为主另含1个Java备份文件整体约74KB体量轻便便于按模块阅读和二次修改。已有1579人学习下载说明该方向具有一定关注度。代码围绕仿真环境、算法实现、传感器数据处理、控制执行、日志评估与参数配置等环节组织可帮助读者梳理救援智能体的决策链路观察策略如何从感知输入落到动作输出并借鉴日志与评估思路排查问题、对比算法效果适合作为课程设计、竞赛预研和AI机器人入门实践的参考材料。1. 仿真救援不是打怪这套代码解决的是灾难前 30 分钟的资源调度赛前一晚我在一台机器上同时拉起地图服务器、仿真内核和六个智能体进程屏幕上火光一片消防车却在地图西北角来回打转。它明明收到了起火建筑的坐标就是不往前走。查了四十分钟才发现那条街道被倒塌建筑堵死而我的路径规划还在拿灾前路网图算最短路。这套仿真救援代码要解决的就是这类问题。它把一座城市丢进一个离散时间步的仿真内核里让消防队、救护队、警察队三类智能体在有限的视野、有限的载重、有限的时间步里抢分灭火、救人、清障。适合两类人一类是想跑通多智能体协同的实验环境另一类是准备打仿真救援类竞赛、需要一份能直接改的决策骨架。它的价值不在于模型多复杂而在于整条链路是通的——感知、建模、决策、动作每一步都能落到代码里。2. 项目定位与技术栈2D 仿真救援到底在模拟什么先把这个项目的边界划清楚。它不是一个渲染精美的游戏也不是一个纯强化学习训练环境而是一套「协议客户端 世界模型 规则决策器」的单机多进程系统。内核负责物理与规则你的代码只负责一件事在每一次收到感知数据后在限定时间内返回一批动作。技术栈相当朴素Python 3.8 负责任务逻辑标准库的socket负责和内核通信heapq实现寻路json做配置。没有 Web 框架没有深度学习依赖甚至不需要 GPU。这种朴素的另一个好处是冷启动快在本机上同时开几十个智能体进程也不会打满内存。2.1 三类智能体与世界的拆解仿真世界里能被感知和操作的实体是有限的几类把它们记熟比读十遍代码管用实体关键字段谁的职责坑点Building燃烧度、破损度、温度、地面面积、坐标消防队灭火、救护队判断掩埋燃烧度不是 0/1是分级的0 到 8 表示从完好到烧塌Civilian血量、伤害值、坐标救护队装载与后送伤害值随时间增长救晚了直接判死Road头节点、尾节点、是否被阻塞警察队清障阻塞是动态的路网不是静态图Blockade剩余清障量、占据的单元警察队处理一个路障可能压着平民优先级要重排Refuge坐标消防队补水、救护队卸人数量稀少去晚了排队Platoon自身位置、水量、载人数、状态全部这是你的智能体本体三类智能体的能力边界不能混淆消防队只能对建筑喷水水量耗尽必须回补给点救护队一次只能载有限人数装卸都要消耗时间步警察队清障是渐进式的一次动作只能削掉一部分清障量。我见过有人让消防队去搬平民结果整个时间步都被内核拒绝白白浪费一轮决策。提示感知数据里每个实体都带唯一 ID但 ID 只在单局内稳定别把它写进配置或缓存跨局复用。规则层面还有一条必须记住的硬约束当被烧毁建筑的地面总面积超过阈值仿真会提前结束你的智能体还没来得及动作就被判负。这条规则直接决定了决策的优先级排序——压火势永远优先于救单个平民。2.2 目录结构代码分成哪几层这套代码的分层思路很直接协议、模型、决策、动作四层隔一层就不许跨层调用rescue-agent/ ├── agents/ # 三类智能体的入口进程 │ ├── fire_brigade.py │ ├── ambulance_team.py │ └── police_force.py ├── core/ │ ├── client.py # 协议层连接、订阅、收发一帧 │ ├── world_model.py # 世界模型把感知流解析成对象图 │ ├── planner.py # 路网构建与寻路 │ └── actions.py # 动作封装统一格式下发 ├── config/ │ ├── team.json # 队伍配置各类型数量、端口段 │ └── weights.json # 决策权重调参只动这个文件 ├── scripts/ │ ├── start_kernel.sh # 拉起内核与地图 │ └── run_team.sh # 批量拉起整支队伍 └── logs/ # 每局一份记录每步决策core/client.py是唯一碰 socket 的地方其它模块全部面向对象操作这样换平台或者换协议版本时只改一个文件。core/world_model.py负责把内核下发的扁平数据转成带索引的字典是整份代码里被调用最频繁的模块。config/weights.json单独抽出来的原因很实际调参的时候不想重启进程把权重外置后改完文件下一局直接生效。特别说一下agents/下的三个入口。它们不是三个独立程序而是同一个控制循环套了不同的策略类通过命令行参数指定智能体 ID 和端口。这样做的代价是入口稍微多几个文件收益是三类智能体的行为差异全部集中在策略层不会散落在启动逻辑里。3. 跑起来从零编译到第一帧感知数据的完整链路跑不通这套代码的人九成卡在启动顺序上。内核、地图、智能体三者有严格的先后依赖顺序错了要么连不上要么连上了收不到数据而且两种错误的表现几乎一样——进程活着屏幕没输出。3.1 环境准备与启动顺序先确认三件事Python 版本不低于 3.8本机防火墙允许内核端口段被访问以及地图和场景配置文件是配套的版本不匹配时内核会加载失败但不一定打印清晰日志。# 1. 安装依赖只有一个第三方包用于日志着色其余全是标准库 pip install -r requirements.txt # 2. 先起内核-c 指定场景配置-t 指定最大时间步 ./scripts/start_kernel.sh --config config/scenario.json --steps 300 # 3. 等内核打印 kernel ready 之后再起地图服务 ./scripts/start_kernel.sh --map-only # 4. 最后批量拉起整支队伍序号必须和 team.json 里的定义一致 python -m scripts.run_team --team config/team.json --host 127.0.0.1 --base-port 7000这几个命令背后有个容易忽略的细节--base-port 7000是起始端口内核会给每个智能体分配一个独立端口7000、7001、7002 依次往上。如果你的智能体数量超过 20 个记得检查端口段有没有和系统其它服务撞上撞上时的表现是连接被拒绝但报错信息很难看出是端口冲突。--steps 300是单局的最大时间步。调这个值会直接影响调试节奏跑通阶段建议设成 50二十秒就能出一局结果调参阶段再放到 300 以上否则统计样本太小指标噪声盖过真实差异。启动后正确的日志长这样内核打印每步的实体统计智能体打印「已连接、已订阅、收到第 N 帧」。如果智能体一直停在「已连接」说明订阅没生效如果连「已连接」都没有先查端口。3.2 连接握手与第一帧感知数据协议层是最容易被跳过、也最容易翻车的一层。仿真救援的通信是长连接 换行分隔帧客户端连上后要先报身份收到确认才能订阅订阅之后内核才会在每个时间步推送感知。少一步都不行。import socket class AgentClient: 协议客户端只负责连接、订阅和按帧收发不做任何业务判断 def __init__(self, host, port, agent_id): self.host host self.port port self.agent_id agent_id self.sock None self.buf # 拼包缓冲TCP 是流式的一帧可能分两次到 def connect(self, timeout10): self.sock socket.create_connection((self.host, self.port), timeouttimeout) # 握手第一步报上自己的全局 ID内核据此绑定端口和实体 self._send_line(f(AK_Connect {self.agent_id})) reply self._read_frame() if OK not in reply: raise RuntimeError(fconnect rejected: {reply}) def subscribe(self, channels): # 订阅决定每帧下发哪些实体订得越多单帧越大解析越慢 self._send_line(f(AK_Subscribe { .join(channels)})) def recv_frame(self): return self._read_frame() def _send_line(self, line): self.sock.sendall((line \n).encode(utf-8)) def _read_frame(self): # 标准做法是按换行切帧且必须保留残余缓冲 while \n not in self.buf: chunk self.sock.recv(65536) if not chunk: raise ConnectionError(kernel closed the connection) self.buf chunk.decode(utf-8, errorsreplace) frame, self.buf self.buf.split(\n, 1) return frame逐段说清楚。agent_id是内核分配的全局编号同一个端口上重复握手会被拒绝所以调试时不要用两个进程抢同一个 ID。subscribe的channels参数我一般传[building, civilian, road, refuge, self]早期调试可以只订self和building帧体积能小一个数量级方便先看清结构再加。_read_frame里的buf是血泪经验不保留残余缓冲在地图实体多、单帧超过 64KB 时必然出现半包表现是 JSON 解析偶尔报错且只在大型地图上复现非常玄学。用errorsreplace而不是直接decode()是为了内核偶尔下发非 UTF-8 字节时不会让整个进程崩掉至少能打日志看到原始内容。提示create_connection的超时参数只作用于连接阶段读数据阶段没有超时保护。生产一点的写法是给 socket 设settimeout否则内核卡住时你的进程会无限等待。握手成功后收到的第一帧建议原样打到日志里看一遍。这一帧的结构决定了后面所有解析代码的形状比任何文档都准。3.3 跑通之后先看这三个指标进程能跑不等于代码对了。第一次跑通先看三个数能省掉后面大量排查时间第一个是每帧的实体数量波动。世界模型解析出来的建筑数、平民数应该随时间单调下降烧毁、死亡如果忽然变成 0多半是解析器在某个字段上抛了异常被吞掉。第二个是智能体的动作响应率即内核接受的动作数除以你发出的动作数低于 90% 说明动作格式或前置条件有问题。第三个是单帧处理耗时用time.time()掐一下稳定在 100ms 以内才算健康超过 500ms 就要考虑优化解析和寻路了。这三个指标建议做成一行日志每 10 帧打一次。后面调参时全靠它定位问题。4. 核心模块拆解一个消防智能体从感知到动作的完整链路链路本身很短收帧、解析成世界模型、挑目标、算路径、下发动作。难的是每一环都有边界条件。下面按执行顺序拆。4.1 世界模型把感知流变成可查询的对象图内核下发的数据是扁平的你需要的查询是「离我最近且还在烧的建筑」「哪些平民压在路障下面」。这两者之间差一层索引结构world_model.py就是干这个的。class WorldModel: def __init__(self): self.step 0 self.buildings {} # id - {fieryness, brokenness, area, pos} self.civilians {} # id - {hp, damage, pos, buried} self.blockades {} # id - {repaire, pos, cells} self.roads {} # id - {head, tail, blocked} self.refuges {} self.me {} # 自身状态pos / water / loaded / hp def update(self, frame): 每帧全量重建简单但可靠实体规模大时改成增量更新 self.step 1 self.buildings.clear() self.civilians.clear() for item in frame.get(buildings, []): self.buildings[item[id]] { fieryness: item[fieryness], brokenness: item[brokenness], area: item.get(ground_area, 1.0), pos: item[pos], } for item in frame.get(civilians, []): self.civilians[item[id]] { hp: item[hp], damage: item[damage], pos: item[pos], # 被压在建筑物下时内核给的标记救人前必须先清障 buried: item.get(buried, False), } self.me frame.get(self, {})update用全量重建而不是增量更新是刻意的选择。增量更新省 CPU但要处理实体消失建筑烧塌、平民死亡的逻辑一旦漏了就会拿到过期对象排查成本远高于省下的那点时间。等单帧耗时真的超过预算了再改增量那时候你已经有日志能对比了。字段里最容易误读的是fieryness。它是分级值不是布尔。0 表示没火中间值表示燃烧程度最高一档表示已经烧塌。判断「要不要去灭」时直接拿它当强度用不要写成if fieryness 0那样会把一处小火和半条街着火同等对待。area同理它是地面面积大火烧大楼和小火烧平房对总分的破坏完全不同。4.2 决策层目标评分与全源最短路目标选择本质是个打分问题火势带来的紧迫性减去过去的路程成本。这个公式简单但足够干活而且每一项目标都能单独调。def score_building(me, bld, dist, w): 返回该建筑的优先级得分越大越该去 # 火势强度 × 面积烧得越狠、面积越大的建筑越优先 urgency w[fire] * bld[fieryness] * bld[area] # 破损度高的建筑救回来价值低稍微降权 value 1.0 - w[broken] * bld[brokenness] # 距离惩罚防止为了灭一处远火跑半个地图 cost w[dist] * dist return urgency * value - cost三个权重的量纲不一样调参的时候不要一起动。我的习惯是先固定dist把fire和broken调出一个合理的取舍关系再调dist控制活动的半径范围。weights.json里存的就是这三项改完下一局生效不用重启。距离从哪来不能每帧现算最短路那是 O(n²) 的开销。常见做法是开局时按路网跑一次全源最短路把结果缓存成二维表之后每帧只做查表。import heapq def dijkstra(adj, src): 单源最短路返回 {node_id: 距离}开局预计算一次之后只查表 dist {src: 0.0} pq [(0.0, src)] while pq: d, u heapq.heappop(pq) if d dist.get(u, float(inf)): continue for v, w in adj.get(u, []): nd d w if nd dist.get(v, float(inf)): dist[v] nd heapq.heappush(pq, (nd, v)) return distadj是邻接表键是路网节点 ID值是一组(邻居节点, 权重)。权重取两个节点之间的欧氏距离或道路长度都行前者够用后者更准但要多解析一个字段。dist缓存后查询建筑距离时要先把建筑坐标映射到最近的路网节点这一步不能省——直接拿坐标算直线距离会让消防车往墙上撞。路障来了怎么办路网会变。我的做法是维护两张表一张静态最短距离表开局算好一张动态阻塞标记每帧更新查询时如果最短路上有阻塞退化成带黑名单的 A* 重算只在真正需要时才付这个开销。4.3 动作下发与状态机闭环决策出目标之后动作不是一次性发一条而是要有状态。消防车在「去目标」「喷水」「回补给点」三个状态之间切换中间被打断要能恢复。RESERVE_WATER 500 # 保留水量低于这个值就回补给点 SPRAY_MIN_WATER 200 # 低于这个值连一次喷水都放不出来 class FireBrigadeController: def act(self, wm, planner): me wm.me # 状态 1水量见底放弃灭火先回补给点 if me[water] RESERVE_WATER: refuge planner.nearest_refuge(me[pos]) return [(AK_Move, refuge)] if refuge else [] # 状态 2没有目标或目标已不燃烧重新选一个 target me.get(target) if target is None or target not in wm.buildings \ or wm.buildings[target][fieryness] 0: target planner.pick_fire_target(me, wm) me[target] target if target is None: return [] # 没火可灭站着不动比乱跑安全 # 状态 3已经贴到目标建筑开始喷水 if planner.adjacent(me[pos], wm.buildings[target][pos]): if me[water] SPRAY_MIN_WATER: return [] return [(AK_Extinguish, target)] # 状态 4还在路上继续走 return [(AK_Move, planner.next_hop(me[pos], target))]几个参数值得单独说。RESERVE_WATER代表的是保命线取值要大于SPRAY_MIN_WATER加上返程耗水否则会出现「跑到目标发现喷不动也回不去」的死局这在早期版本里是真实发生过的翻车。SPRAY_MIN_WATER是内核的硬门槛低于它发动作会被静默拒绝注意是静默——不报错只是动作不生效所以必须自己在客户端侧挡住。状态 2 里那句「目标已不燃烧就重选」看着像废话但没有它会出现一种情况目标被队友先灭了消防车还在原地继续喷空气一个时间步就这么浪费了。仿真救援的每一步都是钱这种空转累积起来非常致命。next_hop返回的是路径上的下一个节点而不是终点因为AK_Move一次只能走一格。想一次走远就得自己维护一个路径队列每步弹一个出来。5. 避坑与排查仿真救援里最容易翻车的五个问题下面五条都是实际跑出来的不是文档里抄的。每条按现象、原因、解决写。5.1 连接与协议类问题现象一智能体连上后一直收不到感知数据进程活着但没有输出。原因是握手后没有订阅或者订阅的通道名写错了。内核不会因为通道名错误而报错只是什么都不推。解决方式是把订阅动作和接收动作之间加一行日志打印订阅请求的原文同时在内核侧确认该 ID 已进入订阅列表。更省事的办法是订阅后设一个 2 秒超时超时就打印警告并重发一次订阅。现象二跑半小时后偶发 JSON 解析失败重启就好大型地图必现。这就是半包问题。TCP 是字节流一帧可能分多次到达只按recv的边界切帧必然出问题。解决办法是保留接收缓冲按分隔符切切剩下的留到下一次。顺带一提缓冲区用字符串累加在内核推送频繁时会有性能问题实体量大时换成bytearray拼接开销能降一个数量级。5.2 决策与性能类问题现象三所有消防车开局全部涌向同一栋建筑其它方向的大火没人管。原因是目标选择是各智能体独立计算的同一个评分函数必然给出同一个答案。解决靠两点一是给已分配目标加惩罚项被别的队友盯上的建筑在评分时扣掉一个固定值二是在队伍层维护一个共享的分配表由队长进程周期性下发。前者改动小后者效果好先上前者验证思路。现象四单帧处理耗时从 80ms 涨到 1.2 秒动作开始落后于时间步。八成是全源最短路算太勤了。检查一下是不是每帧都在调dijkstra。路网在开局和每次清障后才会变化其余时间应该只查表。另一个常见原因是每帧都在遍历所有平民算最近目标改成按建筑分组、按距离剪枝后耗时立刻下来。判断标准很简单把每帧的处理拆成解析、建模、决策三段分别计时哪段超一目了然。5.3 崩溃判定与提交类问题现象五明明灭了不少火总分却很低甚至比不操作还差。先看崩溃判定条件。总分里建筑损毁的权重通常远大于单个平民压火势的优先级如果排在救人后面就会出现「救了一堆人但半座城烧没了」的结果。解决办法是在权重表里把建筑损毁项单独提出来和救人项做一次消融对比看哪一组在多个地图上更稳定。我的经验是在崩溃线附近必须优先压火远离崩溃线时再转向救人做一个基于损毁面积占比的动态切换。再补一个提交环节的坑日志里记录的决策和你最终提交的动作之间必须能对上。我吃过一次亏日志打的是「选中目标 X」实际下发的是上一帧缓存的 Y因为状态对象被跨帧复用了。后来改成每帧下发前把最终动作原文打一条再没出现过。提示logs/目录按「地图名 随机种子 时间戳」命名别只用时间戳。调参时要复现某局地图和种子缺一个就白搭。6. 二次开发把单智能体改成可协同的编队单智能体跑通之后下一步几乎所有人都会想加协同。最省事的切入点不是全局规划器而是共享目标分配表因为改动局限在决策层不碰协议和建模。6.1 用一张共享表切断重复决策在core/下加一个轻量的分配器由队伍里的一个进程兼任每 N 帧广播一次「谁负责哪个目标」。各个消防智能体在选择目标前先查这张表被占用的建筑直接跳过。改造点位置改动量见效速度目标占用表core/planner.py约 40 行立刻重复率明显下降距离动态权重config/weights.json只改数值下一局状态机加「增援」分支agents/fire_brigade.py约 30 行大火场景明显这张表有两个必须处理的细节一是过期占用它的智能体死了、没水了、改目标了表项要能超时自动释放否则会永久锁死建筑二是争抢两个智能体在同一帧都想认领时用 ID 小的优先比加锁简单得多。6.2 验证改动是否真的有效协同改完之后不要凭感觉说「好像好了」。固定地图和随机种子跑十局取均值对比改动前后的三个数建筑总损毁面积、成功后送的平民数、每帧平均耗时。前两个应该上升第三个不能显著变差。我一般会在config/team.json里留一个baseline_mode开关关掉就走原来的独立决策这样 A/B 对比只要改一个布尔值。最后提醒一条每次调完权重先跑 50 步的短局看趋势趋势对了再跑满 300 步。这个习惯帮我省下过无数次等待——有一次我信心满满地把距离权重调低短局里火势压得漂亮满局跑到第 220 步崩了原因是消防车全跑去了地图另一头补给点附近烧成白地。从那以后我每次改决策逻辑都强制先跑短局再跑满局两轮数据对不上就不提交。希望帮到你。本文还有配套的精品资源点击获取
设计与优化实战指南)
