多 Agent 通信协议选型：gRPC、消息队列还是共享存储

一、Agent 之间的信息传递到底有多复杂

单个 Agent 只是调用 LLM + 执行工具，架构上和一个微服务差不多。但一旦你有了两个以上需要协作的 Agent，通信就成了真正的难题。一个 Agent 的输出是另一个 Agent 的输入，中间需要序列化、传递、反序列化、校验。更麻烦的是——Agent 的推理时间不确定，你可能需要异步解耦 + 背压控制。

最简单的实现：把所有 Agent 放在同一个进程里，函数调用完事。但这丧失了独立扩缩容的能力，也不利于故障隔离。

flowchart TD A[Agent A - 任务规划] --> B{通信模式选择} B -->|同步| C[gRPC 直连] B -->|异步| D[消息队列] B -->|松耦合| E[共享存储] C --> F[低延迟 < 10ms<br/>强类型<br/>双向流] D --> G[削峰填谷<br/>背压天然支持<br/>事件溯源] E --> H[最大解耦<br/>数据可复盘<br/>延迟最高] F --> I[适用: Agent链式协作] G --> J[适用: 异步任务分发] H --> K[适用: 人机协作审阅]

二、三种通信模式的深度对比

gRPC（同步直连）

优势是延迟低（同集群 < 5ms）、类型安全（protobuf）、支持双向流。问题在于同步耦合：A 调用 B 时如果 B 的 LLM 推理耗时 30 秒，A 的连接也会被占用 30 秒。这在高并发下是连接池杀手。

消息队列（异步解耦）

RabbitMQ/Kafka/NATS 这类 MQ 天然支持削峰。Agent A 把任务丢到队列就完事，Agent B 消费并处理。好处是背压天然解决——队列满了生产者自然阻塞。坏处是延迟增加（端到端至少 10-50ms），且丢消息需要额外保障（ACK + 死信队列）。

共享存储（最大解耦）

Agent A 把结果写入 Redis/PostgreSQL/MinIO，Agent B 轮询或订阅变更。这是最松耦合的方案，每个 Agent 可以完全独立部署。代价是延迟最高（轮询间隔 + 网络 IO），且需要处理并发读写冲突。

三、混合模式的生产实现

package agentbus import ( "context" "encoding/json" "fmt" "sync" "time" "github.com/nats-io/nats.go" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" ) // Message 统一消息体。 // 关键设计：mediatype 字段用于 Agent 判断消息类型后做反序列化。 // 不同 Agent 可能需要不同的消息结构体，通过 mediatype 做分发。 type Message struct { ID string `json:"id"` From string `json:"from"` To string `json:"to"` MediaType string `json:"mediatype"` Payload json.RawMessage `json:"payload"` Timestamp time.Time `json:"timestamp"` Retry int `json:"retry"` } // AgentBus 多 Agent 通信总线。 // 设计决策： // - 同步调用走 gRPC 通道（低延迟场景） // - 异步调用走 NATS 通道（解耦场景） // - 两种通道共享统一 Message 格式，避免序列化分裂 type AgentBus struct { grpcConns map[string]*grpc.ClientConn natsConn *nats.Conn mu sync.RWMutex // 超时控制：gRPC 调用上限 30s，超过即认为 Agent 不可用 grpcTimeout time.Duration } func NewAgentBus(natsURL string) (*AgentBus, error) { nc, err := nats.Connect(natsURL, nats.ReconnectWait(2*time.Second), nats.MaxReconnects(10), nats.Timeout(5*time.Second), ) if err != nil { return nil, fmt.Errorf("nats connect: %w", err) } return &AgentBus{ grpcConns: make(map[string]*grpc.ClientConn), natsConn: nc, grpcTimeout: 30 * time.Second, }, nil } // SendSync 同步发送消息到目标 Agent，返回响应。 // 使用 gRPC 是因为需要即时响应——典型场景是 Agent 协作链。 // 超时设置为 30s 是因为 LLM 推理存在长尾延迟。 func (b *AgentBus) SendSync(ctx context.Context, msg *Message) (*Message, error) { ctx, cancel := context.WithTimeout(ctx, b.grpcTimeout) defer cancel() conn, err := b.getOrCreateGRPCConn(msg.To) if err != nil { return nil, fmt.Errorf("grpc dial: %w", err) } // 实际的 gRPC 调用需要对应的 stub _ = conn // 这里用占位说明：调用 agentpb.AgentService/Process(ctx, req) return nil, fmt.Errorf("stub not implemented: connect to %s", msg.To) } // SendAsync 异步发送，走 NATS。 // 发布到 subject = agent.{To}.inbox，由目标 Agent 消费。 // 返回 ack 确认消息已被 NATS server 接收，但不代表目标 Agent 已处理。 func (b *AgentBus) SendAsync(msg *Message) error { data, err := json.Marshal(msg) if err != nil { return fmt.Errorf("marshal: %w", err) } subject := fmt.Sprintf("agent.%s.inbox", msg.To) // Publish 是 fire-and-forget，不等待 consumer ack // 如果需要确认，应该用 Request 或 JetStream return b.natsConn.Publish(subject, data) } // Subscribe 订阅异步消息。 // 使用 JetStream 持久化订阅，保证重启后重放未处理消息。 func (b *AgentBus) Subscribe(agentName string, handler func(*Message) error) error { subject := fmt.Sprintf("agent.%s.inbox", agentName) _, err := b.natsConn.Subscribe(subject, func(m *nats.Msg) { var msg Message if err := json.Unmarshal(m.Data, &msg); err != nil { // 反序列化失败，确认消费避免死循环 // 同时在 dead-letter 通道记录异常 _ = b.natsConn.Publish("agent.deadletter", m.Data) return } if err := handler(&msg); err != nil { // handler 失败，根据重试次数决定是否重新入队 if msg.Retry < 3 { msg.Retry++ _ = b.SendAsync(&msg) } else { _ = b.natsConn.Publish("agent.deadletter", m.Data) } } }) return err } func (b *AgentBus) getOrCreateGRPCConn(target string) (*grpc.ClientConn, error) { b.mu.Lock() defer b.mu.Unlock() if conn, ok := b.grpcConns[target]; ok { return conn, nil } conn, err := grpc.Dial(target, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`), grpc.WithKeepaliveParams(/* ... */), ) if err != nil { return nil, err } b.grpcConns[target] = conn return conn, nil } func (b *AgentBus) Close() error { for _, conn := range b.grpcConns { conn.Close() } b.natsConn.Close() return nil }

四、选型决策矩阵

核心取舍

• gRPC：低延迟但耦合，单点故障影响链路
• 消息队列：削峰但运维复杂（需要额外维护 MQ 集群）
• 共享存储：最解耦但延迟最高

五、总结

不存在万能协议。关键是根据延迟敏感度 × 耦合容忍度做决策。我的建议：主链路用 gRPC 保证低延迟，异步分发用 NATS 解耦，状态同步走 Redis。三个通道用统一Message格式串联——这样换通道时不需要改消息结构。