单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, json=message, timeout=30)
return response.json()
# 实现B: Redis消息队列(适合异步)
class RedisTransport:
def __init__(self, redis_client):
self.redis = redis_client
def send(self, channel, message):
self.redis.lpush(f"agent: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, ttl=3600):
return time.time() - self.last_active > ttl
会话层的存在让Agent不用关心"我和谁在说话"——它只管处理收到的消息,session管理全自动.
Layer 3: 路由层 — 消息该发给谁
当集群里有10个Agent时,某个Agent发出的消息应发给谁?三个策略:
class Router:
def route(self, message, known_agents):
# 策略A: 直接寻址(消息头带 to=agent_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多Agent | 5层架构 |
|---|---|---|
| 通信可靠性 | 靠prompt保证 | Layer 1保证 |
| 会话管理 | 各Agent自己存 | Layer 2统一 |
| 寻址路由 | 硬编码 | Layer 3可策略切换 |
| 协作模式 | 每次重写 | Layer 4三种模式复用 |
| 消息契约 | 无,容易出bug | Layer 5强制schema |
| Agent替换 | 牵一发动全身 | 替换单个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.
