1. 先搞清楚这个项目到底解决什么实际问题如果你正在研究AI Agent开发特别是多智能体协作场景这个手撕AI Agent框架项目值得重点关注。它不是一个简单的Demo演示而是完整展示了从单智能体到多智能体系统的全链路实现。核心解决的是生产级AI Agent系统的架构问题如何让不同的AI Agent既能各司其职又能有效协作。传统的单智能体框架在处理复杂业务时往往力不从心而这个项目通过A2A协议和MCP工具箱的集成展示了真正的分布式智能体协作方案。最实际的价值在于用DeepSeek就能跑通。很多多智能体项目对硬件要求极高但这个方案在普通开发环境下就能验证特别适合想要深入理解Agent架构的开发者。2. 环境准备DeepSeek 基础开发环境就能开始2.1 硬件和软件要求这个项目的优势在于对硬件要求相对友好最低配置CPU: 4核以上Intel i5或同等性能内存: 16GB8GB勉强可跑但体验较差存储: 至少20GB可用空间网络: 稳定互联网连接用于模型调用推荐配置CPU: 8核以上内存: 32GBGPU: 可选非必须项目主要使用云端LLM软件环境操作系统: Ubuntu 20.04 / macOS 12 / Windows 10WSL2推荐Python: 3.9-3.11包管理: uvRust编写的快速Python包管理器版本控制: Git2.2 DeepSeek API配置项目使用DeepSeek作为主要LLM需要提前准备# 获取DeepSeek API Key # 访问DeepSeek官网注册账号并获取API密钥 # 设置环境变量 export DEEPSEEK_API_KEYyour_actual_api_key_here export DEEPSEEK_BASE_URLhttps://api.deepseek.com重要提醒先申请API Key再开始实验避免中途中断。DeepSeek通常有新用户免费额度足够完成这个项目的所有实验。2.3 项目结构预览在开始编码前先了解整个项目的模块划分adk-a2a-agent-runtime-starter/ ├── reservation_agent/ # 预订智能体A2A服务 │ ├── agent.py # 核心智能体逻辑 │ ├── a2a_config.py # A2A协议配置 │ └── executor.py # A2A执行器 ├── restaurant_agent/ # 餐厅主智能体 │ └── agent.py # 编排器智能体 ├── toolbox_adk/ # MCP工具箱集成 ├── scripts/ # 部署和测试脚本 └── tools.yaml # 工具配置这种模块化设计让每个智能体职责清晰便于单独测试和部署。3. 从单智能体开始预订Agent的完整实现3.1 创建基础预订智能体先实现最核心的预订功能这是整个多智能体系统的基础服务# reservation_agent/agent.py import os from typing import Any from google.adk.agents import LlmAgent from google.adk.models.google_llm import Gemini from google.adk.tools import ToolContext # 使用应用级状态前缀确保预订数据跨会话持久化 STATE_PREFIX app:reservation: def create_reservation( phone_number: str, name: str, party_size: int, date: str, time: str, tool_context: ToolContext, ) - dict: 创建新的餐厅预订 reservation { name: name, party_size: party_size, date: date, time: time, status: confirmed, } # 使用电话号码作为唯一标识符存储预订 tool_context.state[f{STATE_PREFIX}{phone_number}] reservation return { status: confirmed, message: fReservation created for {name}, party of {party_size} on {date} at {time}., } def check_reservation(phone_number: str, tool_context: ToolContext) - dict: 通过电话号码查询现有预订 reservation tool_context.state.get(f{STATE_PREFIX}{phone_number}) if reservation: return {found: True, reservation: reservation} return {found: False, message: fNo reservation found for {phone_number}.} def cancel_reservation(phone_number: str, tool_context: ToolContext) - dict: 取消现有预订 key f{STATE_PREFIX}{phone_number} reservation tool_context.state.get(key) if not reservation: return {success: False, message: fNo reservation found for {phone_number}.} reservation[status] cancelled tool_context.state[key] reservation return {success: True, message: fReservation for {reservation[name]} has been cancelled.} # 创建主智能体 root_agent LlmAgent( namereservation_agent, modelgemini-3.5-flash, # 可替换为DeepSeek模型 instruction你是Foodie Finds餐厅的友好预订助手。 帮助食客创建、查询和取消餐桌预订。 创建预订时收集姓名、电话号码、人数、日期、时间。 在创建前始终确认详细信息。, tools[create_reservation, check_reservation, cancel_reservation], )关键设计要点使用STATE_PREFIX确保状态隔离避免不同应用间的数据冲突电话号码作为自然键既唯一又便于用户记忆每个工具函数都返回结构化数据便于后续处理3.2 添加A2A协议支持单智能体完成后需要让它能够被其他智能体发现和调用# reservation_agent/a2a_config.py from a2a.types import AgentSkill from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card reservation_skill AgentSkill( idmanage_reservations, nameRestaurant Reservations, description在Foodie Finds餐厅创建、查询和取消餐桌预订, tags[reservations, restaurant, booking], examples[ 为4人预订周五晚上7点的餐桌, 查询电话号码555-0101的预订, 取消我的预订电话号码555-0101, ], input_modes[text/plain], output_modes[text/plain], ) agent_card create_agent_card( agent_nameReservation Agent, description处理餐厅餐桌预订——为Foodie Finds餐厅创建、查询和取消预订。, skills[reservation_skill], )A2A协议的核心价值智能体卡片像名片一样描述智能体能力其他智能体可以自动发现标准化通信不同框架的智能体可以通过A2A协议协作状态保持支持多轮对话而不仅仅是单次函数调用3.3 实现A2A执行器执行器是A2A协议和ADK智能体之间的桥梁# reservation_agent/executor.py import os from typing import NoReturn from a2a.server.agent_execution import AgentExecutor, RequestContext from a2a.server.events import EventQueue from a2a.server.tasks import TaskUpdater from a2a.types import TaskState, TextPart from google.adk.artifacts import InMemoryArtifactService from google.adk.memory.in_memory_memory_service import InMemoryMemoryService from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService, VertexAiSessionService from google.genai import types from reservation_agent.agent import root_agent as reservation_agent class ReservationAgentExecutor(AgentExecutor): A2A协议和ADK预订智能体之间的桥梁 def __init__(self) - None: self.agent None self.runner None def _init_agent(self) - None: if self.agent is not None: return self.agent reservation_agent # 自动检测环境本地测试使用内存会话部署后使用持久会话 engine_id os.environ.get(GOOGLE_CLOUD_AGENT_ENGINE_ID) if engine_id: # 生产环境使用VertexAiSessionService实现持久化 session_service VertexAiSessionService( projectos.environ.get(GOOGLE_CLOUD_PROJECT), locationos.environ.get(GOOGLE_CLOUD_LOCATION, us-central1), agent_engine_idengine_id, ) app_name engine_id else: # 开发环境使用内存会话 session_service InMemorySessionService() app_name self.agent.name self.runner Runner( app_nameapp_name, agentself.agent, artifact_serviceInMemoryArtifactService(), session_servicesession_service, memory_serviceInMemoryMemoryService(), ) async def execute(self, context: RequestContext, event_queue: EventQueue) - None: 处理A2A请求的核心方法 if self.agent is None: self._init_agent() query context.get_user_input() updater TaskUpdater(event_queue, context.task_id, context.context_id) user_id context.message.metadata.get(user_id, a2a-user) if context.message.metadata else a2a-user if not context.current_task: await updater.submit() await updater.start_work() try: session await self._get_or_create_session(context.context_id, user_id) content types.Content(roleuser, parts[types.Part(textquery)]) # 通过ADK运行器处理请求 async for event in self.runner.run_async( session_idsession.id, user_iduser_id, new_messagecontent, ): if event.is_final_response(): parts event.content.parts answer .join(p.text for p in parts if p.text) or No response. await updater.add_artifact([TextPart(textanswer)], nameanswer) await updater.complete() break except Exception as e: await updater.update_status( TaskState.failed, messagefError: {str(e)}, ) raise执行器的关键设计环境自适应自动识别开发/生产环境使用不同的会话服务错误处理完善的异常捕获和状态更新机制异步支持使用async/await确保高并发场景下的性能4. 本地测试确保单智能体稳定运行4.1 编写本地测试脚本在投入多智能体集成前先确保单智能体工作正常# scripts/test_a2a_agent_local.py import asyncio import json import os from dotenv import load_dotenv from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor load_dotenv() async def test_reservation_workflow(): 测试完整的预订工作流 # 创建本地A2A智能体 a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor ) a2a_agent.set_up() print( 测试1: 创建预订 ) # 模拟创建预订请求 create_result await simulate_a2a_request( a2a_agent, Book a table for 2 on Saturday at 6pm. Name: Bob, Phone: 555-0202 ) print(f创建结果: {create_result}) print( 测试2: 查询预订 ) check_result await simulate_a2a_request( a2a_agent, Check the reservation for 555-0202 ) print(f查询结果: {check_result}) print( 测试3: 取消预订 ) cancel_result await simulate_a2a_request( a2a_agent, Cancel the reservation for 555-0202 ) print(f取消结果: {cancel_result}) async def simulate_a2a_request(agent, message_text): 模拟A2A协议请求 # 这里简化了实际的A2A协议调用 # 实际实现需要构建完整的HTTP请求模拟 pass if __name__ __main__: asyncio.run(test_reservation_workflow())4.2 运行测试并排查常见问题执行测试命令# 设置Python路径并运行测试 PYTHONPATH. uv run python scripts/test_a2a_agent_local.py常见问题及解决方案问题1: 模块导入错误ModuleNotFoundError: No module named a2a解决确保安装了正确的依赖版本uv add a2a-sdk0.3.26 google-cloud-aiplatform[agent_engines,adk]1.149.0问题2: 环境变量缺失KeyError: GOOGLE_CLOUD_PROJECT解决检查.env文件配置# 确保.env文件存在且包含必要变量 cat .env # 应该包含GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION等问题3: 会话状态不持久每次请求都丢失之前的预订数据解决检查STATE_PREFIX配置和会话服务初始化确认使用应用级状态前缀验证会话服务正确初始化5. 多智能体集成餐厅主Agent的编排逻辑5.1 实现主协调智能体预订智能体稳定后开始构建协调多个智能体的主Agent# restaurant_agent/agent.py import os import httpx from google.adk.agents import LlmAgent from google.adk.agents.remote_a2a_agent import RemoteA2aAgent from google.auth import default from google.auth.transport.requests import Request as AuthRequest from toolbox_adk import ToolboxToolset # 配置工具和服务端点 TOOLBOX_URL os.environ.get(TOOLBOX_URL, http://127.0.0.1:5000) RESERVATION_AGENT_CARD_URL os.environ.get(RESERVATION_AGENT_CARD_URL, ) # MCP工具箱集成菜单查询功能 toolbox ToolboxToolset(TOOLBOX_URL) class GoogleCloudAuth(httpx.Auth): Google Cloud自动认证处理 def __init__(self): self.credentials, _ default( scopes[https://www.googleapis.com/auth/cloud-platform] ) def auth_flow(self, request): # 令牌过期前自动刷新 if not self.credentials.valid: self.credentials.refresh(AuthRequest()) request.headers[Authorization] fBearer {self.credentials.token} yield request # 创建远程A2A智能体代理 reservation_remote_agent RemoteA2aAgent( namereservation_agent, description处理餐厅餐桌预订——创建、查询和取消预订。当用户想要预订、查询或取消餐桌时委托给此智能体。, agent_cardRESERVATION_AGENT_CARD_URL, httpx_clienthttpx.AsyncClient(authGoogleCloudAuth(), timeout60), ) # 主餐厅智能体 root_agent LlmAgent( namerestaurant_agent, modelgemini-3.5-flash, # 可替换为DeepSeek instruction你是Foodie Finds餐厅友好且知识丰富的礼宾员。 职责 - 帮助食客按类别或菜系浏览菜单 - 提供特定菜品的完整详细信息包括成分、价格和饮食信息 - 基于食客渴望的自然语言描述推荐菜品 - 询问时添加新菜单项 - 对于预订请求预订、查询或取消餐桌委托给reservation_agent 当食客按名称或菜系询问特定菜品时使用get-item-details工具。 当食客要求特定类别或菜系类型时使用search-menu工具。 当食客描述他们想要的食物类型——通过风味、质地、饮食需求或渴望——使用search-menu-by-description工具进行语义搜索。 在search-menu和search-menu-by-description之间不确定时优先使用search-menu-by-description——它搜索菜品描述并找到更相关的匹配项。 如果菜品不可用available为false告知食客并从搜索结果中建议类似的替代品。 保持对话性、知识性和简洁性。, tools[toolbox], sub_agents[reservation_remote_agent], )5.2 智能体协作的关键机制委托决策逻辑主智能体需要智能地判断何时使用本地工具何时委托给子智能体# 在实际的智能体推理过程中LLM会根据以下模式决策 用户输入分析模式 1. 包含reservation, book, cancel, check booking等关键词 → 委托给预订智能体 2. 菜单查询、菜品推荐、饮食需求 → 使用本地MCP工具箱 3. 混合请求如我想订位并了解意大利菜 → 分解任务分别处理 会话保持机制多智能体协作需要保持会话上下文使用相同的context_id确保会话连续性主智能体维护全局对话状态子智能体执行结果返回到主会话6. 部署实战从本地到生产环境6.1 本地集成测试在部署到生产环境前先在本地完成端到端测试# 启动MCP工具箱服务如果尚未运行 ./toolbox --configtools.yaml logs/toolbox.log 21 # 启动ADK Web界面进行交互测试 uv run adk web --port 8080 --allow_origins regex:.*测试用例验证用例1: 纯菜单查询用户你们有什么意大利菜 预期主智能体直接使用MCP工具箱返回结果用例2: 纯预订请求用户我想为4人预订周五晚上7点的餐桌 预期主智能体委托给预订智能体开始多轮对话收集信息用例3: 混合请求用户推荐一些辣菜然后帮我订位 预期先处理菜单推荐再切换到预订流程6.2 生产环境部署配置预订智能体部署A2A服务# scripts/deploy_a2a_agent_runtime.py import os from pathlib import Path import vertexai from dotenv import load_dotenv from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor load_dotenv() def deploy_reservation_agent(): 部署预订智能体到Agent Runtime project_id os.environ[GOOGLE_CLOUD_PROJECT] region os.environ[REGION] vertexai.init(projectproject_id, locationregion) a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor, ) client vertexai.Client(projectproject_id, locationregion) print(开始部署预订智能体...) remote_agent client.agent_engines.create( agenta2a_agent, config{ display_name: agent_card.name, description: agent_card.description, requirements: [ google-cloud-aiplatform[agent_engines,adk]1.149.0, a2a-sdk0.3.26, google-adk1.29.0, ], extra_packages: [./reservation_agent], }, ) # 保存部署信息 resource_name remote_agent.api_resource.name env_path Path(.env) # 更新环境变量 lines env_path.read_text().splitlines() if env_path.exists() else [] lines [l for l in lines if not l.startswith(RESERVATION_AGENT_RESOURCE_NAME)] lines.append(fRESERVATION_AGENT_RESOURCE_NAME{resource_name}) env_path.write_text(\n.join(lines) \n) print(f部署完成资源名称: {resource_name})主餐厅智能体部署Cloud Run# 部署主智能体到Cloud Run gcloud run deploy restaurant-agent \ --source . \ --region$REGION \ --allow-unauthenticated \ --update-env-varsRESERVATION_AGENT_CARD_URL$RESERVATION_AGENT_CARD_URL \ --min-instances0 \ --max-instances1 \ --memory1Gi6.3 权限和网络配置服务账号权限# 授予Cloud Run服务账号调用Agent Runtime的权限 PROJECT_NUMBER$(gcloud projects describe $GOOGLE_CLOUD_PROJECT --formatvalue(projectNumber)) gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \ --memberserviceAccount:${PROJECT_NUMBER}-computedeveloper.gserviceaccount.com \ --roleroles/aiplatform.user网络连通性验证# 测试服务间通信 curl -X POST ${AGENT_URL}/api/chat \ -H Content-Type: application/json \ -d {message: 测试连通性}7. 生产级考量性能、监控和扩展7.1 性能优化策略智能体响应时间优化设置合理的超时时间A2A调用建议30-60秒实现请求缓存避免重复计算使用异步处理提高并发能力资源使用监控# 添加性能监控装饰器 import time from functools import wraps def monitor_performance(func): wraps(func) async def wrapper(*args, **kwargs): start_time time.time() try: result await func(*args, **kwargs) duration time.time() - start_time # 记录到监控系统 if duration 10: # 超过10秒记录警告 print(f警告: {func.__name__} 执行时间 {duration:.2f}秒) return result except Exception as e: duration time.time() - start_time print(f错误: {func.__name__} 在 {duration:.2f}秒后失败: {str(e)}) raise return wrapper7.2 错误处理和重试机制智能A2A调用重试class ResilientA2aClient: 具有重试机制的A2A客户端 def __init__(self, max_retries3, backoff_factor1.0): self.max_retries max_retries self.backoff_factor backoff_factor async def call_with_retry(self, agent, message, context_idNone): 带重试的A2A调用 for attempt in range(self.max_retries): try: response await agent.on_message_send( messageIdfmsg-{attempt}, roleuser, parts[{kind: text, text: message}], contextIdcontext_id ) return response except Exception as e: if attempt self.max_retries - 1: raise # 指数退避 await asyncio.sleep(self.backoff_factor * (2 ** attempt))7.3 扩展性设计多智能体架构的扩展模式垂直扩展为现有智能体添加更多工具和能力水平扩展增加新的专业智能体如支付智能体、配送智能体层次扩展引入更高级别的协调智能体管理多个领域智能体配置化智能体发现# agents_config.yaml agents: reservation: card_url: ${RESERVATION_AGENT_CARD_URL} description: 处理餐厅预订 timeout: 60 menu: card_url: ${MENU_AGENT_CARD_URL} description: 处理菜单查询 timeout: 30 payment: card_url: ${PAYMENT_AGENT_CARD_URL} description: 处理支付流程 timeout: 1208. 实际踩坑经验和排查指南8.1 常见问题快速排查问题1: A2A调用返回401未授权症状智能体协作时出现认证错误 排查步骤 1. 检查服务账号权限确保有aiplatform.user角色 2. 验证令牌刷新确认GoogleCloudAuth正确实现 3. 检查环境变量确认RESERVATION_AGENT_CARD_URL设置正确问题2: 会话状态不保持症状多次调用间丢失对话上下文 排查步骤 1. 检查context_id传递确保每次调用使用相同的context_id 2. 验证会话服务确认使用VertexAiSessionService而非内存会话 3. 检查状态前缀确认STATE_PREFIX正确配置问题3: 智能体委托决策错误症状主智能体错误地将请求委托给子智能体或反之 排查步骤 1. 检查指令提示优化主智能体的instruction文本 2. 验证工具描述确保每个工具的功能描述准确 3. 测试边界案例用混合请求测试委托逻辑8.2 调试技巧和工具启用详细日志import logging logging.basicConfig(levellogging.DEBUG) # 在关键位置添加调试日志 logger logging.getLogger(__name__) async def execute(self, context: RequestContext, event_queue: EventQueue): logger.debug(f收到A2A请求: {context.get_user_input()}) # ... 处理逻辑交互式测试方法# 使用adk web进行实时调试 uv run adk web --port 8080 # 观察智能体的思考过程了解委托决策8.3 性能调优建议针对DeepSeek的优化批量处理将多个相关请求合并为单个调用缓存策略对频繁查询的结果进行缓存超时设置根据操作复杂度设置合理的超时时间并发控制限制同时进行的A2A调用数量这个多智能体框架的真正价值在于它提供了一种可扩展的AI Agent架构模式。一旦理解了A2A协议和MCP工具箱的集成原理你可以将其应用到各种复杂业务场景中而不仅仅是餐厅预订这个示例。
)
)
)
