【A2A】管中窥豹，google源码python-demo介绍

news2025/5/10 11:19:22

在这里插入图片描述

前言

A2A（Agent2Agent）是 Google 推出的一项新协议，旨在解决多智能体（Multi-Agent）系统中跨平台、跨组织协作的难题。它为 AI 代理之间的通信、协作和任务分工提供了一个统一的标准，可以类比为网页世界的 HTTP 协议——即 AI 代理之间的“通用语言”

项目地址：https://github.com/google/A2A/tree/main/samples/python

一、项目架构介绍

1. 项目概述

基于 A2A (Agent-to-Agent) 协议的示例项目，展示了如何使用不同的 AI 框架来实现智能代理之间的通信和协作。

2. 项目结构

samples/python/
├── agents/          # 各种AI代理实现
│   ├── ag2/         # AG2框架实现的代理
│   ├── crewai/      # CrewAI框架实现的代理
│   ├── langgraph/   # LangGraph框架实现的代理
│   ├── google_adk/  # Google ADK框架实现的代理
│   ├── llama_index_file_chat/  # LlamaIndex实现的文件聊天代理
│   ├── marvin/      # Marvin框架实现的代理
│   ├── mindsdb/     # MindsDB实现的代理
│   └── semantickernel/  # Semantic Kernel实现的代理
├── hosts/           # 客户端实现
├── common/          # 公共代码
└── .vscode/         # VS Code配置

3. 技术特点

  1. 多框架支持：

    • 支持多种流行的AI框架（LangGraph、CrewAI、AG2等）
    • 每个框架都有独立的代理实现

  2. 标准化通信：

    • 使用A2A协议进行代理间通信
    • 基于HTTP的通信机制
    • 统一的请求/响应格式

  3. 客户端实现：

    • 提供CLI命令行客户端
    • 支持与多个代理的交互
    • 包含任务编排功能

4. 运行环境要求

  • Python 3.13 或更高版本
  • UV 包管理器
  • 各框架所需的API密钥（如Google API Key等）

5. 使用流程

  1. 启动代理服务器：

    cd samples/python/agents/[agent_name]
uv run .

  2. 启动客户端：

    cd samples/python/hosts/cli
uv run .

6. 主要功能模块

6.1 代理实现（agents/）
  • AG2代理：基于AG2框架的代理实现
  • CrewAI代理：使用CrewAI框架的代理
  • LangGraph代理：基于LangGraph的代理
  • Google ADK代理：使用Google Agent Development Kit的代理
  • LlamaIndex代理：文件聊天功能
  • Marvin代理：基于Marvin框架的代理
  • MindsDB代理：数据库交互代理
  • Semantic Kernel代理：基于Microsoft Semantic Kernel的代理
6.2 客户端实现（hosts/）
  • CLI命令行界面
  • 支持多代理交互
  • 任务编排功能
6.3 公共模块（common/）
  • A2A协议实现
  • 共享工具类
  • 通用接口定义

7. 项目特点

  1. 模块化设计：

    • 每个代理都是独立的模块
    • 可以单独运行和测试

  2. 标准化接口：

    • 统一的A2A协议
    • 一致的通信格式

  3. 可扩展性：

    • 易于添加新的代理实现
    • 支持不同的AI框架

  4. 示例性质：

    • 主要用于演示A2A功能
    • 非生产级代码

二、场景介绍

1. 核心场景代理实现

1.1 LangGraph 货币转换代理
  • 功能：提供货币汇率转换服务
  • 技术特点：
    • 使用 LangGraph 框架
    • 集成 Google Gemini 模型
    • 支持多轮对话
    • 实时流式响应
    • 使用 Frankfurter API 获取实时汇率
  • 通信流程：
    Client Server Agent API 发送货币查询 转发查询 获取汇率数据 返回汇率 处理结果 返回结果 Client Server Agent API
1.2 CrewAI 图像生成代理
  • 功能：基于文本描述生成图像
  • 技术特点：
    • 使用 CrewAI 框架
    • 集成 Google Gemini API
    • 支持图像修改
    • 缓存系统
  • 通信流程：
    Client Server Agent Gemini 发送图像生成请求 转发请求 生成图像 返回图像 存储并返回ID 返回图像 Client Server Agent Gemini

2. 通信协议（A2A）

2.1 请求格式
{
  "jsonrpc": "2.0",
  "id": "unique_id",
  "method": "tasks/send",
  "params": {
    "id": "task_id",
    "sessionId": "session_id",
    "acceptedOutputModes": ["text"],
    "message": {
      "role": "user",
      "parts": [{
        "type": "text",
        "text": "query"
      }]
    }
  }
}
2.2 响应格式
{
  "jsonrpc": "2.0",
  "id": "unique_id",
  "result": {
    "id": "task_id",
    "status": {
      "state": "completed",
      "timestamp": "timestamp"
    },
    "artifacts": [{
      "parts": [{
        "type": "text",
        "text": "response"
      }],
      "index": 0
    }]
  }
}

三、Client代码实现

  1. 核心类：A2AClient

a) 初始化

def __init__(self, agent_card: AgentCard = None, url: str = None, timeout: TimeoutTypes = 60.0):
    # 支持两种初始化方式：
    # 1. 通过 AgentCard 初始化
    # 2. 直接通过 URL 初始化
    # 默认超时时间 60 秒

b) 主要方法

# 1. 发送任务
async def send_task(self, payload: dict[str, Any]) -> SendTaskResponse:
    # 同步任务发送
    # 返回任务响应

# 2. 流式任务
async def send_task_streaming(self, payload: dict[str, Any]) -> AsyncIterable[SendTaskStreamingResponse]:
    # 支持 Server-Sent Events (SSE)
    # 返回流式响应

# 3. 获取任务
async def get_task(self, payload: dict[str, Any]) -> GetTaskResponse:
    # 获取任务状态和结果

# 4. 取消任务
async def cancel_task(self, payload: dict[str, Any]) -> CancelTaskResponse:
    # 取消正在执行的任务

# 5. 设置回调
async def set_task_callback(self, payload: dict[str, Any]) -> SetTaskPushNotificationResponse:
    # 设置任务完成后的回调通知

# 6. 获取回调
async def get_task_callback(self, payload: dict[str, Any]) -> GetTaskPushNotificationResponse:
    # 获取当前任务的回调配置
  1. 辅助类：A2ACardResolver
class A2ACardResolver:
    def __init__(self, base_url, agent_card_path='/.well-known/agent.json'):
        # 初始化解析器
        # 默认从 /.well-known/agent.json 获取代理卡片

    def get_agent_card(self) -> AgentCard:
        # 获取并解析代理卡片
        # 返回 AgentCard 对象
  1. 使用示例
# 1. 基本使用
client = A2AClient(url="http://agent-server")
response = await client.send_task({
    "message": "生成一张图片",
    "sessionId": "session-123"
})

# 2. 流式响应
async for event in client.send_task_streaming({
    "message": "生成一张图片",
    "sessionId": "session-123"
}):
    print(event)

# 3. 获取代理卡片
resolver = A2ACardResolver("http://agent-server")
card = resolver.get_agent_card()

客户端实现提供了：

  1. 完整的 A2A 协议支持
  2. 异步操作支持
  3. 类型安全
  4. 错误处理
  5. 流式响应
  6. 代理发现

四、Server实现

  1. 核心类：A2AServer

a) 初始化

def __init__(self, host='0.0.0.0', port=5000, endpoint='/', agent_card: AgentCard = None, task_manager: TaskManager = None):
    # 配置服务器参数
    # 初始化路由
    # 设置代理卡片和任务管理器

b) 主要功能

# 1. 启动服务器
def start(self):
    # 验证必要组件
    # 启动 uvicorn 服务器

# 2. 处理请求
async def _process_request(self, request: Request):
    # 解析请求
    # 路由到对应的处理方法
    # 返回响应

# 3. 获取代理卡片
def _get_agent_card(self, request: Request) -> JSONResponse:
    # 返回代理卡片信息
  1. 任务管理器：TaskManager

a) 抽象基类

class TaskManager(ABC):
    # 定义任务管理接口
    @abstractmethod
    async def on_get_task(self, request: GetTaskRequest) -> GetTaskResponse
    @abstractmethod
    async def on_cancel_task(self, request: CancelTaskRequest) -> CancelTaskResponse
    @abstractmethod
    async def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse
    # ... 其他抽象方法

b) 内存实现

class InMemoryTaskManager(TaskManager):
    def __init__(self):
        # 初始化存储
        self.tasks = {}  # 任务存储
        self.push_notification_infos = {}  # 推送通知配置
        self.task_sse_subscribers = {}  # SSE 订阅者
  1. 关键功能实现

a) 任务管理

# 1. 获取任务
async def on_get_task(self, request: GetTaskRequest) -> GetTaskResponse:
    # 获取任务状态和结果

# 2. 取消任务
async def on_cancel_task(self, request: CancelTaskRequest) -> CancelTaskResponse:
    # 取消正在执行的任务

# 3. 更新任务
async def update_store(self, task_id: str, status: TaskStatus, artifacts: list[Artifact]) -> Task:
    # 更新任务状态和结果

b) 推送通知

# 1. 设置通知
async def set_push_notification_info(self, task_id: str, notification_config: PushNotificationConfig):
    # 配置任务完成通知

# 2. 获取通知
async def get_push_notification_info(self, task_id: str) -> PushNotificationConfig:
    # 获取任务通知配置

c) SSE 支持

# 1. 设置 SSE 消费者
async def setup_sse_consumer(self, task_id: str, is_resubscribe: bool = False):
    # 创建 SSE 事件队列

# 2. 事件入队
async def enqueue_events_for_sse(self, task_id, task_update_event):
    # 将事件发送给订阅者

# 3. 事件出队
async def dequeue_events_for_sse(self, request_id, task_id, sse_event_queue):
    # 从队列获取事件并发送
  1. 错误处理
def _handle_exception(self, e: Exception) -> JSONResponse:
    # 处理不同类型的错误
    if isinstance(e, json.decoder.JSONDecodeError):
        json_rpc_error = JSONParseError()
    elif isinstance(e, ValidationError):
        json_rpc_error = InvalidRequestError()
    else:
        json_rpc_error = InternalError()

这个服务器实现提供了：

  1. 完整的 A2A 协议支持
  2. 异步操作支持
  3. 任务管理
  4. 推送通知
  5. 流式响应
  6. 错误处理

一个功能完整的 A2A 服务器实现，可以作为其他语言实现的参考。

五、LangGraph 货币转换Agent

/samples/python/agents/langgraph

  1. 整体架构
  • 采用三层架构：
    • __main__.py: 服务器入口点
    • agent.py: 核心代理实现
    • task_manager.py: 任务管理实现
  1. 核心组件详解

a) 入口点 (__main__.py)

- 配置服务器参数（host, port）
- 设置代理能力（streaming, pushNotifications）
- 定义代理技能（convert_currency）
- 创建代理卡片（AgentCard）
- 初始化服务器和任务管理器

b) 代理实现 (agent.py)

- 使用 LangGraph 框架实现 ReAct 模式
- 核心组件：
  1. 工具函数：get_exchange_rate
     - 调用 Frankfurter API 获取实时汇率
     - 支持指定日期查询
     - 错误处理和响应验证
  
  2. 响应格式：ResponseFormat
     - status: input_required/completed/error
     - message: 响应消息
  
  3. CurrencyAgent 类
     - 使用 Google Gemini 2.0 Flash 模型
     - 支持同步调用（invoke）和流式响应（stream）
     - 状态管理和会话追踪

c) 任务管理器 (task_manager.py)

- 继承自 InMemoryTaskManager
- 主要功能：
  1. 任务验证
     - 检查输出模式兼容性
     - 验证推送通知配置
  
  2. 任务处理
     - 同步任务处理（on_send_task）
     - 流式任务处理（on_send_task_subscribe）
     - 任务状态更新和通知
  
  3. 推送通知
     - 支持任务状态变更通知
     - 验证通知 URL 所有权
  1. 关键流程

a) 同步请求流程

1. 客户端发送请求
2. 任务管理器验证请求
3. 代理处理请求
4. 更新任务状态
5. 发送响应

b) 流式请求流程

1. 客户端订阅任务
2. 创建 SSE 事件队列
3. 异步处理代理响应
4. 实时更新任务状态
5. 发送流式事件

六、CrewAI 图像生成Agent

  1. 整体架构
  • 采用三层架构：
    • __main__.py: 服务器入口点
    • agent.py: 核心代理实现
    • task_manager.py: 任务管理实现
  1. 核心组件详解

a) 入口点 (__main__.py)

- 配置服务器参数（host, port）
- 设置代理能力（streaming=False）
- 定义代理技能（image_generator）
- 创建代理卡片（AgentCard）
- 初始化服务器和任务管理器

b) 代理实现 (agent.py)

- 使用 CrewAI 框架实现图像生成
- 核心组件：
  1. 数据模型：Imagedata
     - id: 图像唯一标识
     - name: 图像名称
     - mime_type: MIME类型
     - bytes: Base64编码的图像数据
     - error: 错误信息

  2. 工具函数：generate_image_tool
     - 使用 Google Gemini API 生成图像
     - 支持图像修改
     - 缓存管理
     - 错误处理

  3. ImageGenerationAgent 类
     - 支持文本和图像输入
     - 使用 CrewAI 框架
     - 图像生成和修改功能
     - 会话状态管理

c) 任务管理器 (task_manager.py)

- 继承自 InMemoryTaskManager
- 主要功能：
  1. 任务处理
     - 验证输出模式
     - 处理任务请求
     - 管理任务状态
  
  2. 图像处理
     - 获取图像数据
     - 处理图像响应
     - 错误处理
  1. 关键流程

a) 图像生成流程

1. 接收用户提示
2. 创建 CrewAI 任务
3. 调用 Gemini API
4. 处理生成的图像
5. 返回图像数据

b) 图像修改流程

1. 接收修改请求
2. 获取参考图像
3. 生成新图像
4. 更新缓存
5. 返回结果

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2372260.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

004-nlohmann/json 快速认识-C++开源库108杰

004-nlohmann/json 快速认识-C++开源库108杰

了解 nlohmann/json 的特点；理解编程中 “数据战场”划分的概念；迅速上手多种方式构建一个JSON对象； 1 特点与安装 nlohmann/json 是一个在 github 长期霸占 “JSON” 热搜版第1的CJSON处理库。它的最大优点是与 C 标准库的容器数据&#xf…
阅读更多...
Matlab实现CNN-BiLSTM时间序列预测未来

Matlab实现CNN-BiLSTM时间序列预测未来

Matlab实现CNN-BiLSTM时间序列预测未来 目录 Matlab实现CNN-BiLSTM时间序列预测未来效果一览基本介绍程序设计参考资料 效果一览 基本介绍 1.Matlab实现CNN-BiLSTM时间序列预测未来； 2.运行环境Matlab2023b及以上，data为数据集，单变量时间序…
阅读更多...
C语言| sizeof(array)占多少字节

C语言| sizeof(array)占多少字节

C语言| 数组名作为函数参数 sizeof(数组名); 可以求出整个数组在内存中所占的字节数。 被调函数Array_Sum()中，数组array使用sizeof会得到多少？ 实参数组a占32字节，实参a传给形参array，只占4字节。 原因如下： 数组名做…
阅读更多...
【文件系统—散列结构文件】

【文件系统—散列结构文件】

文章目录 一、实验目的实验内容设计思路 三、实验代码实现四、总结 一、实验目的 理解linux文件系统的内部技术，掌握linux与文件有关的系统调用命令，并在此基础上建立面向随机检索的散列结构文件；## 二、实验内容与设计思想 实验内容 1.设…
阅读更多...
World of Warcraft [CLASSIC][80][Deluyia] [Fragment of Val‘anyr]

World of Warcraft [CLASSIC][80][Deluyia] [Fragment of Val‘anyr]

瓦兰奈尔的碎片 [Fragment of Valanyr] 有时候下个班打个游戏，没想到套路也这么多，唉，何况现实生活，这一个片版本末期才1000G，30个，也就30000G，时光徽章等同月卡15000G，折合一下也就…
阅读更多...
数组和指针典型例题合集（一维数组、字符数组、二维数组）

数组和指针典型例题合集（一维数组、字符数组、二维数组）

1.一维数组 数组名的理解 数组名是数组首元素（第一个元素）的地址 但是有两个例外： 1.sizeof （数组名）—— 数组名表示整个数组，就算的是整个数组的大小，单位是字节。 2.&数组名 —— 数…
阅读更多...
地级市-机器人、人工智能等未来产业水平（2009-2023年）-社科数据

地级市-机器人、人工智能等未来产业水平（2009-2023年）-社科数据

地级市-机器人、人工智能等未来产业水平（2009-2023年）-社科数据https://download.csdn.net/download/paofuluolijiang/90623814 https://download.csdn.net/download/paofuluolijiang/90623814 此数据集统计了2009-2023年全国地级市在机器人、人工智能等…
阅读更多...
epub格式转txt格式工具，txt批量转PDF

epub格式转txt格式工具，txt批量转PDF

epub格式转txt格式工具，功能如图： txt格式批量转PDF 参考原文：epub格式转txt格式工具，txt批量转PDF 轻轻一点就关注, 好运连连挡不住，点个关注吧。
阅读更多...
电赛经验分享——模块篇

电赛经验分享——模块篇

1、前言 打算在这一个专栏中，分享一些本科控制题电赛期间的经验，和大家共同探讨，也希望能帮助刚刚参加电赛的同学，了解一些基本的知识。一些见解和看法可能不同或有错误，欢迎批评指正。 在本文中，主要介绍笔…
阅读更多...
JVM之内存管理（一）

JVM之内存管理（一）

部分内容来源：JavaGuide二哥Java 图解JVM内存结构 内存管理快速复习 栈帧：局部变量表，动态链接（符号引用转为真实引用），操作数栈（存储中间结算结果），方法返回地址 运行时…
阅读更多...
鸿蒙编译boost整合linux跨平台应用

鸿蒙编译boost整合linux跨平台应用

openharmony deveco 4.1支持armeabi-v7a deveco 5.0后不支持arm32位系统 boost编译 使用deveco的写cmake集成boost boost使用1.88的最新版本，带cmake工具链 https://github.com/boostorg/boost.git boost的源码都在sub_module中 deveco 4.1的版本sdk最高到9&am…
阅读更多...
rabbitMQ消息问题与解决

rabbitMQ消息问题与解决

rabbitMQ 消息顺序性、消息幂等性、消息不丢失、最终一致性、补偿机制、消息队列设计 1.消息顺序性 溯源： 消息队列中的若干消息如果是对同一个数据进行操作，这些操作具有前后的关系，必须要按前后的顺序执行，否则就会造成数据异常…
阅读更多...
Java SE(10)——抽象类接口

Java SE(10)——抽象类接口

1.抽象类 1.1 概念 在之前讲Java SE(6)——类和对象（一）的时候说过，所有的对象都可以通过类来抽象。但是反过来，并不是说所有的类都是用来抽象一个具体的对象。如果一个类本身没有足够的信息来描述一个具体的对象，而…
阅读更多...
学习threejs，使用Physijs物理引擎

学习threejs，使用Physijs物理引擎

👨‍⚕️ 主页： gis分享者 👨‍⚕️ 感谢各位大佬 点赞👍 收藏⭐ 留言📝 加关注✅! 👨‍⚕️ 收录于专栏：threejs gis工程师 文章目录 一、🍀前言1.1 ☘️Physijs 物理引擎1.1.1 ☘️…
阅读更多...
allure生成测试报告（搭配Pytest、allure-pytest）

allure生成测试报告（搭配Pytest、allure-pytest）

文章目录 前言allure简介allure安装软件下载安装配置环境变量安装成功验证 allure运行流程allure装饰器函数基本说明装饰器函数使用allure.attach 命令行运行利用allure-pytest生成中间结果json 查看测试报告总览页面每个tab页的说明类别页面测试套图表页面时间刻度功能页面包 …
阅读更多...
龙虎榜——20250509

龙虎榜——20250509

上证指数今天缩量，整体跌多涨少，走势处于日线短期的高位~ 深证指数今天缩量小级别震荡，大盘股表现更好~ 2025年5月9日龙虎榜行业方向分析 一、核心行业方向 军工航天 • 代表个股：航天南湖、天箭科技、襄阳轴承。 • 驱动逻辑…
阅读更多...
操作系统的初步了解

操作系统的初步了解

目录 引言：什么是操作系统？ 一、设计操作系统的目的 二、操作系统是做什么的： 操作系统主要有四大核心任务： 1. 管理硬件 2. 运行软件 3. 存储数据 4. 提供用户界面 如何理解操作系统的管理呢？ 1. 什么是操作…
阅读更多...
软件工程之软件项目管理深度解析

软件工程之软件项目管理深度解析

前文基础： 1.软件工程学概述：软件工程学概述-CSDN博客 2.软件过程深度解析：软件过程深度解析-CSDN博客 3.软件工程之需求分析涉及的图与工具：软件工程之需求分析涉及的图与工具-CSDN博客 4.软件工程之形式化说明技术深度解…
阅读更多...
基于Boost库、Jsoncpp、cppjieba、cpp-httplib等构建Boost搜索引擎

基于Boost库、Jsoncpp、cppjieba、cpp-httplib等构建Boost搜索引擎

⭐️个人主页：小羊 ⭐️所属专栏：项目 很荣幸您能阅读我的文章，诚请评论指点，欢迎欢迎 ~ 目录 项目背景技术栈和项目环境正排索引和倒排索引数据去标签与清洗下载数据源去标签 建立索引构建正排索引构建倒排索引 建立搜索引擎h…
阅读更多...
Qt 通过控件按钮实现hello world + 命名规范（7）

Qt 通过控件按钮实现hello world + 命名规范（7）

文章目录 使用编辑框来完成 hello world通过编辑图形化界面方式通过纯代码方式 通过按钮的方式来创建 hello world通过编辑图形化界面方式通过纯代码方式 总结Qt Creator中的快捷键如何使用文档命名规范 简介：这篇文章着重点并不在于创建hello world程序&#xff0c…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023