从零搭建多Agent系统：基于A2A协议的天气与行程规划AI协作案例详解
·
从零搭建多Agent系统：基于A2A协议的天气与行程规划AI协作案例详解
最近在折腾一个智能旅行规划的小项目时，我遇到了一个典型问题：一个AI助手很难同时精通天气查询、地点推荐、路线规划和预算管理。这让我开始思考，与其费尽心思去训练一个“全能”但可能“全不能”的超级AI，不如让几个各有所长的AI助手协作起来。这就像组建一个团队，让擅长数据分析的同事负责处理数据，让沟通能力强的同事负责对接客户，效率反而更高。多Agent系统（Multi-Agent System）正是为了解决这类问题而生，而让这些AI助手能顺畅“对话”和“协作”的关键，就是一套标准化的通信协议。今天，我就以“天气查询”和“行程规划”这两个最贴近生活的场景为例，带你从零开始，手把手搭建一个基于A2A（Agent-to-Agent）协议的多Agent协作系统。无论你是对AI应用开发感兴趣的工程师，还是想了解前沿技术如何落地的技术爱好者，这篇文章都将为你提供一个清晰、可操作的实践路径。
1. 理解多Agent协作的核心：A2A协议
在深入代码之前，我们必须先搞清楚多Agent系统运作的基石——A2A协议。你可以把它想象成人类团队协作中的“会议纪要”或“工作流程规范”。如果没有统一的沟通语言和协作规则，每个成员（Agent）各说各话，项目必然陷入混乱。
A2A协议的核心目标，就是为不同的AI智能体（Agent）定义一套标准化的“对话”方式。这不仅仅是简单的消息传递，更包括了服务发现、任务编排、状态同步和安全控制等一系列复杂交互的约定。与MCP（Model Context Protocol）协议主要解决AI与外部工具（如数据库、API）的连接问题不同，A2A协议聚焦于AI与AI之间的“社交网络”构建。
注意：A2A协议目前仍处于快速发展阶段，由社区和部分大型科技公司推动。它不是一个像HTTP那样有唯一官方标准的协议，而更像是一套设计理念和最佳实践的集合。我们在实现时，需要把握其核心思想，而非拘泥于某个特定实现。
一个健壮的A2A协议实现，通常包含以下几个关键组件：
- 统一的消息信封（Envelope）：所有Agent间传递的消息都必须包裹在一个标准格式里。这确保了无论消息内容如何，接收方都能正确解析出发送者、消息类型、时间戳等元信息。
- 能力注册与发现机制：每个Agent启动时，需要向一个“服务注册中心”宣告自己具备哪些能力（例如，“我能查询未来三天的天气”）。其他Agent可以通过查询这个中心，找到能帮助自己完成任务的合作伙伴。
- 任务分解与委派逻辑：当一个复杂任务（如“为我规划一次周末露营”）到来时，负责协调的Agent（或称“Orchestrator”）需要能将其分解为子任务（查询天气、推荐露营地、规划路线），并委派给相应的专业Agent。
- 对话上下文管理：多个Agent围绕一个任务的多次交互，需要共享同一个上下文（Conversation Context），以确保它们讨论的是同一件事，避免信息错乱。
为了更直观地理解A2A与MCP的分工，我们可以看下面这个对比表格：
| 特性维度 | A2A (Agent to Agent) | MCP (Model Context Protocol) |
|---|---|---|
| 核心目的 | 实现多个AI智能体之间的协作与对话。 | 实现AI智能体与外部工具、数据源的安全、标准化连接。 |
| 交互关系 | 水平协作，Agent之间是平等的或具有层级关系的伙伴。 | 垂直调用，Agent是“使用者”，工具是“被调用者”。 |
| 典型场景 | 任务分解、结果汇总、决策协商、接力处理。 | 执行单一、具体的操作，如调用API、查询数据库、运行代码。 |
| 协议焦点 | 服务发现、会话管理、任务流编排、状态同步。 | 工具描述、参数验证、执行调用、结果返回。 |
| 类比 | 公司内部不同部门团队的协作会议。 | 员工使用公司统一的软件系统（如ERP、CRM）处理业务。 |
在我们的天气与行程规划案例中，WeatherAgent（天气助手）和TripAgent（行程助手）就是通过A2A协议进行协作的两个独立智能体。用户向TripAgent提出需求，TripAgent发现自己需要天气信息，于是通过A2A协议找到并询问WeatherAgent，最后整合信息回复用户。这个过程完美诠释了A2A的价值。
2. 系统架构设计与技术选型
明确了协议的核心思想后，我们来设计系统的具体架构。一个可扩展、易维护的多Agent系统，不能只是两个脚本的简单连接。我们需要考虑服务化、通信方式、状态管理等工程问题。
我设计的架构如下图所示（概念图），它采用了清晰的分层和模块化思想：
[用户界面/API网关] <- HTTP/WebSocket ->
|
v
[协调层 Agent Orchestrator]
|
v
[A2A 通信层]
/ \
/ \
v v
[WeatherAgent] [TripAgent]
| |
v v
[MCP工具] [MCP工具]
(天气API) (地图/日历API)
各层职责解析：
- 协调层（Orchestrator）：这是系统的大脑。它接收用户初始请求，分析任务意图，并根据注册中心的信息，将任务分解并路由给合适的Agent。在我们的简单案例中，TripAgent暂时兼任了协调者的角色，但在复杂系统中，一个独立的Orchestrator是必要的。
- A2A通信层：这是协议的具体实现层。我们选择使用gRPC作为通信框架。相比于原始的HTTP/REST，gRPC基于HTTP/2，支持双向流、多路复用，性能更高，并且能通过Protocol Buffers（protobuf）自动生成强类型的客户端和服务端代码，完美契合A2A对结构化消息和高性能通信的需求。
- 业务Agent层：即WeatherAgent和TripAgent。它们是独立的微服务，专注于自己的领域。每个Agent内部可以集成MCP客户端来调用所需的外部工具（如WeatherAgent通过MCP调用天气API）。
- 服务注册与发现：我们引入Consul或etcd作为服务注册中心。每个Agent启动时，将自己的网络地址和能力描述注册上去。Orchestrator或其他Agent需要协作时，先查询注册中心获取目标Agent的地址。
技术栈清单：
- 编程语言：Python 3.10+。生态丰富，AI相关库支持好。
- 通信框架：gRPC + Protocol Buffers。用于实现高效的A2A通信。
- 服务框架：FastAPI。用于对外提供HTTP API（用户界面或网关调用），内部轻量高效。
- 服务发现：Consul。轻量级，与微服务架构集成简单。
- Agent核心：LangChain或LlamaIndex。提供构建Agent所需的基础抽象和工具集成能力，但不是必须，我们也可以从零构建以加深理解。
- 部署：Docker + Docker Compose。保证环境一致性，简化部署流程。
这个架构的优势在于解耦清晰。每个Agent可以独立开发、测试、部署和扩展。通信协议标准化后，未来新增一个“BudgetAgent”（预算助手）或“HotelAgent”（酒店助手）会非常容易。
3. 实战：定义A2A协议与实现基础服务
理论说得再多，不如动手写一行代码。让我们从定义最核心的A2A通信协议开始。
首先，我们使用Protocol Buffers来定义Agent之间“说话”的语言。创建一个名为 a2a.proto 的文件：
syntax = "proto3";
package a2a;
// 定义Agent的能力类型
enum Capability {
WEATHER_QUERY = 0;
TRIP_PLANNING = 1;
LOCATION_SEARCH = 2;
// ... 可以扩展更多能力
}
// Agent注册时上报的信息
message AgentInfo {
string agent_id = 1; // 唯一标识
string name = 2; // 名称，如"WeatherAgent"
string endpoint = 3; // gRPC服务地址
repeated Capability capabilities = 4; // 具备的能力列表
}
// 通用的A2A请求消息
message A2ARequest {
string request_id = 1; // 请求唯一ID，用于追踪
string sender_id = 2; // 发送方Agent ID
string conversation_id = 3; // 会话ID，关联同一任务的所有交互
string action = 4; // 请求的动作，如"get_weather"
string payload = 5; // JSON格式的请求参数
}
// 通用的A2A响应消息
message A2AResponse {
string request_id = 1; // 对应请求的ID
string sender_id = 2; // 响应方Agent ID
bool success = 3; // 成功与否
string data = 4; // JSON格式的响应数据
string error_message = 5; // 失败时的错误信息
}
// 服务发现相关的消息
message DiscoveryRequest {
Capability required_capability = 1; // 需要查找的能力
}
message DiscoveryResponse {
repeated AgentInfo agents = 1; // 符合条件的Agent列表
}
// 定义A2A通信服务
service A2AService {
// 处理来自其他Agent的请求
rpc HandleRequest(A2ARequest) returns (A2AResponse);
// 供Orchestrator或Agent发现服务
rpc DiscoverAgents(DiscoveryRequest) returns (DiscoveryResponse);
}
使用protobuf编译器生成Python代码：
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. a2a.proto
接下来，我们实现一个基础的Agent基类。这个基类封装了gRPC服务器的启动、向Consul注册、以及处理请求的框架。
# base_agent.py
import grpc
from concurrent import futures
import logging
import consul
from abc import ABC, abstractmethod
import a2a_pb2
import a2a_pb2_grpc
class BaseAgent(a2a_pb2_grpc.A2AServiceServicer, ABC):
def __init__(self, agent_id, name, capabilities, host='0.0.0.0', port=50051):
self.agent_id = agent_id
self.name = name
self.capabilities = capabilities
self.host = host
self.port = port
self.endpoint = f"{host}:{port}"
self.consul_client = consul.Consul() # 假设Consul运行在默认地址
# 初始化gRPC服务器
self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
a2a_pb2_grpc.add_A2AServiceServicer_to_server(self, self.server)
def register_to_consul(self):
"""向Consul注册服务"""
service_id = f"{self.name}-{self.agent_id}"
check = consul.Check.grpc_check(f"{self.endpoint}", interval="10s")
self.consul_client.agent.service.register(
name=self.name,
service_id=service_id,
address=self.host,
port=self.port,
tags=[cap.name for cap in self.capabilities],
check=check
)
logging.info(f"Agent {self.name} registered to Consul.")
def start(self):
"""启动gRPC服务器并注册服务"""
self.server.add_insecure_port(self.endpoint)
self.server.start()
self.register_to_consul()
logging.info(f"{self.name} started on {self.endpoint}")
self.server.wait_for_termination()
# 实现A2A服务接口
def HandleRequest(self, request, context):
"""处理来自其他Agent的请求，子类需实现具体的路由逻辑"""
logging.info(f"Received request: {request.action} from {request.sender_id}")
# 这里可以添加认证、限流等中间件逻辑
return self._route_request(request)
@abstractmethod
def _route_request(self, request: a2a_pb2.A2ARequest) -> a2a_pb2.A2AResponse:
"""子类必须实现：根据request.action路由到具体的处理方法"""
pass
def DiscoverAgents(self, request, context):
"""供其他服务查询具备特定能力的Agent"""
# 这里简化实现，直接查询Consul。实际可能需更复杂的过滤逻辑。
_, services = self.consul_client.agent.services()
matched_agents = []
for svc_id, svc_info in services.items():
tags = svc_info.get('Tags', [])
if request.required_capability.name in tags:
agent_info = a2a_pb2.AgentInfo(
agent_id=svc_id,
name=svc_info['Service'],
endpoint=f"{svc_info['Address']}:{svc_info['Port']}",
capabilities=[a2a_pb2.Capability.Value(tag) for tag in tags if tag in a2a_pb2.Capability._member_names_]
)
matched_agents.append(agent_info)
return a2a_pb2.DiscoveryResponse(agents=matched_agents)
这个基类完成了大量脏活累活：gRPC服务框架、服务注册、请求接收。现在，我们可以基于它快速构建具体的业务Agent。
4. 构建WeatherAgent与TripAgent
有了坚实的基础设施，构建业务Agent就变得非常专注——只需要关心自己领域的逻辑。
首先，是WeatherAgent。 它的核心是提供一个get_weather的A2A动作。为了获取真实天气，我们通过一个模拟的MCP工具（或直接调用天气API）来实现。
# weather_agent.py
import json
from datetime import datetime
import a2a_pb2
from base_agent import BaseAgent
class WeatherAgent(BaseAgent):
def __init__(self):
# 定义自己的能力是 WEATHER_QUERY
capabilities = [a2a_pb2.Capability.WEATHER_QUERY]
super().__init__(
agent_id="weather_001",
name="WeatherAgent",
capabilities=capabilities,
port=50052 # 使用不同端口
)
# 模拟一个简单的天气数据源，真实场景替换为MCP工具调用
self.weather_db = {
"2024-07-15": {"temperature": 28, "condition": "Sunny", "humidity": 65},
"2024-07-16": {"temperature": 22, "condition": "Rainy", "humidity": 90},
"2024-07-17": {"temperature": 25, "condition": "Cloudy", "humidity": 75},
}
def _route_request(self, request):
if request.action == "get_weather":
return self._handle_get_weather(request)
else:
return a2a_pb2.A2AResponse(
request_id=request.request_id,
sender_id=self.agent_id,
success=False,
error_message=f"Unsupported action: {request.action}"
)
def _handle_get_weather(self, request):
try:
payload = json.loads(request.payload)
date = payload.get("date")
location = payload.get("location", "Beijing") # 默认地点
# 这里是调用MCP工具的示意点
# weather_data = self.mcp_tool_invoke("weather_api", {"date": date, "location": location})
# 为简化，我们使用模拟数据
weather_data = self.weather_db.get(date)
if not weather_data:
return a2a_pb2.A2AResponse(
request_id=request.request_id,
sender_id=self.agent_id,
success=False,
error_message=f"Weather data not found for date: {date}"
)
response_data = {
"location": location,
"date": date,
"weather": weather_data
}
return a2a_pb2.A2AResponse(
request_id=request.request_id,
sender_id=self.agent_id,
success=True,
data=json.dumps(response_data)
)
except json.JSONDecodeError:
return a2a_pb2.A2AResponse(
request_id=request.request_id,
sender_id=self.agent_id,
success=False,
error_message="Invalid payload JSON format."
)
except Exception as e:
return a2a_pb2.A2AResponse(
request_id=request.request_id,
sender_id=self.agent_id,
success=False,
error_message=f"Internal error: {str(e)}"
)
if __name__ == "__main__":
import logging
logging.basicConfig(level=logging.INFO)
agent = WeatherAgent()
agent.start()
接着，是更复杂的TripAgent。 它需要具备发现并调用WeatherAgent的能力，然后结合业务逻辑生成行程建议。
# trip_agent.py
import json
import grpc
import a2a_pb2
import a2a_pb2_grpc
from base_agent import BaseAgent
import consul
class TripAgent(BaseAgent):
def __init__(self):
capabilities = [a2a_pb2.Capability.TRIP_PLANNING]
super().__init__(
agent_id="trip_001",
name="TripAgent",
capabilities=capabilities,
port=50053
)
def _route_request(self, request):
if request.action == "plan_trip":
return self._handle_plan_trip(request)
else:
return a2a_pb2.A2AResponse(
request_id=request.request_id,
sender_id=self.agent_id,
success=False,
error_message=f"Unsupported action: {request.action}"
)
def _handle_plan_trip(self, request):
"""核心逻辑：协调WeatherAgent，生成行程计划"""
try:
payload = json.loads(request.payload)
date = payload.get("date")
activity = payload.get("activity")
location = payload.get("location", "Beijing")
# 1. 发现可用的WeatherAgent
weather_agent = self._discover_agent(a2a_pb2.Capability.WEATHER_QUERY)
if not weather_agent:
return a2a_pb2.A2AResponse(
request_id=request.request_id,
sender_id=self.agent_id,
success=False,
error_message="No available WeatherAgent found."
)
# 2. 通过A2A协议向WeatherAgent发起请求
weather_response = self._call_weather_agent(weather_agent, date, location)
if not weather_response.success:
return weather_response # 直接返回错误响应
weather_info = json.loads(weather_response.data)
condition = weather_info["weather"]["condition"]
# 3. 根据天气和活动生成行程建议（业务逻辑）
plan = self._generate_plan(date, activity, condition, location)
# 4. 返回最终结果
final_result = {
"request": payload,
"weather_info": weather_info,
"recommendation": plan
}
return a2a_pb2.A2AResponse(
request_id=request.request_id,
sender_id=self.agent_id,
success=True,
data=json.dumps(final_result)
)
except Exception as e:
return a2a_pb2.A2AResponse(
request_id=request.request_id,
sender_id=self.agent_id,
success=False,
error_message=f"Trip planning failed: {str(e)}"
)
def _discover_agent(self, capability):
"""通过Consul发现具备特定能力的Agent"""
try:
_, services = self.consul_client.agent.services()
for svc_id, svc_info in services.items():
tags = svc_info.get('Tags', [])
if capability.name in tags and svc_info['Service'] != self.name:
# 返回第一个找到的Agent信息
return {
'name': svc_info['Service'],
'endpoint': f"{svc_info['Address']}:{svc_info['Port']}",
'id': svc_id
}
except Exception as e:
logging.error(f"Service discovery failed: {e}")
return None
def _call_weather_agent(self, agent_info, date, location):
"""通过gRPC调用WeatherAgent的HandleRequest方法"""
channel = grpc.insecure_channel(agent_info['endpoint'])
stub = a2a_pb2_grpc.A2AServiceStub(channel)
request_payload = json.dumps({"date": date, "location": location})
a2a_request = a2a_pb2.A2ARequest(
request_id=f"req_{datetime.now().timestamp()}",
sender_id=self.agent_id,
conversation_id=f"conv_{datetime.now().timestamp()}",
action="get_weather",
payload=request_payload
)
try:
response = stub.HandleRequest(a2a_request)
return response
except grpc.RpcError as e:
logging.error(f"gRPC call to {agent_info['name']} failed: {e}")
return a2a_pb2.A2AResponse(
request_id=a2a_request.request_id,
sender_id=self.agent_id,
success=False,
error_message=f"Communication error with {agent_info['name']}: {e.details()}"
)
finally:
channel.close()
def _generate_plan(self, date, activity, condition, location):
"""简单的行程规划逻辑"""
plans = {
"Sunny": f"**完美天气！** 在{location}的{date}进行{activity}再合适不过了。记得做好防晒，补充水分。",
"Rainy": f"**天气不佳。** {date}在{location}有雨，不适合户外{activity}。建议改为室内活动或改期。",
"Cloudy": f"**天气尚可。** {date}在{location}是多云天气，进行{activity}没问题，但最好带上雨具以防万一。",
"Snowy": f"**下雪天。** {date}在{location}有降雪，进行{activity}需特别注意保暖和安全，评估活动是否可行。"
}
return plans.get(condition, f"天气状况为'{condition}'，请自行判断是否适合进行{activity}。")
if __name__ == "__main__":
import logging
logging.basicConfig(level=logging.INFO)
agent = TripAgent()
agent.start()
至此，两个核心Agent已经构建完毕。它们各自独立运行，通过我们定义的A2A协议（gRPC + protobuf）和Consul进行服务发现与通信。
5. 系统集成、测试与进阶优化
现在，让我们把整个系统跑起来，并进行端到端的测试。
第一步：启动基础设施。 我们需要先启动Consul服务注册中心。使用Docker可以最快速地完成：
docker run -d --name=consul --net=host consul agent -dev -client=0.0.0.0
第二步：启动Agent。 打开两个终端，分别运行：
# 终端1：启动WeatherAgent
python weather_agent.py
# 终端2：启动TripAgent
python trip_agent.py
如果一切正常，你将在日志中看到两个Agent成功启动并向Consul注册的信息。
第三步：测试协作。 我们不需要直接调用gRPC，而是为TripAgent暴露一个简单的HTTP API（使用FastAPI）供用户或前端调用，这更符合实际应用场景。
# trip_api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import json
import a2a_pb2
import a2a_pb2_grpc
import grpc
import logging
app = FastAPI(title="Trip Planning API")
# 假设我们知道TripAgent的gRPC地址（实际应由服务发现获得）
TRIP_AGENT_ENDPOINT = "localhost:50053"
class TripRequest(BaseModel):
date: str
activity: str
location: str = "Beijing"
@app.post("/plan")
async def plan_trip(request: TripRequest):
"""用户入口：请求规划行程"""
channel = grpc.insecure_channel(TRIP_AGENT_ENDPOINT)
stub = a2a_pb2_grpc.A2AServiceStub(channel)
a2a_request = a2a_pb2.A2ARequest(
request_id=f"http_req_{datetime.now().timestamp()}",
sender_id="api_gateway",
conversation_id=f"conv_http_{datetime.now().timestamp()}",
action="plan_trip",
payload=request.json()
)
try:
response = stub.HandleRequest(a2a_request)
channel.close()
if response.success:
return json.loads(response.data)
else:
raise HTTPException(status_code=500, detail=response.error_message)
except grpc.RpcError as e:
logging.error(f"Failed to call TripAgent: {e}")
raise HTTPException(status_code=503, detail="TripAgent service unavailable")
except Exception as e:
logging.error(f"Unexpected error: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
启动API服务：python trip_api.py。现在，你可以用curl或Postman进行测试了：
curl -X POST "http://localhost:8000/plan" \
-H "Content-Type: application/json" \
-d '{"date": "2024-07-15", "activity": "hiking", "location": "Beijing"}'
如果一切顺利，你将收到一个包含天气信息和行程建议的完整JSON响应。这标志着你的多Agent系统已经成功运行！
进阶优化与思考：
一个可用的原型已经搭建完成，但要投入生产环境，还有很长的路要走。以下是一些关键的优化方向：
- 增强Orchestrator：将协调逻辑从TripAgent中剥离，形成一个独立的Orchestrator Agent。它专门负责理解用户意图、任务分解、Agent调度和结果聚合，使系统架构更清晰。
- 引入工作流引擎：对于更复杂的任务链（如：查询天气 -> 推荐景点 -> 规划路线 -> 估算预算），可以考虑集成像LangGraph或Prefect这样的工作流/有向无环图（DAG）引擎，可视化地编排Agent之间的执行顺序和依赖关系。
- 完善可观测性：为每个A2A请求添加唯一的Trace ID，并集成像OpenTelemetry这样的工具，实现分布式追踪。这能让你清晰地看到一个请求在各个Agent间的流转路径和耗时，便于调试和性能分析。
- 实现Agent状态管理：当前的对话是“一问一答”式的。对于需要多轮交互的复杂任务，需要为每个会话（Conversation）维护上下文状态。可以考虑使用Redis或数据库来持久化会话状态。
- 安全与权限：在A2A请求中添加身份认证和授权机制。例如，使用JWT令牌验证发起请求的Agent是否合法，并检查其是否有权调用目标Agent的某个能力。
在项目初期，我过于追求每个Agent功能的“大而全”，结果导致单个Agent逻辑复杂且脆弱。拆分成专注的微服务Agent后，不仅开发调试更简单，系统的弹性和扩展性也大大增强。当需要新增一个“翻译Agent”来服务国际用户时，我只需要按照同样的A2A协议规范实现它，并注册到Consul，Orchestrator就能自动发现并调用它，整个架构的灵活性得到了验证。多Agent系统不是银弹，但对于需要组合多种专业能力的复杂AI应用场景，它提供了一条清晰且富有弹性的架构路径。
更多推荐
11
0- 0
已为社区贡献17条内容
所有评论(0)