《AI Agent 开发入门》13.2 消息总线与共享状态

13.2 消息总线与共享状态

13.1 的 supervisor 把任务派给 worker，但 worker 之间不能直接说话——所有结果都经由 supervisor 中转，而且是串行的。真实的多 Agent 系统里，worker 往往需要彼此交换中间结果：检索员查到的东西，计算员得能拿到；计算员算出的数，写手得能引用。这一节给它们修一条消息总线，再给一块共享状态，把「Agent 之间怎么传数据」这件事讲清楚。

本节把 minagent 推进到 minagent/multi/bus.py：实现进程内 MessageBus（publish / subscribe / history）与带锁的 SharedState，并用 asyncio 真跑两个 worker 通过总线协作，以及共享状态的竞态对照。

13.2.1 两种协作范式：消息传递 vs 共享内存

多进程/多 Agent 协作有两条经典路线，Agent 系统里同样适用：

两条路线不是二选一，而是各管一段：「发生了什么」用消息（谁在什么时候做了什么），「当前是什么」用共享状态（任务的汇总结果、共享的草稿）。消息是流，状态是快照。

一个常见的失败设计是把所有东西都塞进共享状态：worker A 写一块、worker B 改一块，最后谁也说不清那块数据是怎么变成现在这样的。反过来，全用消息、不留共享状态，又会让「谁负责最终产物」变得模糊。本节的做法是：总线负责事件流转，共享状态负责收敛结果。

无论选哪种，Agent 场景里有三个坑必须提前知道：

• 竞态：两个 worker 同时「读—改—写」同一块状态，后写的会覆盖先写的，中间那次更新凭空消失。
• 顺序：消息的发布顺序不等于处理完成顺序。A 先发不代表 A 先处理完。
• 可观测性：并发一上，出问题时最难回答的是「当时到底发生了什么、按什么顺序」。所以总线要留 history。

13.2.2 MessageBus：进程内 pub-sub

MessageBus 是一个极简的进程内发布/订阅实现，三个方法：subscribe 注册处理器、publish 发消息、history 回看历史。

# minagent/multi/bus.py
import asyncio
from collections import defaultdict
from dataclasses import dataclass
from typing import Any, Awaitable, Callable


@dataclass
class BusMessage:
    seq: int
    topic: str
    sender: str
    payload: dict[str, Any]


Handler = Callable[[BusMessage], Awaitable[None]]


class MessageBus:
    def __init__(self) -> None:
        self._subs: dict[str, list[Handler]] = defaultdict(list)
        self._history: list[BusMessage] = []
        self._seq = 0

    def subscribe(self, topic: str, handler: Handler) -> None:
        self._subs[topic].append(handler)

    async def publish(self, topic: str, payload: dict[str, Any], sender: str = "") -> None:
        self._seq += 1
        msg = BusMessage(self._seq, topic, sender, payload)
        self._history.append(msg)
        for h in self._subs.get(topic, []):
            await h(msg)

    def history(self, topic: str | None = None) -> list[BusMessage]:
        if topic is None:
            return list(self._history)
        return [m for m in self._history if m.topic == topic]

几个设计点：

• seq 单调递增：每条消息带一个全局序号，它是「消息顺序」的权威记录。并发下靠它才能重建真实时序。
• 先记账、再派发：_history.append 发生在调用处理器之前。这样即便某个处理器中途抛异常，消息本身也已经留档——历史不会因为一次处理失败而出现空洞。
• await h(msg) 是深度优先：处理器里若再 publish，会先把那条新消息及其下游处理完，才回到当前循环。这让时序可预测（见 13.2.4 的输出），代价是慢——想要真并行得换成 asyncio.create_task，但那样顺序就不保证了。
• topic 是字符串：进程内最省事的做法。真实系统会用 ("task", task_id) 这类带参数的 topic 做定向路由。

13.2.3 SharedState：带锁的共享字典

共享状态就是一块大家都能读写的字典，但写要走原子方法：

class SharedState:
    def __init__(self) -> None:
        self._data: dict[str, Any] = {}
        self.lock = asyncio.Lock()

    def __getitem__(self, key: str) -> Any:
        return self._data[key]

    def __setitem__(self, key: str, value: Any) -> None:
        self._data[key] = value

    def snapshot(self) -> dict[str, Any]:
        return dict(self._data)

    async def atomic_add(self, key: str, delta: int = 1) -> None:
        async with self.lock:
            self._data[key] = self._data.get(key, 0) + delta

关键约定：读可以随手读，写必须走 atomic_add 这类持锁方法。snapshot 返回浅拷贝，是为了让外部拿到一致视图——如果直接返回 self._data，调用方遍历时别的协程改了它，遍历就会出错。

为什么要锁？因为「读—改—写」不是原子的。Python 的 asyncio 是单线程，看起来不会并发，但只要读和写之间有 await，控制权就可能被让出去，另一个协程趁虚而入——两个协程都读到旧值，都写回「旧值 + 1」，一次更新就丢了。13.2.5 用真实数据把这个坑跑出来。

13.2.4 真跑：两个 worker 通过总线协作

搭一条三节点的流水线：adder 收到求和请求后算出结果，squarer 收到结果后求平方，主流程订阅最终结果。用 asyncio 跑：

import asyncio

bus = MessageBus()
done: list[int] = []

async def adder(msg: BusMessage) -> None:
    total = sum(range(1, msg.payload["n"] + 1))
    await bus.publish("sum.done", {"sum": total}, sender="adder")

async def squarer(msg: BusMessage) -> None:
    s = msg.payload["sum"]
    await bus.publish("square.done", {"square": s * s}, sender="squarer")

async def collector(msg: BusMessage) -> None:
    done.append(msg.payload["square"])

bus.subscribe("sum.request", adder)
bus.subscribe("sum.done", squarer)
bus.subscribe("square.done", collector)

asyncio.run(bus.publish("sum.request", {"n": 100}, sender="main"))
for m in bus.history():
    print(f"{m.seq}  {m.sender:<7} -> {m.topic:<12} {m.payload}")
print("最终结果：", done[0])

本机实测（Python 3.14.6）：

消息序列：
  1  main    -> sum.request  {'n': 100}
  2  adder   -> sum.done     {'sum': 5050}
  3  squarer -> square.done  {'square': 25502500}
最终结果：25502500

三条消息的顺序完全符合直觉：主流程发 sum.request，adder 收到后发 sum.done，squarer 收到后发 square.done，collector 把最终结果存进 done。这正是 13.2.2 说的「深度优先」——adder 的处理还没返回，它发出的 sum.done 已经连同下游一起处理完了。

history() 的价值在这一段里体现得很直接：没有它，你只能看到最终结果 25502500；有了它，你能看到 5050 这个中间值是怎么被算出来、又被谁接手的。多 Agent 系统一旦出问题，第一件事就是回看这条消息序列——它相当于 13.2.1 里那个「当时发生了什么」的答案。

13.2.5 真跑：共享状态的竞态与加锁

现在把「读—改—写」的坑跑出来。两个协程各对同一个计数器加 1000 次，先看不加锁的写法：

async def bump_unsafe(state, times):
    for _ in range(times):
        v = state["n"]
        await asyncio.sleep(0)   # 让出控制权，暴露读-改-写窗口
        state["n"] = v + 1

async def bump_safe(state, times):
    for _ in range(times):
        await state.atomic_add("n")

bump_unsafe 在「读」和「写」之间插了一个 await asyncio.sleep(0)——它什么也不做，只是把控制权让出去。正是这个让出，制造了竞态窗口。本机实测：

无锁：期望 2000，实际 1000（丢更新）
加锁：期望 2000，实际 2000（正确）

无锁版本只加到 1000，整整丢了一半的更新。原因很直白：两个协程每次都读到同一个旧值，各自 +1 后写回，两次写合并成一次。加锁版本走 atomic_add，「读—改—写」整体在 async with self.lock 内完成，中途不让出，于是 2000 次一次不落。

这个例子里 await asyncio.sleep(0) 是人为制造的让出点，但真实代码里让出点到处都是：await model.complete(...)、await bus.publish(...)、任何一次网络 I/O。只要「读」和「写」之间隔着任何一个 await，竞态就真实存在。所以规则要反过来记：不是「有没有 await 才需要锁」，而是「只要共享状态会被多个协程写，写操作就一律持锁」。

13.2.6 取舍表

把两条路线摊开对照：

实践口径：默认用消息传递，只在「需要一个大家都能读写的收敛点」时才引入共享状态，且写操作一律持锁。多数多 Agent 系统里，消息占九成、共享状态占一成。

13.2.7 把 worker 接上总线

13.1 的 worker 是被 supervisor 直接调用的（worker.handle(task)）。有了总线，可以改成消息驱动：worker 订阅一个 topic，收到任务消息后自己执行、再发一条完成消息。装配代码只有几行：

def wire(bus: MessageBus, worker: Worker) -> None:
    async def on_task(msg: BusMessage) -> None:
        result = worker.handle(msg.payload["task"])
        await bus.publish(
            "task.done",
            {"worker": worker.name, "final": result.final},
            sender=worker.name,
        )
    bus.subscribe("task.request", on_task)

这样一改，supervisor 不再需要持有 worker 的引用，只需往 task.request 发消息；worker 的结果通过 task.done 回流。发布者与订阅者解耦——想加一个 worker，订阅同一个 topic 即可，supervisor 一行不用改。这正是消息传递范式的核心好处，也是 13.2.6 那张表里「低耦合」的具体含义。

代价是可读性下降：一条任务从「supervisor 调 worker」变成了「发消息 → 某处订阅 → 处理 → 再发消息」，看代码时得多跳几处。所以小系统里直接调用更清楚，只有 worker 数量增长、或需要动态增删 worker 时，消息驱动才划算。

一句话：总线的价值随订阅者数量增长——两个 worker 时它是过度设计，二十个 worker 时它是必需品。判断要不要上总线，就看「worker 的增减是否需要改动调用方代码」。

13.2.8 真实模型下会怎样

本节的两个 worker（adder / squarer）都是纯函数式的假处理——收到消息就确定性地算出一个数。真实 Agent 里，订阅者收到消息后要调用模型决定怎么处理，于是多出三个新问题：

• 消息变成模型输入。payload 不再是一两个数字，而是一段自然语言上下文。订阅者要把消息翻译成提示词喂给模型，模型再产出下一条消息的 payload。消息的 schema 因此要更严格（结构化字段而非裸文本），否则模型容易漏读字段。
• 处理变慢、顺序更不可控。假处理是微秒级，真实调用是秒级。多个订阅者并行时，history 里的 seq（发布序）和处理完成序会明显分叉，排查问题必须两个顺序都看。
• 失败要回灌成消息。真实调用会超时、会被限流。健壮的总线不该让处理器抛异常打断派发，而应把失败包成一条 *.failed 消息发出去，让上游决定重试还是降级——这正是第 14 章要展开的。

一句话：总线把「谁跟谁说话」这件事标准化了，但「说了什么、听没听懂、说错了怎么办」仍是模型的事。本节先把通道修好、把竞态讲透，语义层的健壮性留给后续章节。

阅读导航：上一节：13.1 角色与分工 · 下一节：13.3 协作模式对照 。