A2A 第一堂课
我们来通过展示一个基于A2A协议协议搭建的一个智能体平台，从而实现模型之间的对话
这需要几部分构成，一方面需要一个Host Agent，其会拆解用户的意图，拆解为具体的子任务，之后并行触发多个Remote Agent
将多个负责不同模块的Agent的响应内容汇总之后，由Host Agent格式化后反馈，呈现给用户。
之后回到这个项目，通过git clone https://github.com/huangjia2019/a2a-in-action.git 获取到代码
在这个项目之中，存在了多个Agent的文件夹，每个Agent有不同的环境和依赖。
之后我们启动这个Demo
uv run main.py
启动完成之后可以进入到页面进行交互了，这时候可以进行对话的是Host Agent，当然这个内部叫做Simple Agent
但由于目前没有配置其他的Agent，目前其也没有什么额外的能力，这时候就需要配置Remote Agents
这里我们再启动一些Agent，注册进去，方便Host Agent去进行调用，这里我们启动langgraph的相关Agent。这是一个负责汇率计算的Agent。
这里通过配置Agent启动后暴露的Port，从而获取到Agent Card
之后再去询问一些关于货币兑换的问题就发现Agent支持了。
那么之后我们就可以聚焦内部的A2A系统具体实现。在其中，主要由如下组件构成。
Agent Card 代表着Agent的能力声明， Skill 代表着具体可执行的能力， Task Manager 代表着任务流和状态管理 Server 提供HTTP接口让其他代理和客户端访问。
对于代码来说，首先是Host Agent的服务实现，文件中/demo/ui/service/server/server.py之中的ConversationServer 类，注册了Host Agent相关的代码，其中包含的关键代码有
async def _send_message 接收前端消息，并且返回消息ID
async def _register_agent 注册远程Agent
async def _list_agents 返回已经注册的所有Agent信息
之后是Host Agent，首先是ADKHostManager，继承了ApplicationManager，
其中包含一些核心函数
self._host_agent = HostAgent([], self.task_callback)
负责调用HostAgent，主要是多智能体的注册，能力发现，任务分发，回调等。
之后是async def process_message(self, message: Message)
负责用户消息等，并触发不同智能体的推理和任务流转。
def register_agent(self, url)，支持通过URL动态注册远程Agent
其中的process_message将用户的输入转换为标准的A2A消息格式，交给HostAgent进行协作后处理，最后返回事件流。
之后就是Remote Agent的实现，在这个项目中的Agent目录中，每一个子目录就是一个外部Agent的实现
这里我们看看LangGraph实现的货币转换Agent
在这个langgraph中的main.py中通过A2AServer将CurrencyAgent暴露出来，并且注册了自己的AgentCard。
在这个main函数之中，核心逻辑就是Agent Card 和skill的描述
其中说明了这个AI Agent的能力，能做什么，怎么通信，支持哪些能力
其中还有A2AServer的启动相关代码。
| try:
if not os.getenv(‘GOOGLE_API_KEY’): raise MissingAPIKeyError( ‘GOOGLE_API_KEY environment variable not set.’ ) capabilities = AgentCapabilities(streaming=True, pushNotifications=True) skill = AgentSkill( id=’convert_currency’, name=’Currency Exchange Rates Tool’, description=’Helps with exchange values between various currencies’, tags=[‘currency conversion’, ‘currency exchange’], examples=[‘What is exchange rate between USD and GBP?’], ) agent_card = AgentCard( name=’Currency Agent’, description=’Helps with exchange rates for currencies’, url=f’http://{host}:{port}/’, version=’1.0.0′, defaultInputModes=CurrencyAgent.SUPPORTED_CONTENT_TYPES, defaultOutputModes=CurrencyAgent.SUPPORTED_CONTENT_TYPES, capabilities=capabilities, skills=[skill], ) notification_sender_auth = PushNotificationSenderAuth() notification_sender_auth.generate_jwk() server = A2AServer( agent_card=agent_card, task_manager=AgentTaskManager( agent=CurrencyAgent(), notification_sender_auth=notification_sender_auth, ), host=host, port=port, ) server.app.add_route( ‘/.well-known/jwks.json’, notification_sender_auth.handle_jwks_endpoint, methods=[‘GET’], ) logger.info(f’Starting server on {host}:{port}’) server.start() except MissingAPIKeyError as e: logger.error(f’Error: {e}’) exit(1) except Exception as e: logger.error(f’An error occurred during server startup: {e}’) exit(1) |
最后是A2A Client端的实现
其内部封装了和远程Agent 的HTTP通信，所有的请求以JSON-RPC格式已发送
关键方法如下
async def send_message(self, payload: SendMessageRequest)，通过HTTP POST将消息发送到远程Agent的 /message/send 接口。
async def register_agent(self, payload: RegisterAgentRequest)，远程注册Agent。
最后总结一下，其提供了一个通信机制，让各个Agent可以提供自己的技能给一个主Agent，让其进行扩展，拥有更多的能力。
每个Agent启动的时候会生成自己的AgentCard，包括名称，描述，服务URL等信息。
启动之后进行注册，而Host端通过注册和发现机制拉取到AgentCard，最终通过能力调用，实现代理之间的能力发现和互相操作。