多Agent协作下的点对点通信协议设计与实施

做 Agent 开发做到一定阶段一定会撞上同一个问题单机单 Agent 的 demo 跑得风生水起一旦想让多个 Agent 协作立刻就会被通信方式卡住。HTTP 轮询太笨重中心化消息队列引入一堆运维依赖手动维护 WebSocket 连接又容易出现半开连接、消息乱序、重试风暴这些破事。最近我花了两周时间把一个内部项目全面切到 hermes peer 这套自定义的点对点通信协议上同时跑通了一个前端 后端 多 Agent 协作的全栈案例。这篇文章不聊 PPT 架构全部是我实际踩过坑、调过参、线上验证过的内容想搞清楚 Agent 间点对点通信协议到底怎么设计、代码怎么落地可以直接照着抄。hermes peer 的核心思路很简单把 Agent 之间每一次交互都定义成具备明确语义的消息帧用统一的信封格式承载协议层只负责可靠送达和会话管理业务层完全解耦。它解决的核心痛点是多个 Agent 之间既不想通过中心化 Broker 中转、又需要稳定双向通信的场景适合正在做多智能体协作、私有化部署 Agent 集群、或者想用手写协议替代 HTTP 长轮询的开发者参考。1. 为什么 Agent 间通信不能继续走中心化老路1.1 中心化模式的三个痛点先说说我为什么没有直接选现成的消息队列或者中心化调度框架。不是说 MQ 不好而是要分场景。早期项目里我用过一个中心化的任务队列作为多个 Agent 的通信中枢所有消息都先发给队列再由队列分发给对应的下游 Agent。单机 demo 阶段一切正常一旦 Agent 数量超过五个、消息频率上升之后问题就暴露得特别明显。第一个痛点是消息语义被压缩得过于单一。队列里流动的本质上还是 JSON payload谁发的、发给谁、期望什么响应、超时怎么办这些语义在业务层反复手写每个 Agent 的实现都不同久而久之就成了一团乱麻。第二个痛点是单点依赖中心化节点一旦抖动所有协作方全部感知到排查链路特别长。第三个痛点是消息回传方向非常别扭。Agent A 给 Agent B 派任务B 执行完要把结果回传给 A在中心化架构下要么再建一个逆向队列要么通过回调接口要么靠消息里带 correlationId 硬匹配每一条路都是在给业务层增加复杂度。所以在设计 hermes peer 时我给自己定了一个原则通信要回归对等。任何一个 Agent 既可以发起请求也可以承接请求消息的流向由协议层直接表达而不是靠业务层自己拼拼凑凑。1.2 hermes peer 的核心思路把“调用”变成“交谈”传统 RPC 或者 HTTP 接口本质上是“调用”思维我请求你你返回结果一次请求一次响应结束。但是 Agent 之间的协作往往不是一次简单的请求-响应就完事。规划 Agent 把任务拆分后发给执行 Agent执行 Agent 在执行过程中可能需要向知识 Agent 查询背景资料同时还要实时上报进度最后才能回传最终结果。这个过程更像是一场多轮对话而且对话中的每一条消息都可能有独立的方向和语义。hermes peer 的设计目标就是把这套多轮交互语义化。协议层定义了明确的请求、响应、流式数据、事件通知、心跳控制这几类基础语义Agent 之间通过这些语义的排列组合形成各种协作模式。规划 Agent 发起的是一个 request执行 Agent 收到后可以返回一个 stream 消息持续上报中间状态最终再用 response 返回正式结果知识 Agent 收到查询时返回 response其他 Agent 也可以通过 event 广播状态变化。每一种协作都变成了协议层明确表达的东西而不是 JSON 里某个魔数。1.3 协议要解决的根本问题清单动手设计协议之前我先把要解决的问题列成了一个清单。首先当然是寻址点对点通信要求每条消息明确知道发送方和接收方是谁这需要一套 Agent ID 和地址发现机制。其次是消息可靠性点对点直连不像中心化队列那样有持久化保障所以协议层必须处理连接断开、消息重试、幂等消费这些底层问题。第三是消息语义不同 Agent 之间的协作模式不一样协议层要提供足够的消息类型来表达不同的交互意图而不是把所有情况都塞进一个通用消息里。第四是连接生命周期管理Agent 随时可能启动、退出、崩溃、重启协议层需要有心跳检测和优雅退出的机制。最后是协议的演进能力Agent 的版本不可能永远保持一致消息格式要有向前兼容的余地。把这些清单捋顺之后hermes peer 的框架就清晰了。它不是一个庞大的分布式框架而是轻量、明确的点对点通信协议加上一个参考实现。2. 协议设计hermes peer 的消息格式与握手流程2.1 Envelope 结构设计所有 hermes peer 的消息都用一个统一信封 Envelope 包装就像寄快递不管里面装的是什么外面的面单格式必须是统一的。Envelope 分 Header 和 Payload 两部分Header 负责传输控制Payload 负责业务内容。Header 字段我用的是这套type 表示消息类型version 是协议版本号sender 和 receiver 是双方 Agent IDsessionId 用于标记一次完整的会话messageId 是单条消息的唯一 IDtimestamp 是发送时间correlationId 用于关联请求与响应flags 是扩展位。Payload 则是一段字节流协议层不知道里面是什么由接收方根据消息类型和业务语义自行解析。{ header: { type: request, version: 1.0, sender: planner-01, receiver: worker-03, sessionId: sess_8f3a91c2, messageId: msg_001, correlationId: corr_0001, timestamp: 1710000000123, flags: 0 }, payload: { action: execute_task, taskId: task_42, params: {source: repo_a, command: build} } }type 字段目前定义了六种取值request、response、stream、event、heartbeat、bye。request 和 response 是一对同步语义stream 用于流式数据上报event 是单向事件通知heartbeat 是探活bye 是优雅退出声明。version 字段非常关键我之前在别的项目里吃过协议升级的亏所以从第一天起就强制写入版本号。sender 和 receiver 必须要填任何一条没有明确发送方和接收方的消息在协议层直接丢弃。2.2 消息类型与语义定义Request 消息是 Agent 向另一个 Agent 发起任务请求必须携带 correlationId接收方在处理完成后必须回 response超时机制由发起方控制。Response 消息是请求的响应必须携带与请求相同的 correlationId并且带一个 status 字段表示成功或失败status 里我固定用 0 表示成功非 0 表示业务错误这样协议层和业务层能快速区分传输层失败和业务层失败。Stream 消息用来表达流式中间状态。比如一个执行 Agent 正在跑一个耗时任务它会不断发送 stream 消息上报进度最后发送一条 response 表示任务收敛。Event 消息用于广播或者单向通知不要求响应。最典型的是一个 Agent 上线后广播“我准备好接活了”其他 Agent 收到后各自更新自己的路由表。Heartbeat 消息是固定间隔的探活包Bye 消息用于优雅退出。消息类型方向是否需要响应典型场景request点对点是派发任务、发起查询response点对点回传否返回结果、返回错误stream点对点否进度上报、日志推送event广播/单播否Agent 上线通知、状态变更heartbeat点对点可选连接探活bye点对点否优雅退出消息类型之间最核心的区别在于“是否要求响应”以及“消息是否有终点”。request 和 response 是一对不可分割的语义stream 是单向持续语义event 是一锤子买卖。所有 Agent 在解析消息时先按 type 决定处理策略再进入 payload 的业务逻辑协议层与业务层由此解耦。2.3 握手、心跳与优雅退出握手流程是这个协议里我觉得最值得说的细节。Agent 之间建立连接后不会立刻开始发业务消息而是先完成一次 Triple Handshake。连接建立后发起方发送 hello 请求包含自己的 Agent ID、协议版本、能力列表。接收方收到后如果版本兼容回送 hello-ack并附上自己的 Agent ID 和能力列表如果不兼容回送版本不匹配的错误并主动断开。发起方收到 hello-ack 后回送确认包。A B | hello (id, version) | |------------------------| | hello-ack (id, ver) | |------------------------| | hello-confirm | |------------------------|这一步看起来多此一举但实际上解决了三个问题。第一是双方在进入业务状态前就完成了能力协商避免中间才发现对方不支持某个特性。第二是双向确认了彼此的身份防止连接被复用导致消息串线。第三是建立了会话状态的基础握手成功后双方才创建对应的 Peer 会话对象。心跳这块我踩过坑。最初我用的是固定 30 秒发送一次心跳但是网络抖动时30 秒的探测周期太慢导致上游 Agent 以为下游 Agent 挂了实际上只是网络临时抖动。后来改成自适应心跳正常状态下 30 秒一次连续收到三次心跳响应后如果通信频率变高心跳间隔自动缩短到 10 秒如果最近有业务消息往来则跳过心跳发送。这个机制实测下来能显著减少误杀场景。退出流程同样重要。Agent 在退出前必须发送 bye 消息广播自己的离线状态之后清理自己维护的所有 Peer 会话。接收方收到 bye 后会将该 Peer 标记为离线并触发重连或任务重新派发逻辑。不发送 bye 直接断连的 Agent 会被其他节点视为异常离线从而触发异常重试流程。这个差异是协议层能感知的。3. 代码实操用 Python 实现一个最小可用的 hermes peer 节点3.1 技术选型asyncio 自定义协议实现 hermes peer 参考代码时我选了 Python 的 asyncio。这不是随便选的。首先 asyncio 的事件循环天然适合处理大量长连接一个 Agent 要同时跟多个 Peer 保持通信用同步多线程模型会导致资源浪费严重。其次 asyncio 的 StreamReader 和 StreamWriter 封装了底层 socket 操作协议帧的读写可以直接在这层做。第三是 asyncio 生态里集成 WebSocket、HTTP 都很方便便于做全栈应用的桥接层。这里要说明一下hermes peer 协议本身不绑定 Python它也完全可以跑在 Node.js、Go 或者其他语言上。之所以这一节用 Python 做参考实现是因为 Python 在 Agent 生态和 AI 工具链上最成熟大家都是用它做原型、做验证跑通之后再迁移到高性能语言。核心依赖只有 asyncio 和 json协议层不引入任何第三方库。这样做的好处是环境搭建极简任何有 Python 3.10 以上环境的机器都能直接跑。3.2 Envelope 编解码与 Peer 管理器协议帧的编码规则是这样的先用 4 字节的大端整数表示消息总长度然后是消息体的 JSON 字节流。消息体的具体结构就是前面说的 Envelope。为什么不用 JSON Lines 而是用长度前缀因为 Agent 之间往往是通过 TCP 长连接通信一条消息可能被 TCP 协议拆成多个包发送接收方如果没有明确的边界规则就无法完整重组消息。长度前缀是最简单的定界方案解析起来也很快。import asyncio import json import struct from dataclasses import dataclass, field import time import uuid dataclass class Envelope: msg_type: str sender: str receiver: str session_id: str message_id: str correlation_id: str timestamp: int version: str 1.0 flags: int 0 payload: dict field(default_factorydict) def encode(self) - bytes: header { type: self.msg_type, version: self.version, sender: self.sender, receiver: self.receiver, sessionId: self.session_id, messageId: self.message_id, correlationId: self.correlation_id, timestamp: self.timestamp, flags: self.flags } body {header: header, payload: self.payload} data json.dumps(body, ensure_asciiFalse).encode(utf-8) return struct.pack(I, len(data)) data staticmethod def decode(stream: bytes) - Envelope: # 注意这里是示意实际使用中需要结合 StreamReader 边读边解 obj json.loads(stream.decode(utf-8)) header obj[header] return Envelope( msg_typeheader[type], senderheader[sender], receiverheader[receiver], session_idheader[sessionId], message_idheader[messageId], correlation_idheader[correlationId], timestampheader[timestamp], versionheader[version], flagsheader[flags], payloadobj.get(payload, {}) )这里有个细节值得单独说correlationId 和 messageId 的区别。correlationId 用于关联一次完整的请求-响应过程发起 request 时生成一个后续所有跟这次请求相关的消息都携带同一个 correlationId。messageId 是每一条消息的唯一标识用于幂等处理和去重。一个会话中会有多条消息每条消息 messageId 不同但 correlationId 相同。Peer 管理器是每个 Agent 内部的核心组件负责维护一张 Peer 表表里记录了每个 Peer 的 Agent ID、连接状态、远端地址、最近活跃时间。Peer 管理器同时也是消息路由的核心所有发出去的消息都要经过 Peer 管理器查询对端连接所有收到的消息也要经过 Peer 管理器登记活跃状态。这个设计让业务代码不需要关心底层连接细节只需要调用 send(envelope) 和 on_message(envelope) 两个接口。class PeerManager: def __init__(self, agent_id: str, version: str 1.0): self.agent_id agent_id self.version version self.peers {} # agent_id - (reader, writer, status) self.pending {} # correlation_id - asyncio.Future self.handlers {} # msg_type - callback async def send(self, envelope: Envelope): peer self.peers.get(envelope.receiver) if peer is None: raise RuntimeError(fpeer {envelope.receiver} not found) _, writer, status peer if status ! active: raise RuntimeError(fpeer {envelope.receiver} not active) writer.write(envelope.encode()) await writer.drain()这段代码把路由逻辑压得非常薄。send 方法根据 receiver 查询 Peer 表拿到连接后直接写数据。真正复杂的部分在 Receiver 侧收到一条消息后先解析 Envelope再根据 msg_type 分发到不同 handler。request 类型的消息会被放入请求处理器response 类型的消息会找到对应的 Future 并设置结果stream 类型的消息触发进度回调event 类型的消息直接广播给所有本地监听者。3.3 Agent 角色与协作流光有协议和 Peer 管理器还不够要真正体现出 hermes peer 的价值得让不同角色的 Agent 跑起来。我在案例中设计了三个角色planner、worker、knowledge。Planner 是编排者接收用户或上游系统下达的任务把任务拆解成多个子任务然后分发给不同的 worker。Worker 是执行者收到子任务后执行具体的动作比如运行一段代码、生成一份报告、调用一个外部 API执行过程中通过 stream 消息上报进度完成后通过 response 返回结果。Knowledge 是知识库 Agent负责存储和检索结构化信息其他 Agent 通过 request 向它发起查询它通过 response 返回结果。这三个角色之间的消息流是这样的planner 收到总任务后先向 knowledge 发起一个查询看历史任务中有没有相似的执行经验拿到上下文后planner 把任务拆成三个子任务分别发给 worker-a、worker-b、worker-cworker 之间通过 event 消息同步协调信息避免做重复计算所有 worker 完成后planner 汇总结果并把最终报告写回知识库。这个协作流如果是用 HTTP 方式实现会非常痛苦。planner 需要维护三个异步轮询任务每个 worker 的中间状态也要单独拉取。用 hermes peer 实现planner 只需要维护三个 correlationId每个 worker 的流式消息通过回调实时更新进度最终结果通过 Future 等待代码直观很多。4. 全栈协作案例三 Agent 协作 Web 可视化4.1 系统架构与消息流我把整个案例做成一个可运行的全栈 demo后端是三个 Python Agent通过 hermes peer 点对点互通前端是一个浏览器页面通过 WebSocket 网关连接其中一个 Agent也就是 planner用来下发任务和实时展示状态。浏览器页面 --- WebSocket网关 --- planner Agent | hermes peer ------------------ | | worker Agent knowledge AgentWebSocket 网关在这里只是浅桥接层负责把浏览器的消息转成 Envelope 发给 planner再把 planner 收到的状态消息转发给浏览器。网关本身不承担任何业务逻辑。这个设计的好处是浏览器端只是整个 Agent 协作网络中的一个特殊“观察者”它不参与 Agent 间的协议协商只是通过网关接入网络。消息流的完整路径是用户在浏览器输入一个总任务例如“调研三种异步消息框架的优缺点并给出选型建议”网关将其包装成 request 消息发给 plannerplanner 收到后先请求 knowledge 查询历史材料然后拆解为多个子任务分别发给 worker-a、worker-b、worker-c每个 worker 执行时每隔几秒发送 stream 消息上报进度planner 收到后通过事件回调推送到 WebSocket 网关网关原样转发给浏览器前端渲染滚动状态最终 worker 回传 responseplanner 汇总后生成最终报告并通过网关推回前端。4.2 WebSocket 桥接与前端展示WebSocket 桥接层我直接用的 aiohttp 实现因为它是异步框架和 asyncio 天然兼容。from aiohttp import web, WSMsgType import json import asyncio async def websocket_handler(request): ws web.WebSocketResponse() await ws.prepare(request) planner request.app[planner] async def forward_to_browser(envelope): await ws.send_str(json.dumps(envelope.payload, ensure_asciiFalse)) planner.attach_observer(forward_to_browser) try: async for msg in ws: if msg.type WSMsgType.TEXT: data json.loads(msg.data) await planner.submit_task(data.get(task, )) elif msg.type WSMsgType.ERROR: break finally: planner.detach_observer(forward_to_browser) await ws.close() return ws前端页面没有用任何重型框架原生 HTML JavaScript WebSocket 就好。页面核心是一个输入框、一个任务列表、一个实时状态日志区。用户输入任务后WebSocket 会向后端发送 task 字段后端返回的所有 stream 消息都会以进度日志的形式滚动展示。前端代码里最关键的一点是 WebSocket 重连机制。Browser 端的 WebSocket 连接在 Agent 重启之后会断开如果没有自动重连用户就得手动刷新页面。我加了一个简单的指数退避重连断开后 1 秒重连失败翻倍到 2 秒、4 秒、8 秒最高不超过 30 秒。这个机制看起来简单但在实际演示中特别有用因为 Agent 调试时免不了重启。4.3 端到端演示与效果我把三个 Agent 跑在一台开发机上每个 Agent 监听不同的端口planner 监听 TCP 9100、worker-a 监听 9101、worker-b 监听 9102、worker-c 监听 9103、knowledge 监听 9104。启动时通过配置文件指定每个 Agent 的对端地址。首次启动时planner 会主动向配置好的对端发起握手连接worker 和 knowledge 也会各自启动监听服务等待接入。实际跑起来的效果是用户在浏览器输入任务大约 1 秒内 planner 开始拆分子任务然后多个 worker 的进度消息会像流水一样滚动到浏览器页面上。每个 worker 执行之间的延迟我用 sleep 控制模拟真实计算场景。最终结果从下发到展示大约 15 秒左右这完全是一个真实可复现的端到端链路。# 启动 knowledge agent python agent_knowledge.py # 启动 worker agents python agent_worker.py --port 9101 --name worker-a python agent_worker.py --port 9102 --name worker-b python agent_worker.py --port 9103 --name worker-c # 启动 planner agent web网关 python agent_planner.py启动顺序有讲究。backward 依赖的 Agent 要先启动否则 planner 启动时握手会失败但协议层不会崩溃而是在后台等待重连。这个容错设计也是 hermes peer 的重要特性握手失败是常态Agent 之间需要反复重试。5. 常见问题与排查技巧实录5.1 握手失败与版本不兼容最常遇到的问题是握手后对方迟迟不进入 active 状态。排查时先看握手返回包的 version 字段如果版本不匹配对方会在 hello-ack 中返回错误码和期望版本。这要求协议层在握手时就要把版本检查前置不能等到业务消息发过去之后才报错。其次要检查 Agent ID 是否唯一如果两个 Agent 用了同一个 ID握手后 Peer 表会发生覆盖导致消息路由错乱。我在实际开发中曾经因为复制粘贴代码导致两个 worker 的 agent_id 都是 worker-01planner 发送消息时永远只路由到其中一个另一个完全收不到消息排查了很久才发现是 ID 冲突。后来我特意写了一个启动校验每个 Agent 启动时如果发现 Agent ID 已存在于本机其他进程的监听 Dubbo 或 TCP 端口中就报错退出避免无意义的路由混乱。5.2 消息顺序错乱点对点 TCP 连接本身是保序的但 Agent 到 Agent 之间存在多路复用的时候就会出问题。比如 planner 同时向 worker-a 发送了任务 1 和任务 2worker 的两个并发线程分别处理这两个任务返回结果时可能任务 2 先完成、先回传planner 端如果按到达顺序处理就会把任务 2 的结果错配给任务 1。解决办法是靠 correlationId 做对应关系绝不依赖到达顺序。每个任务对应独立的 correlationIdresponse 返回时必须携带相同的 correlationIdplanner 收到 response 时按 correlationId 更新对应任务的 Future。我的参考实现里pending 表就是一张 correlationId 到 Future 的映射表天然解决了乱序问题。5.3 半开连接与僵尸节点半开连接是 TCP 长连接中最隐蔽的问题。一个 Agent 进程被 kill -9 直接杀掉没有机会发送 bye 消息对端 TCP 连接既不会立刻断开也不会报错看起来还是正常状态。如果没有心跳机制另一个 Agent 压根感知不到对端已经消失所有消息都会一直发送失败重试。心跳机制是我解决这个问题的核心手段。接收方如果发现某个 Peer 连续 N 个心跳周期没有响应就标记为 dead并清理本地连接资源。N 我取的是 3也就是在自适应心跳下最长约 30 秒内能感知对端离线。但这里有一个权衡心跳间隔太短会占用无谓的带宽间隔太长则僵尸节点会被保留很久。实际生产环境里我会根据网络稳定性调整心跳间隔的参数建议初始值设 30 秒网络波动大的场景降到 15 秒内网环境可以放宽到 60 秒。5.4 序列化兼容与版本演进JSON 最大的优势是跨语言通用、可读性强最大问题是 payload 格式变更时容易出现兼容性问题。我的做法是强制每个 Agent 在声明能力时把自己支持的 payload 版本也一并声明。握手时交换的不仅是协议版本还有 payload 的 schema 版本。如果 schema 版本不兼容比如 worker 新增了一个必填字段而 planner 还在用旧格式发送请求握手中的能力协商就能提前发现并提示升级。这里给出一个实操经验不要指望所有 Agent 同步升级生产环境里最好是兼容双版本。协议层在解析 payload 时对于未知字段一律忽略只解析自己认识的关键字段这就是 Forward Compatibility 的落地方式。我在 Envelope 里预留 flags 位就是为了以后做字段扩展时老版本 Agent 可以通过 flags 判断是否忽略这个字段。5.5 常见问题速查表表现可能原因处理办法握手一直不成功对端未启动、端口错、版本不兼容检查对端进程、端口地址看握手返回错误码消息发出去没有响应对端 Agent 已死或半开连接用 heartbeat 探活等超时后清理 Peer 并重连收到的结果对应错了任务没有用 correlationId 对应检查 response 是否带与 request 相同的 correlationIdAgent 重启后消息一直重试对端 Peer 表缓存旧地址监听 hello 广播事件收到新连接后更新 Peer 表同一 Agent ID 有两个进程配置里 agent_id 重复启动时做本机端口和 Agent ID 唯一性检查浏览器看不到实时进度WebSocket 桥接层没转发 stream检查 planner 是否 attach 了 observerstream 事件是否触发回调6. 后续扩展方向与经验小结6.1 从点对点走向组网目前这套实现还停留在静态配置的阶段每个 Agent 的地址和 Agent ID 通过配置文件写死。如果机器数量增多、Agent 动态上下线频繁静态配置就会变成瓶颈。我的规划是在此基础上增加一个轻量级的发现机制每个 Agent 启动时向组播地址广播自己的 Agent ID 和端口其他节点收到广播后更新本地 Peer 表并不需要专门的注册中心。组播只用于发现阶段业务消息仍然走点对点直连这样既能保持轻量又解决动态组网的问题。从工作量上看新增一个 Announcement 消息类型即可协议层不需要做大的改动。广播消息采用当前已经定义好的 event 类型payload 字段放入 Agent ID、地址、端口、能力列表即可。这一块做完之后整个 Agent 群可以实现一定程度的自组织新增节点完全免配置。6.2 加密与鉴权文章里讨论的案例是基于可信内网的所以没有涉及加密和鉴权。但如果 hermes peer 要用于跨网络环境协议层必须要加入这些能力。我的建议是先做连接层的 TLS再做消息层的签名。连接层负责防止窃听消息层负责防止伪造。握手时交换公钥证书业务消息在 payload 中携带签名接收方验签后进入业务处理。这块我还在设计中目前没有可展示的完整代码但字段预留上已经有 planflags 位的扩展空间足够容纳加密标识。6.3 个人实际体会这套协议从设计到跑通第一个全栈 demo我最大的体会是协议设计最怕一开始就是大而全像 hermes peer 这样先抓住最基本的消息类型、握手、心跳、退出这四件事把骨架立住后面再往上加东西就会很自然。反而是一上来就想把所有场景都覆盖每种消息类型都想清楚才动工的话项目大概率会卡在设计阶段出不了活。其次点对点通信的精髓在于“把连接当作一等公民”。HTTP 时代我们习惯了无状态请求但 Agent 协作是一种有状态的长时交互连接的建立、维护、断开都必须有明确的生命周期语义。hermes peer 的握手、心跳、优雅退出一整套机制本质上就是在管理连接生命周期这一步做好上层业务就能写得非常干净。最后说一个很实用的建议如果你也想自己实现一套类似的协议建议先用 Python 把协议跑通验证完消息语义和协作流的合理性再决定是否用更高性能的语言做生产实现。我在 Python 版本里加的日志和重连机制后来做任何迁移都可以作为参考基准省掉了大量排查问题的时间。