swarm 2.1 群体仿真：调度、观察者与可复现实验

简介：Swarm 2.1 仿真工具面向多智能体系统与复杂系统建模的研究者、计算机专业学生及分布式算法开发者，提供可搭建、运行并观察智能体交互过程的仿真环境，适用于人工智能、分布式计算、网络优化与生物启发式算法等场景。压缩包共 221 个文件，以 73 个 .m 实现文件、58 个 .h 头文件及 27 个 .o 目标文件为主体，并含 changelog、makefile、readme、scm 脚本和少量 exe 示例，整体约 691KB，目录结构便于按源码组织理解编译流程。已有 205 人学习。借助智能体行为定义、Boids 群飞模型、FIPA-ACL 通信协议示例与可视化观察能力，读者可复现群体行为、验证调度与协作策略，并借助评估调试工具定位版本差异和运行问题，也可将仿真用于交通管理、社交网络分析、生态建模等课题的建模验证，适合作为多智能体仿真入门与二次开发的参考。

1. 一群人各自按简单规则行动，整体却跑出秩序：swarm 2.1 仿真工具替我省下的四件事

第一次把一群个体的行为规则写完，我以为剩下的就是按回车。真跑起来才发现，卡住人的从来不是规则本身，而是驱动：谁先动、谁后动、动几次、动完的状态从哪一版读、跑出来的现象怎么记下来、换个种子还对不对得上。swarm 2.1 仿真工具解决的正是这一层——它把多智能体群体仿真里最枯燥也最容易写错的四件事做成了固定骨架：调度推进、空间邻域、观察者采样、运行时探针。你只需要往模型层塞规则，剩下的骨架不用每次重写。

它适合做群体行为验证、集合行为复现、规则参数扫描的人，也适合带学生做课程设计、需要一套能讲清楚"个体规则怎么变成整体现象"的现成框架。不适合只想跑个动画看热闹的——这套东西的价值在可复现的数值，不在画面。

2. swarm 2.1 的骨架：调度、空间、观察者三层怎么咬合

很多人解压完打开目录会愣一下：没有想象中的main，只有一堆听起来像名词的类。原因很简单，群体仿真天然是三层的，主函数被拆进了骨架里。模型层描述"一个个体现在是什么状态、下一步想干什么"；调度层回答"这一拍轮到谁、按什么顺序动"；观察层负责"把此刻的世界读出来记下来，但绝不改它"。三层混在一起写，代码在两百个个体以内都能跑，一旦上到几千个就开始出现那种改一行、结果全变、还找不到原因的玄学现象。

先把三层的职责边界摆清楚，后面所有参数才有地方落。

2.1 模型层、调度层、观察层各自的边界在哪

模型层最容易写胖。合格的做法是让个体的一次更新只包含三件事：感知（拿到邻域信息）、决策（算出期望方向或状态转移）、执行（写回自己的位置和速度）。凡是"顺手统计一下当前有多少个簇"这种活，一律挪到观察层，按固定周期跑。这不是洁癖，是因为统计一旦混进更新循环，它就和个体更新共享了同一份随机数流和同一套遍历顺序，你想单独关掉统计做对照实验都做不到。

调度层的边界是"只管什么时候调用，不管调用里干什么"。它可以有全局节拍，也可以挂子调度：移动每拍一次、状态机每五拍一次、环境更新每二十拍一次。把这些周期写成配置而不是散落在代码里的if t % 5 == 0，是这套骨架最省事的地方——调参数的时候你改一处，全实验一致。

观察层的边界最硬：只读。快照、指标、探针，全部只允许读对象状态、写自己的缓冲区。运行时要看某个个体的内部变量，也走探针接口，而不是在更新函数里print。

2.2 用一个最小闭环跑出第一帧

下面这段是照着 swarm 的抽象复刻的最小骨架，四十多行，跑起来能出第一帧快照。用 Python 写是为了让你五分钟内看到闭环，抽象本身和工具里的是一一对应的。

# 最小 swarm 骨架：三层各自独立，任何一层都不越界 import random, math from dataclasses import dataclass @dataclass class Agent: x: float; y: float vx: float = 0.0; vy: float = 0.0 def step(self, world): # 模型层：只做感知-决策-执行，不写任何统计 ns = world.neighbors(self, world.radius) if not ns: return ax = sum(a.vx for a in ns) / len(ns) # 对齐 ay = sum(a.vy for a in ns) / len(ns) self.vx = 0.9 * self.vx + 0.1 * ax # 惯性系数，见参数表 self.vy = 0.9 * self.vy + 0.1 * ay self.x = (self.x + self.vx) % world.size self.y = (self.y + self.vy) % world.size class World: def __init__(self, n, size, radius, seed): rnd = random.Random(seed) # 独立随机源，别用全局 random self.size, self.radius = size, radius self.agents = [Agent(rnd.uniform(0, size), rnd.uniform(0, size), rnd.uniform(-1, 1), rnd.uniform(-1, 1)) for _ in range(n)] def neighbors(self, me, r): # 空间层：邻域查询集中在这一个函数里，将来换网格只改这里 return [a for a in self.agents if a is not me and (a.x-me.x)**2 + (a.y-me.y)**2 < r*r] class Schedule: def __init__(self): self.jobs = [] def at(self, period, offset, fn): # 周期 + 相位偏移 self.jobs.append((period, offset, fn)) def run(self, world, steps): for t in range(steps): for period, offset, fn in self.jobs: if (t - offset) % period == 0: fn(world, t) class Observer: def __init__(self): self.frames = [] def sample(self, world, t): # 观察层：只读，产出可落盘的元组 if t % 10 == 0: self.frames.append((t, [(a.x, a.y) for a in world.agents])) w, s, o = World(300, 100.0, 6.0, 42), Schedule(), Observer() s.at(1, 0, lambda wd, t: [a.step(wd) for a in wd.agents]) # 移动：每拍 s.at(10, 0, o.sample) # 采样：每 10 拍 s.run(w, 200) print(len(o.frames), len(o.frames[0][1])) # 20 帧，300 个点

代码之后要说的不是"它能跑"，而是每一处在替你做决定。Schedule.at的第二个参数是相位偏移，它决定了不同动作是不是在同一拍上撞车——把采样偏移设成 0 和设成 5，在步长较大时能跑出肉眼可见不同的曲线。World.neighbors被单独摘出来，是因为它是后面所有性能问题的唯一入口，第 5 章那个"人数一多就慢十倍"的坑就出在这里。Observer.sample里那个取模判断写死了 10，真实实验里应该是配置项。

几个参数的初始取值和调整方向：

平均邻居数大概等于π·radius²·n / size²，起手那组参数算下来是 3.4 个，属于能形成局部对齐又不会一步全同步的区间。这个估算比反复试参数快得多，先按它定范围，再在一个窄区间里扫。

2.3 调度粒度：一次节拍上到底挂什么

调度是最容易被写废的一层。常见的翻车写法是把所有动作塞进一个每拍执行的循环里，然后在循环体内用一堆条件判断决定"这一拍要不要做状态机更新"。等到你想把状态机频率从 5 改成 7 做敏感性分析，就得满代码找魔数。

用调度对象挂任务的方式，频率和相位都是数据，可以整表配置：

相位偏移这一列看着不起眼，实际影响很大。如果移动和采样都偏移 0，那么采样永远发生在"所有个体都动完"之后，你看到的永远是每拍的终态；如果采样偏移 5、移动周期是 10，那采到的是半拍状态。做对比实验时必须把这张表连同结果一起存下来，否则半年后你根本说不清两条曲线的差异是来自参数还是来自采样相位。

激活方式那一栏的"顺序 vs 随机"也值得单独说。顺序激活下，排在前面的个体永远先读到旧世界，排在后面的个体永远读到"前面已经动过"的混合状态；随机激活每拍打乱一次顺序，能消掉这种系统性偏差，代价是随机数流被额外消耗，复现时顺序也必须一起固定。我的习惯是：做机理验证用随机激活，做性能基准用顺序激活，两者不混着比。

3. 从规则到参数：感知半径、步长与更新顺序怎么定

规则写得再漂亮，落到仿真里都是几个数字。这一章讲的是怎么把"个体应该跟着邻居走"这种描述，翻译成一组能扫、能比、能复现的参数，以及其中哪几个数字一改，结论就完全反过来。

3.1 个体规则的三要素：感知、决策、执行

任何群体规则拆开都是三段。感知段决定"我看到谁"，决策段决定"我倾向于往哪走"，执行段决定"我实际挪多远"。三段里最容易含糊的是执行段——很多人把决策算出来的期望向量直接当速度用，结果个体移动步长随邻居数量变化，人一多就飞出去了。合格的做法是给执行段一个上限：期望向量先归一化，再乘以固定的最大步长。

def step(self, world): ns = world.neighbors(self, world.radius) # 感知 if not ns: return sep = [0.0, 0.0]; ali = [0.0, 0.0]; coh = [0.0, 0.0] for a in ns: dx, dy = a.x - self.x, a.y - self.y d2 = dx*dx + dy*dy if d2 < world.sep_r ** 2: # 分离：太近就推开 sep[0] -= dx / (d2 + 1e-9); sep[1] -= dy / (d2 + 1e-9) ali[0] += a.vx; ali[1] += a.vy # 对齐 coh[0] += dx; coh[1] += dy # 聚集 k = len(ns) # 决策：三个分量加权求和 ax = world.w_sep*sep[0] + world.w_ali*ali[0]/k + world.w_coh*coh[0]/k ay = world.w_sep*sep[1] + world.w_ali*ali[1]/k + world.w_coh*coh[1]/k # 执行：先归一化再乘最大步长，保证速度上限恒定 norm = math.hypot(ax, ay) or 1.0 self.vx = 0.95*self.vx + 0.05*(ax/norm)*world.max_speed self.vy = 0.95*self.vy + 0.05*(ay/norm)*world.max_speed self.x = (self.x + self.vx) % world.size self.y = (self.y + self.vy) % world.size

三个权重的量纲完全不同：分离项被1/d²放大，聚集项是绝对距离，对齐项是速度。不归一化直接相加，实际起作用的基本只有分离项，另外两项等于没写。max_speed是硬上限，0.05是响应系数——它决定个体多快跟上期望方向，取值太大会振荡，太小则整群像在糖浆里游。

max_speed超过感知半径的一半会出问题的原因很直接：个体一步跨过整个感知范围，邻域信息永远滞后一拍，群体看起来在"抽搐"。这条经验值比任何调参技巧都实用。

3.2 同步更新与异步更新：同一条规则，两种结果

这是最容易出结论错误的地方。所谓同步更新，是所有个体都基于同一拍的旧状态算出新状态，然后一次性写回；异步更新是遍历到谁就当场改，后面的个体读到的是已经变过的世界。

# 同步：双缓冲，读旧写新，结果与遍历顺序完全无关 def sync_step(world): new = [] for a in world.agents: na = Agent(a.x, a.y, a.vx, a.vy) na.step_from(world) # 内部仍然读 world.agents 的旧值 new.append(na) world.agents = new # 整批换血，天然可复现 # 异步：就地更新，速度快一倍，但结果依赖遍历顺序 def async_step(world): for a in world.agents: # 顺序一旦改变，结果就变 a.step(world)

同步版本内存占用翻倍，但有一个不可替代的好处：结果和遍历顺序解耦。你换一个 Python 版本、换一台机器、把列表换成字典再换回来，结果都一样。异步版本快，也"更像真实世界"，但它在个体数多、耦合强的时候会引入系统性偏差——排在前面的个体影响被放大，排在后面的被削弱。我做过一次对比，同一组参数下两种更新方式的簇数量能差出 30% 以上，而参数表里根本看不出这个差异来自实现方式。

所以做对照实验之前先定死一件事：更新方式是实验设计的一部分，必须写进配置、跟着结果一起存。中途换更新方式再比参数，等于换了个模型在比。

3.3 参数扫描：把一次跑变成一组跑

单点实验的结论不可信，这是血泪经验。至少要扫三个轴：密度（n / size²）、感知半径、响应系数。扫的时候手写循环很容易把参数组合和输出文件对不上号，我的做法是让配置先落盘、再跑、文件名带哈希。

import itertools, json, hashlib, subprocess base = {"size": 100.0, "max_speed": 0.5, "w_sep": 1.5, "w_coh": 0.6, "steps": 600, "sync": True, "seed": 42} def run_key(cfg): s = json.dumps(cfg, sort_keys=True) return hashlib.md5(s.encode()).hexdigest()[:8] # 配置指纹 for n, radius, resp in itertools.product([200, 400, 800], [4.0, 6.0, 8.0], [0.03, 0.05, 0.1]): cfg = {**base, "n": n, "radius": radius, "resp": resp} key = run_key(cfg) with open(f"runs/{key}.json", "w") as f: json.dump(cfg, f) # 先存配置，再跑 subprocess.run(["python", "run_one.py", "--config", f"runs/{key}.json"])

配置指纹这一步不是形式主义。参数一多，n400_r6.0_resp0.05这种文件名迟早会撞车或者写错，而指纹是配置内容的函数，改任何一个字段都会变。跑完几百组之后，你只需要拿指纹去反查，就能百分之百确认那条异常曲线是哪组参数跑出来的。这套做法在任何仿真工具里都通用，不依赖具体实现。

4. 观察者与序参量：把"看起来像一群"量化成可比的数

图好看不等于结论成立。群体仿真最尴尬的场面是：两张截图摆在面前，一张紧密、一张松散，但你没法说清差异有多大、是不是随机波动、换种子还成不成立。观察层存在的意义就是把这种"看起来"变成能算均值、能算方差、能放进表格的数。

4.1 三个便宜好用的序参量

序参量不用多，三个够用：极化度衡量方向一致性，最近邻距离衡量聚集紧密程度，簇数量衡量结构分裂情况。三个都是 O(N) 或 O(N log N) 能算出来的，采样频率再高也扛得住。

def polarization(agents): sx = sum(a.vx / (math.hypot(a.vx, a.vy) or 1.0) for a in agents) sy = sum(a.vy / (math.hypot(a.vx, a.vy) or 1.0) for a in agents) return math.hypot(sx, sy) / len(agents) # 0~1，越大越同步 def mean_nn_distance(agents, size): # 简化版：网格分桶后只在邻近格子内找最近邻，避免 O(N^2) ...

极化度这个指标特别适合判断"参数是不是调过头了"。当响应系数偏大时，极化度会先升后剧烈摆动，均值看着还行，方差却爆掉——只看均值你完全发现不了这个问题，必须把方差一起记下来。这也是为什么采样要每 N 拍一次持续记，而不是只存最后一帧。

4.2 探针与快照：运行时读取和回放

探针是这套工具里最容易被误用的部件。它的本意是让实验者在运行中查看甚至临时修改某个对象的变量，用来做交互式调试。问题在于，很多人会把探针当采样器用，每拍都读一遍，读的时候顺手还调用了对象的某些方法，而这些方法内部消耗了随机数。结果就是：开着探针跑和关着探针跑，曲线不一样。

规矩只有一条：探针读到的值必须复制出来，绝不回写，也绝不调用任何有副作用的方法。

class Probe: def __init__(self, period=10): self.period = period self.buf = [] # 只存标量副本，不存对象引用 def tick(self, world, t): if t % self.period: return self.buf.append({ "t": t, "phi": polarization(world.agents), "nn": mean_nn_distance(world.agents, world.size), "n": len(world.agents), }) def dump_snapshot(world, path): import numpy as np arr = np.array([[a.x, a.y, a.vx, a.vy] for a in world.agents]) np.savez_compressed(path, arr=arr) # 位置+速度，够回放了

快照只存位置和速度四项，回放时按时间顺序重绘即可。存成压缩的数组而不是文本，是因为一千个个体、五百帧的文本日志轻松上百 MB，读的时候还要解析半天。采样周期设成 10 拍是个折中：足够看清趋势，又不至于让落盘成为瓶颈。做快速振荡的规则时改到 2~3 拍，但要知道这会让日志体积翻几倍。

4.3 一次跑批的固定记录

跑批最容易丢的是"上下文"。三个月后你看到一张曲线图，上面写着 n=400、radius=6，但你不知道当时用的是哪个版本的代码、哪种更新方式、采样相位是多少。所以每次跑必须落一份完整记录，内容至少包括：完整配置、种子、代码版本标识、开始时间、指标序列。

import json, hashlib, time, pathlib def record_run(cfg, probe, code_path="model.py", out="runs"): pathlib.Path(out).mkdir(exist_ok=True) meta = { "config": cfg, "seed": cfg["seed"], "code_sha": hashlib.sha1(open(code_path, "rb").read()).hexdigest()[:12], "finished_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "n_samples": len(probe.buf), } stem = hashlib.md5(json.dumps(cfg, sort_keys=True).encode()).hexdigest()[:8] with open(f"{out}/{stem}.meta.json", "w") as f: json.dump(meta, f, ensure_ascii=False, indent=2) with open(f"{out}/{stem}.series.json", "w") as f: json.dump(probe.buf, f)

code_sha那一行是后悔药。只要模型文件动过一个字符，哈希就变，你立刻知道两次运行的差异里混进了代码变更。没有这一行，所有"为什么重跑对不上"的排查都会变成猜谜。

5. 避坑与排查：五次返工换来的现象、原因与解决

这一章全是踩过的坑，按"看到什么—为什么—怎么办"写，能省一次返工是一次。

5.1 五个高频翻车现场

第一坑：曲线很漂亮，换个种子就崩。现象是某一组参数下群体稳定成一个大簇，极化度冲到 0.95，兴奋地截图存档。换个种子再跑，散成一盘沙。原因是只跑了一次，把单次轨迹当成了系统的稳态。解决办法是把种子当成实验维度，至少 20 个不同种子，报均值和分位数而不是单值。判断依据很简单：如果 20 个种子的极化度分布跨越了 0.3 以上，那这组参数根本不能得出"能形成有序群体"的结论。

第二坑：个体数从 300 加到 1000，运行时间翻了十倍不止。现象是跑批队列卡住，单次从几秒变成几分钟。原因是邻域查询是双重循环，复杂度 O(N²)，加三倍个体时间涨九倍。解决办法是把空间分桶：按radius划格，每个个体只查自己所在格和相邻八格。改动只发生在neighbors一个函数里，其他代码一行不用动——这也是第 2 章强调邻域查询必须集中在一处的原因。

第三坑：群体一直在抖，参数怎么调都不收敛。现象是位置序列上每个个体都在小幅高频振荡，簇结构时聚时散。原因通常有两个：一是最大步长超过了感知半径的一半，邻域信息永远滞后；二是响应系数太大，速度在期望值两侧来回过冲。解决顺序是先降max_speed到radius/2以下，再看是否还抖；仍然抖就把响应系数减半。别一上来就调权重，八成不是权重的问题。

第四坑：打开探针之后结果就变了。现象是加了监控代码，同样的种子跑出不同的曲线。原因是探针里调用了会消耗随机数或修改状态的方法。解决办法是探针只做标量复制，需要随机性就在探针内部用独立随机源。这条尤其隐蔽，因为探针本身写得"看起来"很无辜。

第五坑：重跑永远对不上，差一点点。现象是同一份配置两次运行，前 100 拍完全一致，后面开始分叉。原因可能是浮点累加顺序不一致——比如用集合去存邻居，遍历顺序随哈希变化。解决办法是所有会参与累加的容器都保持确定性顺序（列表、排序后的数组），随机源用带种子的独立实例而不是全局random。这一条排查起来最费劲，因为误差是逐步放大的，前几拍根本看不出来。

5.2 一份能贴在显示器边上的排查顺序

出问题时按症状走，别凭感觉改参数：

最后一行尤其值得记住：很多"参数效应显著"的结论，实际上是两次运行用了不同的更新方式或者不同的采样相位。把这两个变量固定下来之后，一半的"显著差异"会消失。

6. 从跑通到可复现：种子、对照与收敛判定的三件套

跑通一次不算本事，能重复跑出同一组结论才是。把仿真做成实验，靠的是三件事：种子成组、对照成对、收敛有判据。

种子成组的意思是任何一组参数都要跑多次，报告分布而不是单点。下面这段把批量结果聚合成均值和 95% 区间，直接可以贴进表里：

import json, glob, statistics as st, math def summarize(pattern="runs/*.series.json", burn_in=100): # burn_in：丢掉前 100 拍，避开初始化的暂态 phis = [] for p in glob.glob(pattern): series = json.load(open(p)) tail = [s["phi"] for s in series if s["t"] >= burn_in] if tail: phis.append(sum(tail) / len(tail)) # 每个种子一个均值 m = sum(phis) / len(phis) sd = st.stdev(phis) if len(phis) > 1 else 0.0 # 95% 区间：种子少时用 t 近似偏乐观，够做初筛 half = 1.96 * sd / math.sqrt(len(phis)) return {"n_seeds": len(phis), "mean": round(m, 4), "ci_low": round(m - half, 4), "ci_high": round(m + half, 4)} print(summarize())

burn_in这个参数必须显式写出来。群体仿真前几十拍通常是初始化暂态，把这个区间算进均值，会让所有参数组的差异都被稀释。判断 burn_in 够不够，看极化度曲线什么时候进入平稳段——一般取平稳段起点再加 20% 余量。

对照成对的意思是每次只动一个参数，其余全固定，包括种子。要比较"同步 vs 异步"，就用同一批种子各跑一遍，做配对比较，而不是各跑各的二十个种子再比均值。配对能消掉种子带来的方差，样本量小的时候尤其明显。

收敛判据是最后一件。常见做法是看指标序列的后半段是否已经不再有趋势：把后 50% 的采样点分成两段，两段均值之差小于标准差的一半，就算收敛。判据不收敛时不要急着下结论说"系统不稳定"，先加长步数再看——很多时候只是步数不够。

顺带说一个进阶用法：敏感性分析不要只做单因素扫描。单因素扫描在参数耦合强的时候会给出误导性结论，比如分离权重和感知半径单独扫都很平稳，一起变却在某个对角区域突然坍缩。预算够的话用拉丁超立方在多维空间里撒点，二十组就能看出有没有这种对角结构。

我自己的习惯是，每接一个新的群体仿真任务，先花半天只做一件事：把上面这套记录和汇总跑通，用一组随便什么参数跑二十个种子，确认文件命名、配置指纹、区间统计全部对得上，再开始调真正的模型。这个习惯是被坑出来的——有一次连续调了三天参数，最后发现所有对比都建立在一个会随遍历顺序变化的邻域实现上，等于三天的数据全部作废。从那以后我每次开新模型，都强制先把复现链路走一遍，再动规则本身。希望这套流程帮到你，少走那三天。

本文还有配套的精品资源，点击获取