多Agent系统的通信协议：事件总线、消息格式与服务发现的标准化设计
·
多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-10，10最高
TTLSeconds int `json:"ttlseconds,omitempty"` // 消息存活时间
}
// Agent间内部消息协议（比CloudEvents更紧凑，适合高吞吐场景）
type AgentMessage struct {
Header MessageHeader
Payload []byte
}
type MessageHeader struct {
Version uint8 // 协议版本
Type uint8 // 消息类型（0=Event, 1=Request, 2=Response, 3=Stream）
Flags uint8 // bit0=compressed, bit1=encrypted, bit2=priority
MessageID uint64 // 单调递增消息ID
Timestamp int64 // Unix毫秒时间戳
Source uint32 // Agent ID（内部数字标识）
Target uint32 // 目标Agent ID（0=广播）
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基于能力而非地址进行发现。超时重试必须和幂等性配套设计，幂等的实现成本远低于处理重复执行引发的一致性问题的成本。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
7
0- 0
已为社区贡献18条内容
所有评论(0)