多智能体通信怎样记录消息先后？用 Lamport 逻辑时钟核对发送与接收

三个AI智能体分别整理资料、审查资料、汇总结果时，日志中的本机时间可能不同步，接收顺序也可能变化。要说明“这次接收发生在那次发送之后”，可以在消息里传递Lamport逻辑时钟，并在接收端推进计数。它记录事件顺序约束，不能单凭数字大小证明两条消息有因果关系。

本文面向已理解发送与接收、想核对多智能体通信日志的开发者。下面的A、B、C是本地模拟角色，没有调用模型或真实网络。示例已在Python 3.11.15执行；算法依据是Lamport发表于1978年的原论文，读取于2026-10-03北京时间。

先定义消息与事件，别把计数当秒数

A代表资料整理角色，B代表审查角色，C代表汇总角色。每个角色独立保存整数clock，初始为0；本地动作、发送、接收都作为事件。发送消息带sender、recipient、clock、text四项，接收方必须确认recipient等于自己。文字可以由模型生成，但计数和消息身份由程序负责，不能让模型随意填一个“看起来递增”的时钟。

本地事件和发送事件先加1。接收事件取本地计数与消息计数中较大者，再加1。这样本地相邻事件严格递增，同一消息的发送计数小于接收计数。Lamport原论文的Logical Clocks部分给出这两个方向的约束，并明确逆命题不成立；这里采用整数计数器实现它们。

让AI提出实现时，把验收关系一起给出

实现本地A/B/C三角色的消息传递示例：每角色独立clock和事件列表，发送先加1，接收max(local,incoming)+1。
验证m1发送2/接收3，m2发送4/接收5；再故意让B先收到A的较新消息、后收到较旧消息，B计数不能倒退。
消息收件人错误时先抛错，不改变接收角色clock或事件列表。另举两个没有交换消息的独立角色，说明小计数不证明因果。只模拟，不接外部系统。

检查生成代码是否把“收到旧消息”误处理为本机计数回退，也检查它是否在验证收件人前已经写入事件列表。后者会留下伪造的接收记录。接收计数递增不等于业务版本递增，模型不能把这一点解释成“所有旧状态更新都自动安全”。

保存并运行可复算示例

把以下代码保存为clock_demo.py，在空目录运行python -X utf8 clock_demo.py。Python 3.11标准库即可运行，不需要账号、密钥或第三方服务。dataclass用于固定消息字段，不意味着消息已经认证。

from dataclasses import dataclass

NAMES = {'A', 'B', 'C'}

@dataclass(frozen=True)
class Message:
    sender: str
    recipient: str
    clock: int
    text: str

class Agent:
    def __init__(self, name):
        if name not in NAMES:
            raise ValueError('unknown agent')
        self.name, self.clock, self.events = name, 0, []

    def local(self, label):
        self.clock += 1
        self.events.append((self.clock, 'local', label))

    def send(self, recipient, text):
        if recipient not in NAMES or recipient == self.name:
            raise ValueError('invalid recipient')
        self.clock += 1
        self.events.append((self.clock, 'send', recipient))
        return Message(self.name, recipient, self.clock, text)

    def receive(self, msg):
        if (not isinstance(msg, Message) or msg.recipient != self.name
                or msg.sender not in NAMES or msg.sender == self.name
                or type(msg.clock) is not int or msg.clock < 1):
            raise ValueError('invalid message')
        self.clock = max(self.clock, msg.clock) + 1
        self.events.append((self.clock, 'receive', msg.sender))

a, b, c = [Agent(n) for n in 'ABC']
a.local('prepare')
m1 = a.send('B', 'source ready')
b.receive(m1)
m2 = b.send('C', 'review ready')
c.receive(m2)
assert (m1.clock, b.events[0][0], m2.clock, c.clock) == (2, 3, 4, 5)
print('chain', m1.clock, b.events[0][0], m2.clock, c.clock)

m3 = a.send('B', 'older')
m4 = a.send('B', 'newer')
b.receive(m4)
b.receive(m3)
assert b.clock == 6
print('out_of_order_receive', [e[0] for e in b.events[-2:]])
for agent in [a, b, c]:
    clocks = [e[0] for e in agent.events]
    assert all(x < y for x, y in zip(clocks, clocks[1:]))

before = (c.clock, c.events[:])
try:
    c.receive(m1)  # recipient is B
except ValueError:
    pass
else:
    raise AssertionError('wrong recipient accepted')
assert (c.clock, c.events) == before
print('wrong_recipient_rejected_unchanged', True)

# No message connects these two agents' events.
x, y = Agent('A'), Agent('B')
x.local('independent x')
y.local('independent y1'); y.local('independent y2')
assert x.clock < y.clock
print('independent_clocks', x.clock, y.clock)

逐项核对实际输出

chain 2 3 4 5
out_of_order_receive [5, 6]
wrong_recipient_rejected_unchanged True
independent_clocks 1 2

chain的2、3、4、5对应表中的发送与接收。B此前停在4，先收时间戳4的较新消息时变5，再收时间戳3的较旧消息时变6。因此本地接收顺序严格递增，却没有把业务内容重新排成发送顺序。wrong_recipient的True只表示程序拒绝交给C的m1，且C的计数和事件列表与调用前相同。

最后两角色从未交换消息，计数仍分别是1和2。数字1小于2无法推出第一个事件导致第二个事件。若已知发送—接收边和本地事件链，可以沿它们推导先后；若只剩clock列，就缺少这些因果证据。给同值附角色ID可以获得稳定排序，但人为打破平局仍不证明因果。

接回真实多智能体通信前，明确哪些事还没解决

本例只验证受信任本地对象的时钟规则。真实消息还需要身份认证、来源校验、消息ID、重传处理、持久化以及重启后的计数恢复。重复接收同一消息在这里会成为新的接收事件，不会自动去重；丢消息也不会由逻辑时钟补回来。

若业务是“新版草稿覆盖旧版”，还要使用业务版本与明确的冲突规则。只按接收计数保存最后到达内容，仍可能把旧草稿当最新稿。时钟没有验证模型回答的事实，也不替代权限检查、交接条件或真实物理时间。

下一步用自己的事件表验收

保留角色、事件类型、消息ID、发送计数和接收计数五列，逐消息检查发送小于接收，再逐角色检查本地严格递增。对没有消息边相连的事件标为关系未确定，不强行从日志排序推断依赖。若要重放业务，另存业务版本和原始内容摘要，并按实际系统契约设计重放规则。