多Agent系统的通信协议：事件总线、消息格式与服务发现的标准化设计

发布时间：2026/7/17 16:17:01

多Agent系统的通信协议事件总线、消息格式与服务发现的标准化设计一、Agent间通信的根本挑战多Agent系统区别于传统微服务的关键在于Agent不仅是请求的被动响应者更是具有自主决策能力的协作节点。一个Agent可能根据任务进展主动向其他Agent发起通信、委托子任务或广播状态变更。这要求通信协议在可靠性、可追溯性和语义表达力上超越传统的REST/gRPC。二、消息格式的标准化——CloudEvents与自定义选择消息格式的选择取决于互通性需求。如果多Agent系统涉及与外部系统的集成CloudEvents是一个被广泛采纳的标准。如果只在内部Agent间通信更紧凑的自定义格式可以减少序列化开销。// CloudEvents规范的消息结构 // 符合 CNCF CloudEvents v1.0 规范 type CloudEvent struct { // 必填字段 ID string json:id // 唯一事件标识 Source string json:source // 事件来源Agent标识 SpecVersion string json:specversion // CloudEvents版本1.0 Type string json:type // 事件类型如 task.assigned // 可选字段 DataContentType string json:datacontenttype,omitempty // application/json DataSchema string json:dataschema,omitempty // JSON Schema URI Subject string json:subject,omitempty // 事件主题 Time time.Time json:time,omitempty // 事件发生时间 Data interface{} json:data,omitempty // 事件载荷 // 扩展上下文属性Agent系统特有 TaskID string json:taskid,omitempty AgentRole string json:agentrole,omitempty // planner/executor/reviewer CorrelationID string json:correlationid,omitempty // 关联ID用于追踪调用链 Priority int json:priority,omitempty // 0-1010最高 TTLSeconds int json:ttlseconds,omitempty // 消息存活时间 } // Agent间内部消息协议比CloudEvents更紧凑适合高吞吐场景 type AgentMessage struct { Header MessageHeader Payload []byte } type MessageHeader struct { Version uint8 // 协议版本 Type uint8 // 消息类型0Event, 1Request, 2Response, 3Stream Flags uint8 // bit0compressed, bit1encrypted, bit2priority MessageID uint64 // 单调递增消息ID Timestamp int64 // Unix毫秒时间戳 Source uint32 // Agent ID内部数字标识 Target uint32 // 目标Agent ID0广播 BodyLength uint32 // Payload长度 }事件总线的发布-订阅与请求-响应模式事件总线是多Agent系统的通信中枢。它需要同时支持两种通信模式发布-订阅一对多异步和请求-响应一对一同步或异步。package eventbus import ( context sync time ) // Agent通信接口 type AgentCommunicator interface { // 发布-订阅广播事件 Publish(ctx context.Context, topic string, event *CloudEvent) error Subscribe(ctx context.Context, topic string, handler EventHandler) (Subscription, error) // 请求-响应点对点通信 Request(ctx context.Context, targetAgentID string, req interface{}) (interface{}, error) RegisterHandler(method string, handler RequestHandler) error } type EventHandler func(ctx context.Context, event *CloudEvent) error type RequestHandler func(ctx context.Context, payload []byte) ([]byte, error) type Subscription interface { Unsubscribe() error Topic() string } // NATS JetStream 实现支撑高吞吐与持久化 type NATSEventBus struct { conn *nats.Conn js nats.JetStreamContext subscriptions map[string][]*nats.Subscription requestHandlers map[string]RequestHandler mu sync.RWMutex } func NewNATSEventBus(natsURL string) (*NATSEventBus, error) { nc, err : nats.Connect(natsURL, nats.ReconnectWait(2*time.Second), nats.MaxReconnects(-1), nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { // 断连告警 }), ) if err ! nil { return nil, fmt.Errorf(nats connect: %w, err) } js, err : nc.JetStream() if err ! nil { return nil, fmt.Errorf(jetstream init: %w, err) } return NATSEventBus{ conn: nc, js: js, subscriptions: make(map[string][]*nats.Subscription), requestHandlers: make(map[string]RequestHandler), }, nil } func (bus *NATSEventBus) Publish( ctx context.Context, topic string, event *CloudEvent, ) error { data, err : json.Marshal(event) if err ! nil { return fmt.Errorf(marshal event: %w, err) } // JetStream发布保证至少一次投递 _, err bus.js.Publish(topic, data, nats.MsgId(event.ID), nats.ExpectStream(agent-events), ) return err } func (bus *NATSEventBus) Subscribe( ctx context.Context, topic string, handler EventHandler, ) (Subscription, error) { sub, err : bus.js.Subscribe(topic, func(msg *nats.Msg) { var event CloudEvent if err : json.Unmarshal(msg.Data, event); err ! nil { // 无效消息确认消费避免死信堆积 msg.Ack() return } if err : handler(ctx, event); err ! nil { // 处理失败Nak触发重投 msg.Nak() return } msg.Ack() }, nats.ManualAck(), nats.Durable(fmt.Sprintf(agent-%s, topic)), ) if err ! nil { return nil, fmt.Errorf(subscribe: %w, err) } bus.mu.Lock() bus.subscriptions[topic] append(bus.subscriptions[topic], sub) bus.mu.Unlock() return natsSubscription{sub: sub, topic: topic}, nil } // 请求-响应模式带超时和重试 func (bus *NATSEventBus) Request( ctx context.Context, targetMethod string, payload []byte, ) ([]byte, error) { // 使用NATS的Request-Reply模式 msg, err : bus.conn.RequestWithContext(ctx, targetMethod, payload) if err ! nil { return nil, fmt.Errorf(request %s: %w, targetMethod, err) } return msg.Data, nil }三、服务发现与健康检查机制多Agent系统中的服务发现需要感知Agent的能力而不仅仅是IP地址。Agent注册时需要声明自己的能力清单、当前负载和状态。// Agent注册信息 type AgentRegistration struct { AgentID string json:agent_id AgentType string json:agent_type // planner / executor / reviewer Capabilities []string json:capabilities // [code_gen, code_review, test_gen] Status AgentStatus json:status // IDLE / BUSY / DRAINING Address string json:address // gRPC地址 Metadata map[string]string json:metadata // 扩展属性 MaxConcurrency int json:max_concurrency CurrentLoad int json:current_load LastHeartbeat time.Time json:last_heartbeat } type AgentStatus string const ( AgentStatusIdle AgentStatus IDLE AgentStatusBusy AgentStatus BUSY AgentStatusDraining AgentStatus DRAINING ) // 服务注册与发现接口 type ServiceRegistry interface { Register(ctx context.Context, info AgentRegistration) error Deregister(ctx context.Context, agentID string) error Discover(ctx context.Context, filters ...AgentFilter) ([]AgentRegistration, error) Heartbeat(ctx context.Context, agentID string) error } type AgentFilter func(AgentRegistration) bool func ByCapability(cap string) AgentFilter { return func(a AgentRegistration) bool { for _, c : range a.Capabilities { if c cap { return true } } return false } } func ByType(agentType string) AgentFilter { return func(a AgentRegistration) bool { return a.AgentType agentType a.Status ! AgentStatusDraining } } func ByAvailability() AgentFilter { return func(a AgentRegistration) bool { return a.Status AgentStatusIdle a.CurrentLoad a.MaxConcurrency } } // etcd实现的注册中心 type EtcdRegistry struct { client *clientv3.Client leaseID clientv3.LeaseID selfInfo AgentRegistration ttlSeconds int64 } func (r *EtcdRegistry) Heartbeat(ctx context.Context, agentID string) error { _, err : r.client.KeepAliveOnce(ctx, r.leaseID) if err ! nil { // 心跳失败重新注册 return r.Register(ctx, r.selfInfo) } return nil } func (r *EtcdRegistry) Discover( ctx context.Context, filters ...AgentFilter, ) ([]AgentRegistration, error) { resp, err : r.client.Get(ctx, /agents/, clientv3.WithPrefix(), ) if err ! nil { return nil, fmt.Errorf(etcd get agents: %w, err) } var agents []AgentRegistration for _, kv : range resp.Kvs { var agent AgentRegistration if err : json.Unmarshal(kv.Value, agent); err ! nil { continue // 跳过格式错误的数据 } // 应用过滤器 match : true for _, f : range filters { if !f(agent) { match false break } } if match { agents append(agents, agent) } } return agents, nil }四、通信的超时重试与幂等性多Agent系统中通信失败是常态而非异常。Agent可能正在处理长时间任务、正在重启或网络抖动。超时和重试策略需要配合幂等性设计。// 重试策略指数退避抖动 type RetryPolicy struct { MaxRetries int // 最大重试次数 InitialDelay time.Duration // 初始延迟 MaxDelay time.Duration // 最大延迟上限 Multiplier float64 // 退避乘数 JitterFactor float64 // 抖动因子 (0.0 ~ 1.0) } func DefaultRetryPolicy() RetryPolicy { return RetryPolicy{ MaxRetries: 3, InitialDelay: 100 * time.Millisecond, MaxDelay: 5 * time.Second, Multiplier: 2.0, JitterFactor: 0.3, } } func (p RetryPolicy) Execute( ctx context.Context, operation func(context.Context) error, isIdempotent func(error) bool, ) error { var lastErr error delay : p.InitialDelay for attempt : 0; attempt p.MaxRetries; attempt { if attempt 0 { // 检查是否应该重试 if lastErr ! nil !isIdempotent(lastErr) { return fmt.Errorf(non-idempotent error, not retrying: %w, lastErr) } // 带抖动的指数退避 jitter : time.Duration(float64(delay) * p.JitterFactor * (float64(time.Now().UnixNano()%100) / 100.0)) sleepDuration : delay jitter if sleepDuration p.MaxDelay { sleepDuration p.MaxDelay } select { case -ctx.Done(): return ctx.Err() case -time.After(sleepDuration): } delay time.Duration(float64(delay) * p.Multiplier) } lastErr operation(ctx) if lastErr nil { return nil } } return fmt.Errorf(all %d retries exhausted: %w, p.MaxRetries, lastErr) } // 幂等性保障基于消息ID的去重 type IdempotencyGuard struct { processed sync.Map // messageID → result ttl time.Duration } func (g *IdempotencyGuard) Process( messageID string, handler func() (interface{}, error), ) (interface{}, error) { // 检查是否已处理 if result, ok : g.processed.Load(messageID); ok { return result, nil // 幂等返回 } result, err : handler() if err ! nil { return nil, err } g.processed.Store(messageID, result) // TTL后自动清理防止内存泄漏 time.AfterFunc(g.ttl, func() { g.processed.Delete(messageID) }) return result, nil }幂等性的关键设计决策是去重窗口TTL的大小。窗口太小可能导致重复执行窗口太大会积累太多已处理消息的记录。合理的TTL应大于消息的最大重试时间窗口——通常设置为重试次数 × 最大延迟的2-3倍。五、总结多Agent系统的通信协议需要同时支持发布-订阅、请求-响应和流式三种模式。CloudEvents提供了标准化的消息格式与外部系统集成时有天然优势内部通信可以使用更紧凑的自定义格式。事件总线的选型上NATS的JetStream在性能、持久化和复杂度之间取得了好的平衡——它不像Kafka那样需要复杂的基础设施又比Redis Streams提供更强的持久化保证。服务发现需要从IP级别升级到能力级别——Agent注册的是能力清单和当前负载其他Agent基于能力而非地址进行发现。超时重试必须和幂等性配套设计幂等的实现成本远低于处理重复执行引发的一致性问题的成本。

相关新闻

相关新闻

最新新闻

周新闻

日新闻

月新闻