多智能体系统中的通信协议与并发控制
[!NOTE] 执行摘要（TL;DR） Actor 模型与邮箱并发：使用 Actor 消息信封和隔离状态机来协调多智能体集群，消除死锁风险，不再需要裸互斥锁。乐观并发与背压：在共享任务板上利用 OCC 版本控制，配合令牌桶限流器，平滑缓冲 LLM API 的突发流量。生态堆叠：各个 actor 运行在《构建轻量级 Agent 框架》和《高级函数调用引擎》之上。
多智能体系统中的消息所有权至关重要
当任务可拆分、结果可验证、并行收益超过通信成本时，多个智能体就能发挥作用。本文演示进程内的 asyncio actor、有界投递、CAS 更新和令牌桶。本地队列不是持久化的分布式消息代理。
信封、邮箱与背压
每个 actor 按顺序消费自己的邮箱，同时其他 actor 可以并发等待 I/O。冻结的 dataclass 并不能冻结嵌套字典，因此投递时会为每个接收者复制 payload。有界队列在超过投递截止时间后抛出 BackpressureError。广播不是原子操作：当某个接收者超时时，其他接收者可能已经收到了事件。调用方需要按消息 ID 去重，并处理部分失败。本示例未实现持久化、投递确认或请求-响应 future。
flowchart LR
Developer[Developer] --> Bus[Bounded message bus]
Bus --> Reviewer[Reviewer mailbox]
Bus --> Scanner[Scanner mailbox]
Reviewer --> Coordinator[Collect simulated results]
Scanner --> Coordinator
# 协调器 --> CAS（CAS 文档提交）
上面是一个可完整运行的示例。下载 Python 源码，在 Python 3.11+ 环境下执行 python3 actor_runtime.py。评审者通过字符串检查来演示消息顺序，并非真实的安全分析。协调器会等待双方完成后再提交；关闭时先清空各邮箱中的消息，再取消任务。
"""
进程内 actor 实现，带边界投递能力；不是分布式持久化消息中间件。
"""
import asyncio
import copy
import json
import time
import uuid
from dataclasses import asdict, dataclass, field
from enum import Enum
class MessageType(str, Enum):
REQUEST = "request"
RESPONSE = "response"
EVENT = "event"
@dataclass(frozen=True)
class MessageEnvelope:
source_actor: str = ""
target_actor: str = ""
message_type: MessageType = MessageType.REQUEST
correlation_id: str = ""
payload: dict = field(default_factory=dict)
message_id: str = field(default_factory=lambda: str(uuid.uuid4()))
timestamp: float = field(default_factory=time.time)
def to_json(self):
return json.dumps(asdict(self))
@classmethod
def from_json(cls, value):
data = json.loads(value)
data["message_type"] = MessageType(data["message_type"])
return cls(**data)
class BackpressureError(RuntimeError):
pass
class MessageBus:
def __init__(self, mailbox_size=100, delivery_timeout=1):
if mailbox_size < 1 or delivery_timeout <= 0:
raise ValueError("mailbox_size 与 delivery_timeout 必须为正数")
self.mailbox_size = mailbox_size
self.delivery_timeout = delivery_timeout
self.actor_mailboxes = {}
self.subscriptions = {}
def register_actor(self, actor_id):
if actor_id in self.actor_mailboxes:
raise ValueError("actor 已注册")
mailbox = asyncio.Queue(maxsize=self.mailbox_size)
self.actor_mailboxes[actor_id] = mailbox
return mailbox
def subscribe(self, topic, actor_id):
self.subscriptions.setdefault(topic, set()).add(actor_id)
async def send(self, envelope):
mailbox = self.actor_mailboxes.get(envelope.target_actor)
if mailbox is None:
raise ValueError("未知 actor")
await self.
（本段截断，后续内容在下一分段中。）
async def _deliver(self, mailbox, envelope):
try:
await asyncio.wait_for(mailbox.put(copy.deepcopy(envelope)), self.delivery_timeout)
except asyncio.TimeoutError as error:
raise BackpressureError("mailbox delivery timed out") from error
async def publish(self, topic, envelope):
# Each subscriber receives a separate payload; delivery is not an atomic broadcast.
await asyncio.gather(*(
self._deliver(self.actor_mailboxes[actor], envelope)
for actor in self.subscriptions.get(topic, ())
if actor in self.actor_mailboxes
))
class BaseActor:
def __init__(self, actor_id, bus):
self.actor_id = actor_id
self.bus = bus
self.mailbox = bus.register_actor(actor_id)
self.task = None
self.errors = []
async def start(self):
if self.task is not None:
raise RuntimeError("actor already started")
self.task = asyncio.create_task(self._event_loop())
async def stop(self, timeout=5):
if self.task is None:
return
try:
await asyncio.wait_for(self.mailbox.join(), timeout)
finally:
self.task.cancel()
await asyncio.gather(self.task, return_exceptions=True)
self.task = None
async def _event_loop(self):
while True:
envelope = await self.mailbox.get()
try:
await self.on_message(envelope)
except asyncio.CancelledError:
raise
except Exception as error:
self.errors.append(error)
finally:
self.mailbox.task_done()
async def on_message(self, envelope):
raise NotImplementedError
class ConcurrentDocumentStore:
def __init__(self):
self._content, self._version = "", 1
self._lock = asyncio.Lock()
async def get_document(self):
async with self._lock:
return {"content": self._content, "version": self._version}
async def update_document(self, new_content, expected_version, author):
async with self._lock:
if self._version != expected_version:
return False
self._content = new_content
self.
_version += 1
return True
class TokenBucketRateLimiter:
def __init__(self, capacity, refill_rate_per_sec):
if capacity <= 0 or refill_rate_per_sec <= 0:
raise ValueError("capacity and refill rate must be positive")
self.capacity, self.refill_rate = capacity, refill_rate_per_sec
self.tokens, self.last_refill = float(capacity), time.monotonic()
self._lock = asyncio.Lock()
async def acquire(self, tokens_needed=1):
if not 0 < tokens_needed <= self.capacity:
raise ValueError("requested tokens must fit bucket capacity")
while True:
async with self._lock:
now = time.monotonic()
self.tokens = min(
self.capacity,
self.tokens + (now - self.last_refill) * self.refill_rate,
)
self.last_refill = now
if self.tokens >= tokens_needed:
self.tokens -= tokens_needed
return
delay = (tokens_needed - self.tokens) / self.refill_rate
await asyncio.sleep(delay)
async def main():
bus, store = MessageBus(), ConcurrentDocumentStore()
limiter = TokenBucketRateLimiter(2, 2)
class Developer(BaseActor):
async def on_message(self, envelope):
await limiter.acquire()
await self.bus.publish(
"code",
MessageEnvelope(
source_actor=self.actor_id,
message_type=MessageType.EVENT,
correlation_id=envelope.correlation_id,
payload={"code": "def login(): return True"},
),
)
class Reviewer(BaseActor):
def __init__(self, actor_id, bus):
super().__init__(actor_id, bus)
self.approved = False
async def on_message(self, envelope):
self.approved = envelope.payload["code"].startswith("def login")
developer, reviewer, scanner = (
Developer("dev", bus),
Reviewer("review", bus),
Reviewer("scan", bus),
)
actors = [developer, reviewer, scanner]
bus.subscribe("code", "review")
bus.subscribe("code", "scan")
for actor in actors:
await actor.start()
try:
await bus.send(
MessageEnvelope(target_actor="dev", correlation_id="example-1")
)
await developer.mailbox.join()
await asyncio.gather(
reviewer.mailbox.join(),
scanner.mailbox.join(),
)
join()) # Simulated checks only; the coordinator commits after both completed.
if not all(actor.errors == [] for actor in actors) or not (reviewer.approved and scanner.approved):
raise RuntimeError("review failed")
snapshot = await store.get_document()
merged = await store.update_document("def login(): return True", snapshot["version"], "coordinator")
print("Simulated merge: ", merged)
finally:
for actor in actors:
await actor.stop()
if __name__ == "__main__":
asyncio.run(main())
扩展本地示例
CAS（比较并交换）能发现过期的写入，但不会合并冲突的文档。真实的数据库必须把「比较版本」和「更新内容」做成一个原子操作。
令牌桶（token bucket，一种限流算法）用的是单调时钟，并会拒绝超出桶容量的请求。它限制的是请求速率，而不是每分钟的 token 数，也不管并发数，更不涉及各家模型服务商自己的重试策略。
跨进程部署还需要序列化、身份认证、可靠投递、确认机制、稳定的幂等键（idempotency key，同一个请求重发也只生效一次），以及死信处理（dead-letter，投递失败的消息单独存放）。asyncio 的锁只能保护单个进程。
延伸阅读：工具执行进阶 与 审批工作流。
常见问题（FAQ）
Q1：消息加上 frozen=True 就算不可变了吗？
不算。嵌套的字典仍然可以被修改。示例里每次投递都会复制一份；分布式传输则应把每条消息独立序列化。
Q2：邮箱写满了会怎样？
投递最多等待 delivery_timeout，超时后抛出明确的错误。接下来是重试、丢弃还是落盘持久化，由调用方决定。
Q3：处理函数抛异常，会不会让 queue.join 永远等下去？
示例在 finally 里调用了 task_done()，并把错误记录下来交给协调者。但生产系统仍然需要失败消息的重试或死信策略。