多智能體系統的通訊協定與並行控制實作示例
Communication Protocols and Concurrency Control in Multi-Agent Systems
文章以可執行的 Python asyncio 範例，示範多智能體系統如何以 Actor 郵箱、有界訊息傳遞、CAS 更新及 Token Bucket 限流協調工作。CAS 可偵測過期寫入，但不會合併衝突文件。本地佇列並非持久化的分散式 broker；跨進程部署需加入序列化、身份驗證、持久化交付、確認機制、穩定冪等鍵及死信處理。
[!NOTE] Executive Summary (TL;DR)
- Actor Model & Mailbox Concurrency: Coordinates multi-agent swarms using Actor message envelopes and isolated state machines, eliminating deadlock hazards and raw mutexes.
- Optimistic Concurrency & Backpressure: Leverages OCC versioning on shared task boards alongside Token Bucket rate limiters to buffer LLM API bursts smoothly.
- Ecosystem Stacking: Individual actors run on top of Building Lightweight Agent Frameworks and Advanced Function Calling Engines.
Message ownership matters in multi-agent systems
Multiple agents help when tasks are separable, results are verifiable, and parallel gains exceed communication costs. This article demonstrates in-process asyncio actors, bounded delivery, CAS updates, and token buckets. A local queue is not a durable distributed broker.
Envelopes, mailboxes, and backpressure
Each actor consumes its mailbox sequentially while other actors can await I/O concurrently. A frozen dataclass does not freeze nested dictionaries, so delivery copies the payload for each recipient.
Bounded queues raise BackpressureError after a delivery deadline. Broadcast is not atomic: some recipients may already have received an event when another times out. Callers need message-ID deduplication and partial-failure handling. The example does not implement persistence, delivery acknowledgements, or request-response futures.
flowchart LR
Developer[Developer] --> Bus[Bounded message bus]
Bus --> Reviewer[Reviewer mailbox]
Bus --> Scanner[Scanner mailbox]
Reviewer --> Coordinator[Collect simulated results]
Scanner --> Coordinator
Coordinator --> CAS[CAS document commit]Complete runnable example
Download the Python source and run python3 actor_runtime.py with Python 3.11+. Reviewers perform string checks to demonstrate ordering, not real security analysis. The coordinator waits for both before committing; shutdown drains mailboxes before cancelling tasks.
"""In-process actors with bounded delivery; not a distributed durable broker."""
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 and delivery_timeout must be positive")
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 already registered")
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("unknown actor")
await self._deliver(mailbox, envelope)
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())
# 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())
Extending the local example
CAS detects stale writes but does not merge conflicting documents. Real databases must compare the version and update the content atomically. The token bucket uses a monotonic clock and rejects requests larger than its capacity. It limits request rate, not tokens per minute, concurrency, or provider-specific retry behavior.
Cross-process deployment needs serialization, authentication, durable delivery, acknowledgements, stable idempotency keys, and dead-letter handling. An asyncio lock protects only one process. See advanced tool execution and the approval workflow.
Frequently Asked Questions (FAQ)
Q1: Is frozen=True enough for immutable messages?
No. Nested dictionaries remain mutable. The example copies each delivery; distributed transports should serialize independent messages.
Q2: What happens when a mailbox fills?
Delivery waits for at most delivery_timeout, then raises an explicit error. The caller chooses retries, shedding, or persistence.
Q3: Can a handler exception leave queue.join waiting forever?
The example calls task_done() in finally and records the error for the coordinator. Production systems still need failed-message retry or dead-letter policies.
References: asyncio Queue, Python dataclasses.
Originally published at LLM Systems Architecture on blog.llmgo.top.
來源：DEV Community · dev.to