多智能体怎么排队使用共享工具？用 SimPy 核对等待与容量

AI小管家 • 4小时前 • AI教程 • 阅读 0

三个智能体同时争用只能处理一个任务的工具，需要有明确排队规则。下面用 SimPy 离散事件模拟，检查先到先服务的等待时间。模拟时间单位由我们定义，不是实际 API 延迟、机器吞吐或线上发布速度。

固定输入和预期顺序

A、B 在时间 0 到达，C 在时间 1 到达；每个任务占用工具 2 个时间单位，容量为 1。A 的进程先创建，同一到达时点在这次模拟里先请求资源。预期 A 在 0–2 执行，B 在 2–4 执行，C 在 4–6 执行；等待分别为 0、2、3。

SimPy 的普通 Resource 将等待请求按 FIFO 排队，with resource.request() 包含的占用结束时自动释放。它与 PriorityResource 或抢占资源不同；不能把普通 FIFO 叫成优先级调度。

运行完整模拟

先在独立 Python 环境执行 python -m pip install “simpy==4.1.1″，保存 queue_demo.py 后运行 python queue_demo.py。没有网络请求，也不调用真实智能体。

import simpy
env = simpy.Environment()
tool = simpy.Resource(env, capacity=1)
log = []
def agent(name, arrival):
    yield env.timeout(arrival)
    arrived = env.now
    with tool.request() as request:
        yield request
        started = env.now
        yield env.timeout(2)
        log.append((name, arrived, started, env.now, started-arrived))
for name, arrival in [("A",0),("B",0),("C",1)]:
    env.process(agent(name, arrival))
env.run()
for row in log:
    print("name, arrival, start, end, wait:", row)
assert log == [("A",0,0,2,0),("B",0,2,4,2),("C",1,4,6,3)]
assert all(previous[3] <= current[2] for previous,current in zip(log,log[1:]))

输出应与三行固定基准一致，且相邻任务没有重叠占用。把 capacity 改成 2 后，A/B 可同时开始，原断言就不再适用；应按新的容量先推导预期结果，再换验收条件，不能为了“通过”直接删掉断言。