基于A2A协议的Golang多智能体协同系统实战-平芜编程栈

引言

随着人工智能技术的迅猛发展，单一智能体系统已难以应对日益复杂的现实世界任务。多智能体系统（Multi-Agent System, MAS）通过分布式智能体之间的协同与合作，展现出强大的问题解决能力，在自动驾驶、智能制造、智慧城市等领域得到广泛应用。

在多智能体系统中，智能体间的通信协议是系统设计的核心。传统的中心化通信架构存在单点故障、可扩展性差等局限性，而基于对等网络的Agent-to-Agent（A2A）协议则提供了更加灵活、鲁棒的解决方案。A2A协议允许智能体直接进行通信，无需通过中心节点转发，不仅降低了通信延迟，还提高了系统的容错能力。

本文将深入探讨基于A2A协议的Golang多智能体协同系统设计与实现。我们将构建一个完整的分布式智能体框架，涵盖：

A2A协议设计与消息格式：定义智能体间通信的标准消息格式
gRPC通信层实现：基于gRPC构建高性能的对等通信网络
智能体角色与能力管理：实现多种角色智能体（感知、规划、执行、监控）
分布式任务分配算法：基于拍卖机制的任务分配策略
协同决策与冲突解决：多智能体协商与共识达成机制
系统监控与容错处理：智能体状态监控与故障恢复

通过本文的学习，你将掌握分布式智能体系统的核心设计原则，获得可直接应用于实际项目的生产级代码，并为构建复杂的大规模智能体系统奠定坚实基础。

系统架构设计

整体架构概述

基于A2A协议的多智能体协同系统采用分层对等架构设计，摒弃了传统中心化的协调模式，赋予每个智能体更大的自主性和决策权。系统整体架构分为四个核心层次：

通信网关层（API Gateway）：提供统一的外部接口，负责请求路由和协议转换
协调器层（Coordinator）：轻量级协调节点，负责任务分发和状态监控，不参与具体决策
智能体层（Agents）：核心处理单元，包含多种角色的智能体，通过A2A协议直接通信
共享数据层（Shared Knowledge Base）：分布式存储系统，维护全局状态和共享知识

架构图展示

架构设计要点：

对等通信架构：智能体之间直接建立通信连接，减少中间环节，降低延迟
角色多样化：不同智能体承担不同职责，形成专业化分工体系
分布式决策：决策权下放到各个智能体，提高系统响应速度和鲁棒性
容错机制：智能体故障不影响整体系统运行，其他智能体可接管任务
可扩展性：支持动态添加/移除智能体，适应不同规模的应用场景

核心组件职责划分

组件	主要职责	关键技术特性
API Gateway	外部请求接入、协议转换、负载均衡	RESTful API、JWT认证、请求限流
Task Coordinator	任务分发、状态监控、资源调度	任务队列、健康检查、故障转移
Perception Agent	环境感知、数据收集、特征提取	传感器融合、实时数据处理
Planning Agent	策略制定、路径规划、方案评估	启发式算法、约束满足、多目标优化
Execution Agent	动作执行、工具调用、结果反馈	并发控制、事务管理、错误恢复
Monitoring Agent	系统监控、性能分析、异常检测	指标采集、日志聚合、告警触发
Shared Knowledge Base	全局状态存储、知识共享、历史记录	分布式存储、数据一致性、查询优化

核心模块实现

1. A2A协议消息格式定义

A2A协议是智能体间通信的基础，我们定义了一套完整的消息格式标准，支持多种类型的交互场景。

// protocol/a2a_messages.go package protocol import ( "encoding/json" "time" ) // MessageType 定义消息类型枚举 type MessageType int const ( TypeTaskAnnouncement MessageType = iota + 1 // 任务公告 TypeBidSubmission // 投标提交 TypeTaskAssignment // 任务分配 TypeTaskCompletion // 任务完成 TypeStatusUpdate // 状态更新 TypeEmergencyAlert // 紧急警报 TypeNegotiationRequest // 协商请求 TypeNegotiationResponse // 协商响应 ) // A2AMessage A2A协议基础消息结构 type A2AMessage struct { ID string `json:"id"` // 消息唯一标识 Type MessageType `json:"type"` // 消息类型 SenderID string `json:"sender_id"` // 发送方ID ReceiverID string `json:"receiver_id"` // 接收方ID（空表示广播） Timestamp time.Time `json:"timestamp"` // 时间戳 Payload interface{} `json:"payload"` // 消息负载 Signature string `json:"signature"` // 数字签名（可选） TTL int `json:"ttl"` // 生存时间（秒） } // TaskAnnouncement 任务公告消息负载 type TaskAnnouncement struct { TaskID string `json:"task_id"` Description string `json:"description"` Priority int `json:"priority"` // 1-10, 10最高 Deadline time.Time `json:"deadline"` Requirements map[string]interface{} `json:"requirements"` Reward float64 `json:"reward"` // 任务奖励（虚拟货币） } // BidSubmission 投标提交消息负载 type BidSubmission struct { TaskID string `json:"task_id"` BidderID string `json:"bidder_id"` Capability []string `json:"capability"` // 投标者能力列表 EstimatedCost float64 `json:"estimated_cost"` // 预估成本 EstimatedTime float64 `json:"estimated_time"` // 预估时间（秒） Reputation float64 `json:"reputation"` // 投标者信誉评分 } // TaskAssignment 任务分配消息负载 type TaskAssignment struct { TaskID string `json:"task_id"` AssigneeID string `json:"assignee_id"` CoordinatorID string `json:"coordinator_id"` AssignmentTime time.Time `json:"assignment_time"` Constraints map[string]interface{} `json:"constraints"` } // NegotiationProposal 协商提案 type NegotiationProposal struct { NegotiationID string `json:"negotiation_id"` ProposerID string `json:"proposer_id"` Proposal map[string]interface{} `json:"proposal"` Utility float64 `json:"utility"` // 提案效用值 Deadline time.Time `json:"deadline"` // 提案有效期 } // Marshal 序列化消息为JSON func (msg *A2AMessage) Marshal() ([]byte, error) { return json.Marshal(msg) } // Unmarshal 从JSON反序列化消息 func Unmarshal(data []byte) (*A2AMessage, error) { var msg A2AMessage if err := json.Unmarshal(data, &msg); err != nil { return nil, err } return &msg, nil } // Validate 验证消息有效性 func (msg *A2AMessage) Validate() bool { if msg.ID == "" || msg.SenderID == "" || msg.Timestamp.IsZero() { return false } // 检查TTL if msg.TTL > 0 { expireTime := msg.Timestamp.Add(time.Duration(msg.TTL) * time.Second) if time.Now().After(expireTime) { return false } } return true }

2. 智能体基础结构

智能体是所有功能的核心载体，我们定义了一个可扩展的基础智能体结构。

// agent/base_agent.go package agent import ( "context" "fmt" "log" "sync" "time" "github.com/yourusername/multi-agent/protocol" ) // AgentRole 定义智能体角色 type AgentRole string const ( RolePerception AgentRole = "perception" RolePlanning AgentRole = "planning" RoleExecution AgentRole = "execution" RoleMonitoring AgentRole = "monitoring" RoleGeneral AgentRole = "general" // 通用角色 ) // Capability 定义智能体能力 type Capability struct { ID string `json:"id"` Name string `json:"name"` Description string `json:"description"` Parameters map[string]interface{} `json:"parameters"` CostModel func(params map[string]interface{}) float64 `json:"-"` } // AgentStatus 定义智能体状态 type AgentStatus string const ( StatusIdle AgentStatus = "idle" StatusBusy AgentStatus = "busy" StatusProcessing AgentStatus = "processing" StatusFaulty AgentStatus = "faulty" ) // BaseAgent 智能体基础结构 type BaseAgent struct { ID string Name string Role AgentRole Status AgentStatus Capabilities []Capability Reputation float64 // 信誉评分（0-1） // 通信相关 IncomingChan chan *protocol.A2AMessage OutgoingChan chan *protocol.A2AMessage PeerAgents map[string]string // agentID -> address // 状态管理 currentTasks map[string]*TaskContext statusLock sync.RWMutex stopChan chan struct{} // 性能指标 metrics *AgentMetrics } // AgentMetrics 智能体性能指标 type AgentMetrics struct { TasksCompleted int64 TasksFailed int64 TotalProcessingTime time.Duration AvgResponseTime time.Duration MessageSent int64 MessageReceived int64 } // TaskContext 任务上下文 type TaskContext struct { TaskID string Description string StartTime time.Time Deadline time.Time Status string Progress float64 // 0-1 Result interface{} } // NewBaseAgent 创建新的基础智能体 func NewBaseAgent(id, name string, role AgentRole) *BaseAgent { return &BaseAgent{ ID: id, Name: name, Role: role, Status: StatusIdle, Capabilities: make([]Capability, 0), Reputation: 0.8, // 初始信誉评分 IncomingChan: make(chan *protocol.A2AMessage, 100), OutgoingChan: make(chan *protocol.A2AMessage, 100), PeerAgents: make(map[string]string), currentTasks: make(map[string]*TaskContext), stopChan: make(chan struct{}), metrics: &AgentMetrics{}, } } // Start 启动智能体 func (a *BaseAgent) Start(ctx context.Context) error { a.statusLock.Lock() if a.Status == StatusBusy || a.Status == StatusProcessing { a.statusLock.Unlock() return fmt.Errorf("agent already running") } a.Status = StatusIdle a.statusLock.Unlock() log.Printf("Agent %s (%s) starting...", a.ID, a.Role) // 启动消息处理循环 go a.messageLoop(ctx) // 启动状态维护循环 go a.statusLoop(ctx) return nil } // Stop 停止智能体 func (a *BaseAgent) Stop(ctx context.Context) error { a.statusLock.Lock() a.Status = StatusFaulty a.statusLock.Unlock() close(a.stopChan) log.Printf("Agent %s stopped", a.ID) return nil } // messageLoop 消息处理循环 func (a *BaseAgent) messageLoop(ctx context.Context) { for { select { case <-ctx.Done(): return case <-a.stopChan: return case msg := <-a.IncomingChan: a.handleMessage(ctx, msg) } } } // statusLoop 状态维护循环 func (a *BaseAgent) statusLoop(ctx context.Context) { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-a.stopChan: return case <-ticker.C: a.updateStatus() } } } // handleMessage 处理接收到的消息 func (a *BaseAgent) handleMessage(ctx context.Context, msg *protocol.A2AMessage) { a.metrics.MessageReceived++ if !msg.Validate() { log.Printf("Agent %s received invalid message: %v", a.ID, msg.ID) return } switch msg.Type { case protocol.TypeTaskAnnouncement: a.handleTaskAnnouncement(ctx, msg) case protocol.TypeTaskAssignment: a.handleTaskAssignment(ctx, msg) case protocol.TypeNegotiationRequest: a.handleNegotiationRequest(ctx, msg) case protocol.TypeEmergencyAlert: a.handleEmergencyAlert(ctx, msg) default: log.Printf("Agent %s received unsupported message type: %v", a.ID, msg.Type) } } // AddCapability 添加能力 func (a *BaseAgent) AddCapability(capability Capability) { a.statusLock.Lock() defer a.statusLock.Unlock() a.Capabilities = append(a.Capabilities, capability) } // UpdateReputation 更新信誉评分 func (a