单Agent的天花板一个Agent能做的事,本质受限于三件事:上下文窗口: 不管多大的模型,塞满100万字也开始失忆技能边界: 一个Agent精通SQL,但不一定懂合同审核执行时间: 单线程执行,复杂任务跑到一半容易因token超限或prompt变长而劣化解决方案看起来很简单——多Agent协作.但当你真开始组织10个Agent一起干活时,马上撞到下一个墙:Agent之间怎么通信.REST API?太重.共享数据库?太慢.消息队列?运维复杂.每一个方案都有它的工程包袱.5层通信架构的设计哲学设计来自开源仓库 agent-cluster-comm,核心思想是像TCP/IP那样分层,每层只管一件事:Layer 5: 语义层 (Semantic) — Agent说什么意思 Layer 4: 协作层 (Coordination) — Agent怎么分工 Layer 3: 路由层 (Routing) — 消息发给谁 Layer 2: 会话层 (Session) — 谁在跟谁说话 Layer 1: 传输层 (Transport) — 怎么把字节传过去每一层只依赖下一层,上层不感知下层的实现细节.这种分层的好处是:换传输层(比如从HTTP换成消息队列),上层Agent代码完全不动.Layer 1: 传输层 — 字节级搬运工最底层只做一件事:把一个Agent发出的JSON消息可靠地送到另一个Agent.三种实现可选:# 实现A: HTTP直连(默认) class HTTPTransport: def send(self, target_url, message): response requests.post(target_url, jsonmessage, timeout30) return response.json() # 实现B: Redis消息队列(适合异步) class RedisTransport: def __init__(self, redis_client): self.redis redis_client def send(self, channel, message): self.redis.lpush(fagent:inbox:{channel}, json.dumps(message)) # 实现C: 文件系统(适合本地调试) class FileTransport: def send(self, target_dir, message): msg_id message[id] with open(f{target_dir}/{msg_id}.json, w) as f: json.dump(message, f)生产环境用HTTP或Redis,本地调试用文件系统——上层Agent代码完全一样,只是构造时注入不同的Transport.Layer 2: 会话层 — 维护对话身份两个Agent之间的多轮对话需要session.这一层负责:生成全局唯一的session_id跟踪参与方 (participants)记录对话历史 (history)管理超时与重连class Session: def __init__(self, session_id, participants): self.session_id session_id self.participants participants # [agent_a_id, agent_b_id] self.history [] self.created_at time.time() self.last_active time.time() def append(self, message): self.history.append(message) self.last_active time.time() def is_expired(self, ttl3600): return time.time() - self.last_active ttl会话层的存在让Agent不用关心我和谁在说话——它只管处理收到的消息,session管理全自动.Layer 3: 路由层 — 消息该发给谁当集群里有10个Agent时,某个Agent发出的消息应发给谁?三个策略:class Router: def route(self, message, known_agents): # 策略A: 直接寻址(消息头带 toagent_id) if to in message[header]: return [message[header][to]] # 策略B: 广播给所有Agent if message[header].get(broadcast): return list(known_agents.keys()) # 策略C: 基于能力的路由(根据消息类型找对应的Agent) msg_type message[header][type] return [ agent_id for agent_id, agent in known_agents.items() if msg_type in agent.capabilities ]基于能力的路由是最有意思的:消息上标我需要一个会SQL的Agent,路由层自动把请求发给所有声明了SQL能力的Agent.这样新加入的Agent只要正确声明自己的能力,就能无缝接入集群.Layer 4: 协作层 — Agent怎么分工多Agent协作的核心模式有三种,这一层封装了它们的实现:模式一: Pipeline(流水线)甲→乙→丙,前一个的输出是后一个的输入.适合可分解的串行任务:def run_pipeline(agents, input_data): data input_data for agent in agents: data agent.process(data) return data模式二: Fan-out/Fan-in(扇出扇入)一个主Agent把任务拆成N份,发给N个子Agent并行处理,最后聚合结果:def run_fanout_fanin(master, workers, task): subtasks master.split(task) futures [worker.process(st) for worker, st in zip(workers, subtasks)] results [f.result() for f in futures] return master.aggregate(results)适合可以并行的任务,比如批量评估100个客户.模式三: Debate(辩论式)多个Agent对同一问题给出不同意见,由仲裁Agent综合判断:def run_debate(proposers, judge, question): opinions [p.answer(question) for p in proposers] return judge.synthesize(opinions)适合需要多视角的决策,比如风控审批.Layer 5: 语义层 — 让Agent相互听懂最高层关心消息的内容到底是什么意思.即使两个Agent都懂JSON,也可能因为对字段含义理解不同而出错.语义层定义了一个消息契约:MESSAGE_SCHEMA { header: { id: uuid, # 消息唯一ID type: task_request, # 消息类型(枚举) from: agent_id, # 发送方 to: agent_id or null, # 接收方,null表示广播 session_id: uuid, # 会话ID capabilities: [sql], # 需要的能力 timestamp: ISO8601 }, body: { content: any, # 主体内容 context: {}, # 上下文/引用 expected_response: json_schema # 期望的响应格式 }, envelope: { schema_version: 1.0, compression: none, encryption: none } }expected_response字段是关键:发送方明确告诉接收方我要这种格式的回答,接收方按schema构造响应.这样Agent之间就不需要互相磨合prompt了.5层架构的实战价值对比一下裸调LLM写多Agent和5层架构的差异:维度裸LLM多Agent5层架构通信可靠性靠prompt保证Layer 1保证会话管理各Agent自己存Layer 2统一寻址路由硬编码Layer 3可策略切换协作模式每次重写Layer 4三种模式复用消息契约无,容易出bugLayer 5强制schemaAgent替换牵一发动全身替换单个Agent不影响其他关键工程决策决策点选择理由分层架构5层而非3层协作和语义分开,职责更清晰传输层接口抽象,多实现生产用Redis,调试用文件路由策略三种可切换不同场景用不同路由消息格式强schema避免Agent互相猜字段含义会话管理TTL自动失效避免僵尸会话占内存适用场景与边界适合5层架构的场景:企业级Agent集群: 10个Agent需要长期稳定协作跨域Agent集成: 比如银行风控Agent电信运维Agent联合分析可观测性要求高: 需要每条消息可追溯、每层可独立监控不适合的场景:单Agent能搞定的事: 杀鸡焉用牛刀2-3个Agent的临时协作: 直接函数调用更快对延迟极敏感的实时系统: 分层带来额外开销总结5层通信架构的核心价值是让Agent之间的协作变成工程问题,而不是prompt艺术问题.每一层职责清晰、可独立替换、可单独监控.上层Agent完全不关心底层用的是HTTP还是消息队列,只关心我要发什么消息、消息格式是什么、希望谁来处理.这种解耦让Agent集群具备真正的可演进性——加新Agent、换传输协议、改协作模式,都不会让现有系统推倒重来.完整实现见 agent-cluster-comm 仓库,包含5层完整代码、3种传输实现、3种协作模式示例和监控接口定义.有问题欢迎提 Issue.