关于 simpy
simpy 是一项用于 Python 进程式离散事件仿真的技能,适合构建和评估事件驱动模型。内容涵盖 SimPy 的事件与资源语义、有限运行、监测及仿真研究方法。
主要功能
- 01事件驱动的进程建模
- 02共享资源与队列原语
- 03有限仿真执行
- 04监测与事件追踪
- 05重复运行与验证指南
- 06安全且有界的仿真 CLI
适用场景
- 01模拟服务队列和生产线
- 02模拟物流与库存系统
- 03测试资源竞争和路由
- 04运行有界队列场景与重复试验
安装
复制下面的命令,在终端中运行。
npx skills add https:/ / github.com/ kalarislabs/ research-agent-skills --skill simpy安装后在 Claude Code 中输入/simpy确认已加载。
包含的文件
除 SKILL.md 外,安装时一起下载的参考资料和脚本。
references/cli-guide.mdevents.mdmonitoring.mdprocess-interaction.mdreal-time.mdresources.mdsimulation-methodology.mdsources.mdscripts/_common.pybasic_simulation_template.pybounded_queue_scenario.pyevent_trace_summary.pyreplication_runner.pyresource_monitor.pyvalidate_simulation_config.py详细介绍
SimPy
适用范围
将此技能用于基于流程的离散事件模型,其中活跃实体会让出事件并竞争资源:队列、生产系统、物流、网络、服务运营、库存及其他事件驱动系统。
SimPy 提供事件调度器和建模原语。它不会替你选择科学上有效的概念模型、输入分布、预热期、运行时长、重复次数、估计目标或因果解释。应将这些视为仿真研究方法,而不是 SimPy API 的行为。
当前版本与安装
已于 2026-07-23 验证:
- 最新稳定版本:SimPy 4.1.2,于 2026-05-24 在 PyPI 发布;源码标签
4.1.2指向提交f4381649。 - 软件包元数据要求 Python >=3.8,并标明支持 CPython 3.8-3.14 和 PyPy。SimPy 没有运行时依赖项。
- 4.1.2 增加了对 Python 3.13/3.14 的支持,并修复了现代解释器相关的测试问题。
- 上游项目和此技能均采用 MIT 许可证。
创建可复现的环境:
uv venv --python 3.13
source .venv/bin/activate
uv pip install "simpy==4.1.2"
python -c "import importlib.metadata; print(importlib.metadata.version('simpy'))"
不要默默替换为 latest 文档构建版本:它可能描述尚未发布的开发版本。请使用 references/sources.md 中的 4.1.2 版本链接。
模型工作流
- 定义目的和估计目标。 说明决策/问题、系统边界、实体、资源、状态、输出、时间单位,以及终止事件或稳态目标。
- 先编写概念模型。 记录假设、分布、路由、优先级、初始条件和省略的机制。
- 实现生成器。 SimPy 流程是一个会让出事件的 Python 生成器。使用
env.process(...)注册生成器对象。 - 限定执行范围。 为每次生产运行明确设置时间、实体、事件和重复次数上限。切勿对包含无限流程的模型调用
env.run()。 - 分离随机流。 对逻辑上不同的随机来源使用独立的本地 RNG 实例;保留种子清单。
- 有意设置监测。 在目标状态转换之后观察状态,在时间范围终点结算时间加权区间,并测试监测是否改变事件顺序。
- 验证与确认。 测试确定性边界情形、守恒恒等式、事件轨迹、队列规则和解析基准;针对既定目的,与系统证据或专家证据进行比较。
- 运行独立重复试验。 根据每次重复试验的估计值计算区间,而不是使用单次运行中相互关联的实体。
- 报告局限性。 包括初始化、未完成的实体、运行时长、种子/随机流、精度、敏感性和验证证据。切勿将仿真中的关联转化为因果主张。
提出推断性主张前,请阅读 references/simulation-methodology.md。
最小有界模型
import random
import simpy
HORIZON = 480.0
arrival_rng = random.Random(101)
service_rng = random.Random(202)
env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
completed = []
def customer(arrival):
with server.request() as request:
yield request
wait = env.now - arrival
yield env.timeout(service_rng.expovariate(1 / 6.0))
completed.append((env.now, wait))
def arrivals():
for _ in range(10_000): # Entity cap.
delay = arrival_rng.expovariate(1 / 4.0)
if env.now + delay >= HORIZON:
return
yield env.timeout(delay)
env.process(customer(env.now))
env.process(arrivals())
env.run(until=HORIZON)
数值时间范围采用左闭右开区间:恰好安排在 480.0 的常规事件不会被处理。应报告未完成的实体,而不是默默将其视为已完成的观测。
核心语义
环境与确定性排序
Environment 是单线程的。队列按仿真时间、事件优先级排序,然后按严格递增的事件 ID 排序。因此,同一时间、同一优先级的事件会按调度顺序先进先出地处理。模型进程可以表示并发,但回调会按顺序确定性地执行。
env.now:无单位的仿真时钟;请选择并记录一个时间单位。env.peek():下一个事件的时间,或无穷大。env.step():处理一个事件;队列为空时会引发EmptySchedule。env.active_process:当前正在执行的进程,否则为None。env.run():清空队列;存在循环或无限进程时使用不安全。
env.run(until=number) 和 env.run(until=event) 在边界处并不等价:
- 数值会调度一个紧急停止事件,并排除恰好发生在该时间的普通事件。
- Event 条件会在该事件的停止回调触发时返回其值。其他同一时间事件的顺序取决于优先级和调度顺序。
- 在 4.1.2 中,
Environment.step()会通过重新调度目标事件,保留StopSimulation之后剩余的回调。因此,env.run(until=target)返回后,即使目标事件的值已返回,target.processed仍可能是False,直到再执行一次step()/run()。不要将processed作为运行后判断完成状态的唯一依据。
请参阅 references/events.md 和 references/monitoring.md。
Event、Timeout、Process 和 Condition
Event只会经历一次以下状态变化:未触发 -> 已触发/已调度 -> 已处理。succeed(value)或fail(exception)会触发它一次。Timeout在创建时触发,并安排在now + delay时发生;之后无法再次手动使其成功。env.process(generator)会创建一个Process;生成器恢复执行时会收到所产出的事件值。生成器返回时,Process 会以该返回值成功结束。未捕获的异常会导致其失败。AnyOf/a | b和AllOf/a & b会产生一个ConditionValue:这是一个有序、类似字典的映射,其键为事件对象,值为对应事件的值。请使用原始事件对象检查成员关系;不要假定结果是标量。AnyOf不会取消未获胜的事件。放弃资源请求时,请显式取消尚未处理的请求;普通超时仍会保持已调度状态。
中断
process.interrupt(cause) 会调度一个紧急中断,将 simpy.Interrupt 注入目标生成器。在可能被中断的已产出工作代码周围捕获该异常,检查 interrupt.cause,更新剩余工作,然后选择继续执行、重新产出原事件或终止。
中断进程会从其当前目标事件中移除恢复回调,但不会取消该目标事件。进程不能中断自身或已终止的进程。请参阅 references/process-interaction.md。
共享资源
| 类型 | 语义 |
|---|---|
Resource | 类似信号量的 FIFO 使用名额 |
PriorityResource | 按数值优先级从低到高对排队请求排序 |
PreemptiveResource | 优先级队列,并可选择抢占当前使用者 |
Container | 同质数值级别;put/get 会等待容量/物料 |
Store | FIFO Python 对象 |
FilterStore | 首个符合请求谓词的可用项 |
PriorityStore | 按优先级顺序返回可比较的项 |
使用请求上下文管理器:
def job(env, resource):
with resource.request() as request:
yield request
yield env.timeout(3)
退出时,它会释放已获取的请求,或取消仍在等待的请求,包括异常展开期间。对于手动保留的待处理 put/get/请求,如果中断或超时导致进程放弃该请求,请调用 cancel()。
PreemptiveResource.request(priority=..., preempt=True) 使用较小的数字表示较高的优先级。被抢占的进程会收到一个 Interrupt,其原因是一个 Preempted 对象:cause.by 是发起抢占的 Process,cause.usage_since 是开始使用的时间,cause.resource 是该资源。排队优先级优先于 preempt 标志;混用抢占与非抢占请求时,需要进行明确的测试。
请阅读 references/resources.md,了解被阻塞的操作、队列规则和示例。
监控与步进
优先在状态转换时进行明确的领域观测。对于通用资源监控,可通过包装器或子类检查 count、queue、level、items、put_queue 和 get_queue。对于事件跟踪,schedule() 和 step() 是核心钩子。
队列测量对时序很敏感:
- 请求方法调用前的状态、调用后的状态、授予回调和释放回调,在同一仿真时间戳下都可能不同。
- 样本均值对事件观测加权,而非对时间加权。计算左连续状态路径下的曲线面积,并除以经过的时间。
- 添加初始和最终样本;在分析时域处结束最后一个时间区间。
env._queue、资源_env和猴子补丁都是实现细节。固定 SimPy 版本、隔离监控代码,并在升级后进行回归测试。- 跟踪每个事件会增加运行时间和内存占用;请限制跟踪记录数量。
请使用 scripts/resource_monitor.py 和 references/monitoring.md。
实时执行
simpy.rt.RealtimeEnvironment(initial_time=0, factor=1.0, strict=True) 将一个仿真时间单位映射为 factor 秒的实际时间。在严格模式下,如果计算落后,step()/run() 会引发 RuntimeError。strict=False 会容忍延迟,但不会恢复计时精度。使用 Environment 开发逻辑,然后在单独的计时测试中采用适用于平台且较宽松的容差。请参阅 references/real-time.md。
附带的安全 CLI
所有 CLI 都使用固定的内置队列模型,或汇总本地构件。它们会拒绝未知的 JSON 键、URL、符号链接、非有限数、超大输入,以及无界的时间/事件/实体/重复次数。它们绝不会评估配置文本、执行用户 Python、导入插件或调用网络服务。
# Inspect all options.
python skills/simpy/scripts/bounded_queue_scenario.py --help
python skills/simpy/scripts/replication_runner.py --help
python skills/simpy/scripts/event_trace_summary.py --help
python skills/simpy/scripts/validate_simulation_config.py --help
# Deterministic built-in scenario.
python skills/simpy/scripts/bounded_queue_scenario.py
# Independent replications with replication-level Student-t intervals.
python skills/simpy/scripts/replication_runner.py
# Validate only; no simulation runs.
python skills/simpy/scripts/validate_simulation_config.py config.json
重复运行器会拒绝仅有一次重复的区间。其区间量化的是配置模型下的 Monte Carlo 不确定性;它们既不能验证模型,也不能识别因果效应。请参阅 references/cli-guide.md。
测试
使用确定性单元测试覆盖排序、边界时间、条件、中断、所有资源调度策略、守恒、事件/实体限制、种子可复现性和监控的非干扰性。仅添加用于宽泛分布检查的随机测试,并固定种子;避免脆弱的精确样本估计。
请在完全匹配的固定环境中运行此技能的测试套件,且不生成字节码构件:
PYTHONDONTWRITEBYTECODE=1 uv run --isolated --no-project \
--python 3.13 --with "simpy==4.1.2" \
python -m unittest discover -s tests/simpy -v
参考资料
references/events.md— 调度器、生命周期、运行边界、条件references/process-interaction.md— 生成器、共享事件、中断references/resources.md— 所有 Resource、Container 和 Store 变体references/monitoring.md— 时间加权、队列计时、追踪、步进references/real-time.md— 因子、严格模式、漂移、计时测试references/simulation-methodology.md— 重复运行、预热、验证、CIreferences/cli-guide.md— 架构、边界、输出和安全的 CLI 示例references/sources.md— 标注日期的官方来源和一手方法来源