Google的A2A智能体群聊

Google的A2A智能体群聊

针对Google的A2A智能体群聊，进行一个基础的Demo演示

1-核心知识点

• 1-开发一个最简单的A2A大模型demo示例
• 2-熟悉A2A代码框架和基础使用

2-参考网址

• A2A代码示例仓库：https://gitee.com/enzoism/python_a2a_travel

3-动手实操

1-UV环境搭建

# 1-uv环境搭建uv python pin3.11.4 uv init python_a2a&&cdpython_a2a uv venv&&source.venv/bin/activate uvaddpython-dotenv pydantic# 2-uv安装requirements.txtuv python pin3.11.4 uv venv&&source.venv/bin/activate uv pipinstall-r requirements.txt uvadd-r requirements.txt

2-A2A项目

3-启动A2A

adk web

• 打印日志

(python_a2a)(base)MacBook-Pro:python_a2a rong$ adk web INFO: Started server process[31551]INFO: Waitingforapplication startup. +-----------------------------------------------------------------------------+|ADK Web Server started||||Forlocaltesting, access at http://127.0.0.1:8000.|+-----------------------------------------------------------------------------+ INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8000(Press CTRL+C to quit)

4-访问网址

拷贝控制台的地址：http://127.0.0.1:8000

4-代码说明

目录

• 项目概述
• 技术架构
• 项目结构
• 环境配置
• 核心代码解析
• 运行方式
• 扩展开发

项目概述

本项目是一个基于 Google ADK (Agent Development Kit) 的Agent-to-Agent (A2A)通信框架示例。通过多代理协作模式，实现了一个智能旅行规划系统，展示了主代理如何与多个子代理协同工作以完成复杂任务。

核心特性

• 多代理协作：主代理 (Coordinator) 与子代理 (Flight Agent、Hotel Agent) 协同工作
• 自然语言处理：通过 LLM 理解用户意图并提取结构化信息
• 分布式任务执行：不同代理专注于特定领域的任务处理
• 统一状态管理：自动管理会话状态和代理间通信

应用场景

用户输入 → 协调员代理 → [航班代理 + 酒店代理] → 综合结果

技术架构

架构分层

┌─────────────────────────────────────────────────────────┐ │ 用户交互层 │ │ (自然语言输入/输出) │ └────────────────────┬────────────────────────────────────┘ │ ┌────────────────────▼────────────────────────────────────┐ │ 协调员代理 (Coordinator) │ │ • 信息收集与验证 │ │ • 任务分发与协调 │ │ • 结果聚合与呈现 │ └─────┬──────────────────────────┬────────────────────────┘ │ │ ┌─────▼────────┐ ┌──────▼─────────┐ │ 航班代理 │ │ 酒店代理 │ │ (Flight Agent)│ │ (Hotel Agent) │ │ │ │ │ │ • 航班查询 │ │ • 酒店查询 │ │ • 价格计算 │ │ • 房间推荐 │ │ • 时刻匹配 │ │ • 设施展示 │ └──────────────┘ └────────────────┘

技术栈

项目结构

目录树

python_a2a/ ├── .env # 环境变量配置 (API密钥等) ├── .gitignore # Git忽略规则 ├── pyproject.toml # 项目元数据与依赖声明 ├── requirements.txt # 简化的依赖列表 ├── settings.py # 全局配置模块 ├── uv.lock # 依赖锁定文件 (UV包管理器) │ ├── travel_planner_agent/ # 主应用包 │ ├── agent.py # 【核心】主协调员代理定义 │ ├── subagents/ # 子代理模块 │ │ ├── flight_agent.py # 航班查询代理 │ │ └── hotel_agent.py # 酒店推荐代理 │ └── .adk/ │ └── session.db # 会话持久化存储 │ └── .venv/ # Python虚拟环境

关键文件说明

环境配置

1. 安装依赖

使用 UV (推荐，快速)

# 安装 UV (如果尚未安装)curl-LsSf https://astral.sh/uv/install.sh|sh# 同步依赖uvsync# 激活虚拟环境source.venv/bin/activate# Linux/Mac# 或.venv\Scripts\activate# Windows

使用传统 pip

# 创建虚拟环境python3.11 -m venv .venv# 激活虚拟环境source.venv/bin/activate# Linux/Mac# 安装依赖pipinstall-r requirements.txt

2. 配置环境变量

创建.env文件（已包含在项目中）：

# OpenAI 兼容 API 配置OPENAI_API_KEY=your_api_key_hereOPENAI_BASE_URL=https://api.openai.com/v1OPENAI_MODEL=openai/gpt-4# 或使用其他兼容服务 (例如 DeepSeek)OPENAI_API_KEY=sk-xxxxxxxxxxxxOPENAI_BASE_URL=https://api.deepseek.com/v1OPENAI_MODEL=openai/deepseek-chat

注意：.env文件已在.gitignore中，不会提交到代码仓库

3. 验证安装

# test_setup.pyfromgoogle.adk.agentsimportLlmAgentfromgoogle.adk.models.lite_llmimportLiteLlmfromsettingsimportOPENAI_MODELprint(f"✓ ADK 已安装")print(f"✓ 模型配置:{OPENAI_MODEL}")# 测试模型连接model=LiteLlm(model=OPENAI_MODEL)print("✓ 环境配置完成")

核心代码解析

1. 配置管理 (settings.py)

importosfrompathlibimportPathfromdotenvimportload_dotenv# 显式指定 .env 路径，避免路径问题load_dotenv(Path(__file__).resolve().parent/".env")# 导出环境变量OPENAI_BASE_URL=os.getenv("OPENAI_BASE_URL")OPENAI_API_KEY=os.getenv("OPENAI_API_KEY")OPENAI_MODEL=os.getenv("OPENAI_MODEL")

设计原则：

• 单一职责：配置管理独立于业务逻辑
• 路径安全：使用Path对象确保跨平台兼容
• 集中管理：所有环境变量统一加载

2. 子代理实现

航班代理 (flight_agent.py)

fromgoogle.adk.agentsimportLlmAgentfromgoogle.adk.models.lite_llmimportLiteLlmfromsettingsimportOPENAI_MODEL# 初始化模型openai_model=LiteLlm(model=OPENAI_MODEL)# 定义航班代理flight_agent=LlmAgent(name="flight_agent",model=openai_model,description="根据出发地、目的地、旅行日期和预算提供建议的航班。",instruction=""" 你是一个航班预订代理。协调员会给你以下信息： - 出发地 - 目的地 - 起始日期 - 结束日期 - 预算金额 - 预算货币 返回1-2个模拟航班选项，包括： - 航空公司名称 - 出发和返回日期/时间 - 价格（以指定货币表示） - 班级（经济舱/商务舱） 确保总价格在预算范围内。 """)

关键设计：

酒店代理 (hotel_agent.py)

fromgoogle.adk.agentsimportLlmAgentfromgoogle.adk.models.lite_llmimportLiteLlmfromsettingsimportOPENAI_MODEL openai_model=LiteLlm(model=OPENAI_MODEL)hotel_agent=LlmAgent(name="hotel_agent",model=openai_model,description="在给定的旅行日期内找到符合目的地和预算的酒店。",instruction=""" 你是一个酒店预订代理。协调员会给你以下信息： - 目的地 - 起始日期 - 结束日期 - 预算金额 - 预算货币 返回1-2个模拟酒店建议，包括： - 酒店名称 - 每晚价格和总成本（以给定货币表示） - 主要设施 确保总价格符合预算。 """)

3. 主协调员代理 (agent.py)

fromdatetimeimportdatetimefromdotenvimportload_dotenvfromgoogle.adk.agentsimportLlmAgentfromgoogle.adk.models.lite_llmimportLiteLlmfromsettingsimportOPENAI_MODELfromtravel_planner_agent.subagents.flight_agentimportflight_agentfromtravel_planner_agent.subagents.hotel_agentimporthotel_agent load_dotenv()openai_model=LiteLlm(model=OPENAI_MODEL)today=datetime.today().date()# 协调员代理coordinator_agent=LlmAgent(name="TravelCoordinator",model=openai_model,description="主要的协调员代理，收集旅行偏好并查询子代理。",instruction=f""" 你是一个旅行规划协调员。 你的任务是从用户那里收集旅行偏好，并与子代理协调， 提供航班、酒店建议和日程安排。 步骤1：提取信息 - 出发地、目的地 - 起始日期、结束日期 (格式: YYYY-MM-DD) - 预算金额、预算货币 步骤2：处理缺失信息 - 起始日期缺失：建议使用{today}- 结束日期缺失：根据天数计算 - 货币缺失：默认为 "USD" 步骤3：确认信息 向用户确认旅行偏好 步骤4：调用子代理 - `flight_agent` 用于航班建议 - `hotel_agent` 用于酒店建议 步骤5：呈现结果 - 旅行摘要 - 航班建议 - 酒店建议 - 总估算成本 - 一日行程计划示例 """,sub_agents=[flight_agent,hotel_agent]# 🔑 关键：子代理列表)# 导出根代理 (ADK 框架约定)root_agent=coordinator_agent

核心机制：

1. sub_agents参数：声明子代理列表，框架自动处理代理间通信
2. 模板字符串：使用f"{today}"动态插入当前日期
3. 导出约定：必须导出root_agent变量作为入口点

运行方式

方式 1：使用 ADK CLI (推荐)

# 安装 ADK CLI (如果需要)pipinstallgoogle-adk# 启动交互式会话adk start travel_planner_agent# 或者指定配置文件adk start --config travel_planner_agent/.adk/config.json

方式 2：程序化调用

# run_agent.pyimportasynciofromgoogle.adk.agentsimportLlmAgentfromtravel_planner_agent.agentimportroot_agentasyncdefmain():# 用户输入user_message=""" 我想从北京到上海旅行，从 2025-01-15 到 2025-01-20， 预算是 5000 元人民币 """# 调用代理response=awaitroot_agent.arun(user_message)# 输出结果print(response.content)if__name__=="__main__":asyncio.run(main())

方式 3：作为服务运行

# server.pyfromgoogle.adk.serverimportAgentServerfromtravel_planner_agent.agentimportroot_agent# 启动 FastAPI 服务器server=AgentServer(agent=root_agent)server.run(host="0.0.0.0",port=8000)

然后通过 HTTP 调用：

curl-X POST http://localhost:8000/chat\-H"Content-Type: application/json"\-d'{"message": "帮我计划从纽约到洛杉矶的旅行"}'

扩展开发

添加新的子代理

假设要添加一个"租车代理"：

1. 创建新文件travel_planner_agent/subagents/car_rental_agent.py

fromgoogle.adk.agentsimportLlmAgentfromgoogle.adk.models.lite_llmimportLiteLlmfromsettingsimportOPENAI_MODEL openai_model=LiteLlm(model=OPENAI_MODEL)car_rental_agent=LlmAgent(name="car_rental_agent",model=openai_model,description="提供目的地租车服务建议",instruction=""" 你是一个租车代理。你会收到： - 目的地 - 起始日期 - 结束日期 - 预算 返回 1-2 个租车选项，包括： - 租车公司 - 车型 - 每日租金和总价 - 保险选项 """)

2. 修改agent.py，集成新代理

fromtravel_planner_agent.subagents.car_rental_agentimportcar_rental_agent coordinator_agent=LlmAgent(# ... 其他参数不变 ...instruction=""" ... 现有指令 ... 步骤4：调用子代理 - `flight_agent` 用于航班建议 - `hotel_agent` 用于酒店建议 - `car_rental_agent` 用于租车建议 # 新增 步骤5：呈现结果 ... 现有内容 ... - 租车建议 # 新增 """,sub_agents=[flight_agent,hotel_agent,car_rental_agent]# 添加新代理)

集成外部 API

使用工具 (Tools)扩展代理能力：

fromgoogle.adk.toolsimporttool@tooldefsearch_flights(origin:str,destination:str,date:str)->dict:"""调用真实航班 API"""# 示例：使用 requests 调用外部 APIimportrequests response=requests.get("https://api.example.com/flights",params={"origin":origin,"destination":destination,"date":date})returnresponse.json()# 将工具绑定到代理flight_agent=LlmAgent(name="flight_agent",model=openai_model,tools=[search_flights],# 添加工具列表instruction="...")

数据持久化

ADK 自动将会话状态存储在.adk/session.db(SQLite)。

自定义持久化配置：

fromgoogle.adk.storageimportSQLiteStorage# 自定义存储路径storage=SQLiteStorage(db_path="/path/to/custom.db")coordinator_agent=LlmAgent(name="TravelCoordinator",model=openai_model,storage=storage,# 注入自定义存储# ...)

最佳实践

1. 提示词工程

# ✅ 推荐：结构化、清晰的指令instruction=""" 你是航班代理。请按以下步骤操作： 步骤1：解析输入 - 提取出发地、目的地、日期 - 验证日期格式 (YYYY-MM-DD) 步骤2：生成建议 - 返回 2-3 个选项 - 包含价格、时间、航空公司 步骤3：验证预算 - 确保总价不超过预算 """# ❌ 避免：模糊、冗长的指令instruction="你是一个很棒的航班代理，请帮助用户找到最好的航班..."

2. 代理职责分离

# ✅ 推荐：单一职责flight_agent=LlmAgent(name="flight_agent",...)hotel_agent=LlmAgent(name="hotel_agent",...)car_agent=LlmAgent(name="car_agent",...)# ❌ 避免：全能代理super_agent=LlmAgent(name="super_agent",...)# 处理所有任务

3. 错误处理

fromgoogle.adk.agentsimportLlmAgentfromgoogle.adk.exceptionsimportAgentErrortry:response=awaitroot_agent.arun(user_input)exceptAgentErrorase:print(f"代理执行失败:{e}")# 实现回退逻辑或用户通知

依赖项详解

核心依赖

# pyproject.toml [project] name = "python-a2a" version = "0.1.0" requires-python = ">=3.11.4" dependencies = [ "google-adk>=1.21.0", # Agent 框架核心 "pydantic>=2.12.5", # 数据验证 "python-dotenv>=1.2.1", # 环境变量 "rawdog-ai>=0.1.6", # 代码执行能力 ]

依赖关系图

python-a2a ├── google-adk (Agent Development Kit) │ ├── llm-models (LiteLLM 集成) │ ├── storage (SQLite 持久化) │ └── communication (Agent 间消息传递) │ ├── pydantic (数据验证) │ ├── 类型检查 │ └── 序列化/反序列化 │ ├── python-dotenv │ └── .env 文件解析 │ └── rawdog-ai (可选) └── 动态代码执行沙箱

常见问题

Q1: 如何切换 LLM 提供商?

修改.env文件：

# OpenAIOPENAI_MODEL=openai/gpt-4# Anthropic ClaudeOPENAI_MODEL=anthropic/claude-3-sonnetOPENAI_BASE_URL=https://api.anthropic.com# 本地模型 (Ollama)OPENAI_MODEL=ollama/llama2OPENAI_BASE_URL=http://localhost:11434

Q2: 代理间如何传递数据?

ADK 框架自动处理。在instruction中使用变量引用：

instruction=""" 将以下信息传递给 flight_agent: - {origin} # 自动从上下文提取 - {destination} - {budget} """

Q3: 如何调试代理行为?

启用详细日志：

importlogging logging.basicConfig(level=logging.DEBUG)logger=logging.getLogger("google.adk")logger.setLevel(logging.DEBUG)

进阶话题

1. 流式响应

asyncforchunkinroot_agent.astream("帮我规划旅行"):print(chunk.content,end="",flush=True)

2. 并行代理调用

fromgoogle.adk.agentsimportParallelAgent# 并行组parallel_agent=ParallelAgent(name="parallel_search",sub_agents=[flight_agent,hotel_agent],mode="all"# 同时调用所有子代理)

3. 条件路由

fromgoogle.adk.agentsimportConditionalAgent router=ConditionalAgent(name="task_router",rules={"flight":flight_agent,"hotel":hotel_agent},default=coordinator_agent)

资源链接

• Google ADK 文档: https://github.com/google/adk
• LiteLLM 文档: https://docs.litellm.ai/
• Pydantic 文档: https://docs.pydantic.dev/

总结

A2A 框架通过代理协作模式简化了复杂 AI 应用的开发：

通过本指南，你应该能够：

• ✅ 理解 A2A 架构设计
• ✅ 创建自己的多代理系统
• ✅ 集成外部 API 和工具
• ✅ 部署到生产环境