如果你还在为AI Agent开发感到困惑觉得多智能体协作、MCP协议、A2A架构这些概念听起来高大上但实际操作无从下手那么这篇文章就是为你准备的。2026年已经到来AI Agent技术正在从单兵作战向多智能体协作演进但很多开发者仍然停留在“对话式AI”的认知层面无法构建真正具备专业分工的智能体系统。本文将通过一个完整的餐厅预订系统实战案例带你从零构建一个包含菜单查询代理和预订代理的多智能体系统。你将学会如何让不同专业能力的AI Agent通过MCP协议访问工具和数据通过A2A协议进行智能体间通信并最终部署到生产环境。最重要的是本文提供的代码完全兼容DeepSeek模型让你可以用更经济的成本实现相同的功能。1. 为什么单智能体已经不够用了在传统的AI应用开发中我们通常构建一个“全能型”智能体来处理所有任务。比如一个餐厅助手既要能查询菜单又要能处理预订还要回答各种咨询问题。这种架构在简单场景下还能应付但随着业务复杂度增加问题就暴露出来了单智能体的局限性功能臃肿一个智能体需要掌握所有领域的知识模型容易产生混淆更新困难修改预订逻辑可能影响菜单查询功能风险集中资源浪费简单的菜单查询也要动用复杂的预订处理能力扩展性差新增功能需要重新训练或大幅调整现有智能体多智能体的优势专业分工每个智能体专注特定领域表现更专业独立部署不同智能体可以有不同的更新周期和部署策略资源优化根据任务复杂度分配合适的智能体易于扩展新增功能只需添加新的专业智能体在实际项目中我们采用的多智能体架构是这样的一个“餐厅礼宾代理”作为总协调员负责理解用户意图并将任务分发给专业的“菜单查询代理”通过MCP协议和“预订代理”通过A2A协议。2. 核心概念解析MCP vs A2A vs 传统技能2.1 MCPModel Context Protocol智能体与工具的桥梁MCP的核心作用是让AI智能体能够安全、标准化地使用外部工具和数据源。可以把MCP理解为智能体的“手和眼睛”——它让智能体能够操作工具、获取信息。# MCP工具示例菜单查询工具 class MenuQueryTool: def search_by_category(self, category: str) - List[MenuItem]: 根据分类查询菜单 pass def search_by_description(self, description: str) - List[MenuItem]: 语义搜索菜单项 pass2.2 A2AAgent-to-Agent智能体间的协作协议A2A协议解决了不同智能体之间如何发现彼此、如何通信、如何协作的问题。与MCP主要连接工具不同A2A连接的是其他智能体。关键区别MCP工具无状态单一功能执行完即结束A2A代理有状态具备推理能力支持多轮对话2.3 智能体卡片Agent Card智能体的“名片”每个通过A2A协议暴露的智能体都需要一个标准化的描述文件这就是智能体卡片。它包含智能体的功能描述、技能列表、输入输出格式等信息。{ agent_name: Reservation Agent, description: 处理餐厅预订创建、查询和取消, skills: [ { id: manage_reservations, name: 餐厅预订管理, examples: [ 预订4人周五晚上7点的座位, 查询电话号码555-0101的预订 ] } ] }3. 环境准备与项目结构3.1 开发环境要求Python版本3.9或更高版本包管理uv快速Python包管理器AI框架Google ADKAgent Development Kit模型支持Gemini系列或DeepSeek本文兼容3.2 项目目录结构adk-a2a-agent-runtime-starter/ ├── reservation_agent/ # 预订代理 │ ├── agent.py # 代理核心逻辑 │ ├── a2a_config.py # A2A配置和智能体卡片 │ └── executor.py # A2A执行器 ├── restaurant_agent/ # 餐厅总代理 │ └── agent.py # 集成了预订代理的餐厅代理 ├── tools.yaml # MCP工具箱配置 ├── scripts/ # 部署和测试脚本 └── .env # 环境配置3.3 初始环境设置# 克隆项目代码 git clone https://github.com/alphinside/adk-a2a-agent-runtime-starter.git cd adk-a2a-agent-runtime-starter # 使用uv安装依赖uv比pip快10倍以上 uv sync # 环境配置 cp .env.example .env # 编辑.env文件配置项目ID和区域4. 构建专业预订代理4.1 预订代理的核心功能设计我们的预订代理需要处理三种核心操作创建预订收集用户信息并创建新的预订记录查询预订根据电话号码查询现有预订取消预订取消已存在的预订# reservation_agent/agent.py import os from google.adk.agents import LlmAgent 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: Reservation cancelled} # 创建主要的LLM代理 root_agent LlmAgent( namereservation_agent, modelgemini-3.5-flash, # 可替换为DeepSeek模型 instruction你是Foodie Finds餐厅的友好预订助手。 帮助食客创建、查询和取消餐桌预订。 创建预订时需要收集以下信息 - 预订姓名 - 电话号码作为预订ID - 人数 - 日期 - 时间 在创建前始终确认详细信息。 查询或取消时如果未提供电话号码请询问。 保持简洁专业。, tools[create_reservation, check_reservation, cancel_reservation], )4.2 状态管理的设计考量在AI Agent开发中状态管理是一个关键问题。我们采用了以下几种策略会话状态在同一对话中保持上下文应用状态跨会话持久化数据如预订记录临时状态单次请求的临时数据对于预订数据我们使用应用级状态确保即使对话结束预订信息仍然保留。这种设计比使用数据库更轻量适合中小型应用。5. 配置A2A协议支持5.1 创建智能体卡片智能体卡片是其他智能体发现和理解我们预订代理功能的途径。# 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, name餐厅预订管理, 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_name预订代理, description处理餐厅餐桌预订——为Foodie Finds餐厅创建、查询和取消预订, skills[reservation_skill], )5.2 实现A2A执行器执行器是A2A协议与ADK代理之间的桥梁负责处理传入的A2A请求并通过我们的代理执行。# reservation_agent/executor.py import os import vertexai from a2a.server.agent_execution import AgentExecutor, RequestContext from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService, VertexAiSessionService class ReservationAgentExecutor(AgentExecutor): 连接A2A协议和ADK预订代理的桥梁 def __init__(self) - None: self.agent None self.runner None def _init_agent(self): if self.agent is not None: return from reservation_agent.agent import root_agent as reservation_agent self.agent reservation_agent # 根据环境选择会话服务 engine_id os.environ.get(GOOGLE_CLOUD_AGENT_ENGINE_ID) if engine_id: # 生产环境使用持久化会话服务 project os.environ.get(GOOGLE_CLOUD_PROJECT) location os.environ.get(GOOGLE_CLOUD_LOCATION, us-central1) vertexai.init(projectproject, locationlocation) session_service VertexAiSessionService( projectproject, locationlocation, 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, session_servicesession_service, # 其他服务配置... )这种设计的关键优势在于环境自适应——在本地开发时使用轻量级的内存存储部署到生产环境时自动切换为持久化存储。6. 本地测试与验证6.1 完整的本地测试流程在部署之前我们需要确保A2A代理在本地正常工作。以下是完整的测试脚本# scripts/test_a2a_agent_local.py import asyncio import os from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor async def main(): # 创建本地A2A代理 a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor ) a2a_agent.set_up() # 测试1获取智能体卡片 print( 测试智能体卡片 ) card_response await a2a_agent.handle_authenticated_agent_card() print(f代理名称: {card_response.get(name)}) print(f技能列表: {[s.get(name) for s in card_response.get(skills, [])]}) # 测试2创建预订 print(\n 测试创建预订 ) message_data { message: { messageId: test-001, content: [{text: 预订2人周六晚上6点的座位姓名Bob电话555-0202}], role: ROLE_USER, }, } response await a2a_agent.on_message_send(**message_data) task_id response[task][id] print(f任务ID: {task_id}) # 等待任务完成并获取结果 result await wait_for_task_completion(a2a_agent, task_id) print(f结果: {result}) if __name__ __main__: asyncio.run(main())6.2 测试结果验证运行测试脚本后你应该看到类似以下的输出 测试智能体卡片 代理名称: 预订代理 技能列表: [餐厅预订管理] 测试创建预订 任务ID: f7f7004d-cfea-49c2-b57d-5bca9959e193 结果: 您的预订已确认Bob2人周六晚上6点电话555-0202这个测试验证了A2A代理能够正确暴露其功能描述代理能够处理预订请求并返回正确结果整个A2A协议流程正常工作7. 部署到Agent运行时环境7.1 部署脚本实现将代理部署到生产环境需要处理依赖管理、环境配置和部署流程。# scripts/deploy_a2a_agent_runtime.py import os from pathlib import Path import vertexai from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor def deploy_reservation_agent(): 部署预订代理到Agent运行时 # 环境配置 project_id os.environ[GOOGLE_CLOUD_PROJECT] region os.environ[REGION] staging_bucket os.environ.get(STAGING_BUCKET) vertexai.init(projectproject_id, locationregion, staging_bucketstaging_bucket) # 创建A2A代理实例 a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor, ) print(开始部署预订代理到Agent运行时...) print(预计需要3-5分钟...) # 执行部署 client vertexai.Client(projectproject_id, locationregion) 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], }, ) print(f部署完成资源名称: {remote_agent.api_resource.name}) return remote_agent.api_resource.name7.2 部署后的验证部署完成后我们需要验证代理在生产环境的正常运行# scripts/test_deployed_agent.py import asyncio import vertexai from dotenv import load_dotenv load_dotenv() async def test_deployed_agent(): 测试已部署的代理 project_id os.environ[GOOGLE_CLOUD_PROJECT] region os.environ[REGION] resource_name os.environ[RESERVATION_AGENT_RESOURCE_NAME] vertexai.init(projectproject_id, locationregion) client vertexai.Client(projectproject_id, locationregion) # 获取已部署的代理 agent client.agent_engines.get(nameresource_name) # 测试智能体卡片 card await agent.handle_authenticated_agent_card() print(f已部署代理: {card.name}) print(f访问URL: {card.url}) # 测试预订功能 response await agent.on_message_send( messageIdprod-test-001, roleuser, parts[{kind: text, text: 预订3人周日下午12点的座位}] ) print(生产环境测试通过) if __name__ __main__: asyncio.run(test_deployed_agent())8. 集成多智能体系统8.1 升级餐厅主代理现在我们需要将预订代理集成到餐厅主代理中实现真正的多智能体协作。# 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 toolbox_adk import ToolboxToolset 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(httpx.Request(POST, https://oauth2.googleapis.com/token)) request.headers[Authorization] fBearer {self.credentials.token} yield request # 配置工具和代理 TOOLBOX_URL os.environ.get(TOOLBOX_URL, http://127.0.0.1:5000) RESERVATION_AGENT_CARD_URL os.environ.get(RESERVATION_AGENT_CARD_URL, ) toolbox ToolboxToolset(TOOLBOX_URL) # 创建远程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, instruction你是Foodie Finds餐厅的知识渊博的礼宾员。 你的工作 - 帮助食客按分类或菜系浏览菜单 - 提供菜品的完整详细信息包括成分、价格和饮食信息 - 根据食客渴望的描述推荐菜品 - 当用户想要预订、查询或取消餐桌时委托给reservation_agent 当食客按名称或菜系询问特定菜品时使用get-item-details工具。 当食客要求特定分类或菜系类型时使用search-menu工具。 当食客描述他们想要的食物类型时使用search-menu-by-description工具进行语义搜索。 对于预订请求预订、查询或取消餐桌委托给reservation_agent。 保持对话性、知识渊博且简洁。, tools[toolbox], sub_agents[reservation_remote_agent], )8.2 多智能体协作流程这个集成实现了以下协作流程用户请求分析主代理首先分析用户意图任务路由如果是菜单相关请求路由到MCP工具箱如果是预订相关请求路由到A2A预订代理结果整合主代理将子代理的结果整合后返回给用户9. 完整系统测试与验证9.1 端到端测试场景我们需要测试整个多智能体系统的各种使用场景# scripts/test_full_system.py import asyncio import requests async def test_full_system(): 测试完整的多智能体系统 agent_url os.environ.get(AGENT_URL, http://localhost:8080) test_scenarios [ { name: 菜单查询测试, message: 你们有哪些意大利菜, expected_keywords: [pasta, pizza, 意大利] }, { name: 预订创建测试, message: 我想预订周五晚上7点4个人的座位姓名Eve电话555-0505, expected_keywords: [确认, 预订, Eve] }, { name: 预订查询测试, message: 查询电话555-0505的预订, expected_keywords: [预订, Eve, 周五] } ] for scenario in test_scenarios: print(f\n {scenario[name]} ) # 发送请求到代理 response requests.post( f{agent_url}/api/chat, json{message: scenario[message]} ) if response.status_code 200: result response.json() print(f响应: {result[response]}) # 验证响应包含预期关键词 for keyword in scenario[expected_keywords]: if keyword in result[response]: print(f✓ 包含预期关键词: {keyword}) else: print(f✗ 缺少预期关键词: {keyword}) else: print(f请求失败: {response.status_code})9.2 性能与稳定性考量在生产环境中我们还需要考虑错误处理网络超时、代理不可用等情况重试机制临时故障的自动恢复限流控制防止单个用户过度使用资源监控告警实时监控系统健康状态10. 生产环境最佳实践10.1 安全配置# 生产环境安全配置示例 security: # API密钥管理 api_keys: rotation_period: 90 days emergency_rotation: enabled # 访问控制 access_control: ip_whitelist: [10.0.0.0/8] rate_limiting: 100 requests/minute # 数据保护 data_protection: encryption: enabled retention_period: 30 days10.2 监控与日志建立完整的可观测性体系# 监控配置示例 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider # 设置分布式追踪 tracer_provider TracerProvider() trace.set_tracer_provider(tracer_provider) # 关键指标监控 monitoring_metrics [ agent_response_time, a2a_request_success_rate, concurrent_users, error_rate_by_agent ]10.3 成本优化策略特别是使用DeepSeek等成本较低的模型时缓存策略对常见查询结果进行缓存请求批处理合并多个小请求模型选择根据任务复杂度选择合适的模型规格自动扩缩根据负载动态调整资源11. 常见问题与解决方案11.1 部署问题排查问题现象可能原因解决方案部署超时网络问题或依赖下载慢检查网络连接使用国内镜像源依赖冲突版本不兼容使用uv管理依赖确保版本一致权限错误服务账号权限不足配置正确的IAM角色11.2 运行时问题问题现象可能原因解决方案A2A调用失败身份验证令牌过期实现自动令牌刷新机制响应缓慢模型推理时间过长优化提示词启用流式响应状态丢失会话配置错误检查会话服务配置11.3 模型兼容性问题由于我们设计时考虑了多模型兼容性从Gemini切换到DeepSeek只需要修改配置# 修改模型配置 # 从Gemini切换到DeepSeek model_config { gemini: gemini-3.5-flash, deepseek: deepseek-chat # 根据实际模型名称调整 } # 在agent初始化时选择模型 selected_model model_config[os.environ.get(MODEL_PROVIDER, deepseek)]12. 扩展与进阶方向12.1 添加新的专业智能体基于这个架构你可以轻松添加新的智能体支付处理代理处理订单支付和退款推荐代理基于用户历史提供个性化推荐客服代理处理投诉和复杂咨询12.2 性能优化建议异步处理对耗时操作使用异步模式连接池重用数据库和API连接预加载提前加载常用数据到内存压缩传输对大量数据使用压缩算法12.3 企业级特性对于大型企业应用可以考虑多租户支持隔离不同客户的数据和配置审计日志记录所有操作用于合规审查备份恢复定期备份智能体状态和配置蓝绿部署实现零停机更新通过本文的实战指南你应该已经掌握了构建生产级多智能体系统的核心技能。从单智能体到多智能体协作的转变不仅仅是技术架构的升级更是开发思维的转变。这种架构为构建复杂、可扩展的AI应用提供了坚实的基础。记住最好的学习方式是实践。建议你按照本文的步骤亲手实现这个系统然后在基础上进行扩展和优化。随着经验的积累你将能够设计出更加复杂和强大的多智能体解决方案。
AI Agent多智能体协作实战：基于MCP与A2A协议的餐厅预订系统开发
如果你还在为AI Agent开发感到困惑觉得多智能体协作、MCP协议、A2A架构这些概念听起来高大上但实际操作无从下手那么这篇文章就是为你准备的。2026年已经到来AI Agent技术正在从单兵作战向多智能体协作演进但很多开发者仍然停留在“对话式AI”的认知层面无法构建真正具备专业分工的智能体系统。本文将通过一个完整的餐厅预订系统实战案例带你从零构建一个包含菜单查询代理和预订代理的多智能体系统。你将学会如何让不同专业能力的AI Agent通过MCP协议访问工具和数据通过A2A协议进行智能体间通信并最终部署到生产环境。最重要的是本文提供的代码完全兼容DeepSeek模型让你可以用更经济的成本实现相同的功能。1. 为什么单智能体已经不够用了在传统的AI应用开发中我们通常构建一个“全能型”智能体来处理所有任务。比如一个餐厅助手既要能查询菜单又要能处理预订还要回答各种咨询问题。这种架构在简单场景下还能应付但随着业务复杂度增加问题就暴露出来了单智能体的局限性功能臃肿一个智能体需要掌握所有领域的知识模型容易产生混淆更新困难修改预订逻辑可能影响菜单查询功能风险集中资源浪费简单的菜单查询也要动用复杂的预订处理能力扩展性差新增功能需要重新训练或大幅调整现有智能体多智能体的优势专业分工每个智能体专注特定领域表现更专业独立部署不同智能体可以有不同的更新周期和部署策略资源优化根据任务复杂度分配合适的智能体易于扩展新增功能只需添加新的专业智能体在实际项目中我们采用的多智能体架构是这样的一个“餐厅礼宾代理”作为总协调员负责理解用户意图并将任务分发给专业的“菜单查询代理”通过MCP协议和“预订代理”通过A2A协议。2. 核心概念解析MCP vs A2A vs 传统技能2.1 MCPModel Context Protocol智能体与工具的桥梁MCP的核心作用是让AI智能体能够安全、标准化地使用外部工具和数据源。可以把MCP理解为智能体的“手和眼睛”——它让智能体能够操作工具、获取信息。# MCP工具示例菜单查询工具 class MenuQueryTool: def search_by_category(self, category: str) - List[MenuItem]: 根据分类查询菜单 pass def search_by_description(self, description: str) - List[MenuItem]: 语义搜索菜单项 pass2.2 A2AAgent-to-Agent智能体间的协作协议A2A协议解决了不同智能体之间如何发现彼此、如何通信、如何协作的问题。与MCP主要连接工具不同A2A连接的是其他智能体。关键区别MCP工具无状态单一功能执行完即结束A2A代理有状态具备推理能力支持多轮对话2.3 智能体卡片Agent Card智能体的“名片”每个通过A2A协议暴露的智能体都需要一个标准化的描述文件这就是智能体卡片。它包含智能体的功能描述、技能列表、输入输出格式等信息。{ agent_name: Reservation Agent, description: 处理餐厅预订创建、查询和取消, skills: [ { id: manage_reservations, name: 餐厅预订管理, examples: [ 预订4人周五晚上7点的座位, 查询电话号码555-0101的预订 ] } ] }3. 环境准备与项目结构3.1 开发环境要求Python版本3.9或更高版本包管理uv快速Python包管理器AI框架Google ADKAgent Development Kit模型支持Gemini系列或DeepSeek本文兼容3.2 项目目录结构adk-a2a-agent-runtime-starter/ ├── reservation_agent/ # 预订代理 │ ├── agent.py # 代理核心逻辑 │ ├── a2a_config.py # A2A配置和智能体卡片 │ └── executor.py # A2A执行器 ├── restaurant_agent/ # 餐厅总代理 │ └── agent.py # 集成了预订代理的餐厅代理 ├── tools.yaml # MCP工具箱配置 ├── scripts/ # 部署和测试脚本 └── .env # 环境配置3.3 初始环境设置# 克隆项目代码 git clone https://github.com/alphinside/adk-a2a-agent-runtime-starter.git cd adk-a2a-agent-runtime-starter # 使用uv安装依赖uv比pip快10倍以上 uv sync # 环境配置 cp .env.example .env # 编辑.env文件配置项目ID和区域4. 构建专业预订代理4.1 预订代理的核心功能设计我们的预订代理需要处理三种核心操作创建预订收集用户信息并创建新的预订记录查询预订根据电话号码查询现有预订取消预订取消已存在的预订# reservation_agent/agent.py import os from google.adk.agents import LlmAgent 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: Reservation cancelled} # 创建主要的LLM代理 root_agent LlmAgent( namereservation_agent, modelgemini-3.5-flash, # 可替换为DeepSeek模型 instruction你是Foodie Finds餐厅的友好预订助手。 帮助食客创建、查询和取消餐桌预订。 创建预订时需要收集以下信息 - 预订姓名 - 电话号码作为预订ID - 人数 - 日期 - 时间 在创建前始终确认详细信息。 查询或取消时如果未提供电话号码请询问。 保持简洁专业。, tools[create_reservation, check_reservation, cancel_reservation], )4.2 状态管理的设计考量在AI Agent开发中状态管理是一个关键问题。我们采用了以下几种策略会话状态在同一对话中保持上下文应用状态跨会话持久化数据如预订记录临时状态单次请求的临时数据对于预订数据我们使用应用级状态确保即使对话结束预订信息仍然保留。这种设计比使用数据库更轻量适合中小型应用。5. 配置A2A协议支持5.1 创建智能体卡片智能体卡片是其他智能体发现和理解我们预订代理功能的途径。# 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, name餐厅预订管理, 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_name预订代理, description处理餐厅餐桌预订——为Foodie Finds餐厅创建、查询和取消预订, skills[reservation_skill], )5.2 实现A2A执行器执行器是A2A协议与ADK代理之间的桥梁负责处理传入的A2A请求并通过我们的代理执行。# reservation_agent/executor.py import os import vertexai from a2a.server.agent_execution import AgentExecutor, RequestContext from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService, VertexAiSessionService class ReservationAgentExecutor(AgentExecutor): 连接A2A协议和ADK预订代理的桥梁 def __init__(self) - None: self.agent None self.runner None def _init_agent(self): if self.agent is not None: return from reservation_agent.agent import root_agent as reservation_agent self.agent reservation_agent # 根据环境选择会话服务 engine_id os.environ.get(GOOGLE_CLOUD_AGENT_ENGINE_ID) if engine_id: # 生产环境使用持久化会话服务 project os.environ.get(GOOGLE_CLOUD_PROJECT) location os.environ.get(GOOGLE_CLOUD_LOCATION, us-central1) vertexai.init(projectproject, locationlocation) session_service VertexAiSessionService( projectproject, locationlocation, 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, session_servicesession_service, # 其他服务配置... )这种设计的关键优势在于环境自适应——在本地开发时使用轻量级的内存存储部署到生产环境时自动切换为持久化存储。6. 本地测试与验证6.1 完整的本地测试流程在部署之前我们需要确保A2A代理在本地正常工作。以下是完整的测试脚本# scripts/test_a2a_agent_local.py import asyncio import os from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor async def main(): # 创建本地A2A代理 a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor ) a2a_agent.set_up() # 测试1获取智能体卡片 print( 测试智能体卡片 ) card_response await a2a_agent.handle_authenticated_agent_card() print(f代理名称: {card_response.get(name)}) print(f技能列表: {[s.get(name) for s in card_response.get(skills, [])]}) # 测试2创建预订 print(\n 测试创建预订 ) message_data { message: { messageId: test-001, content: [{text: 预订2人周六晚上6点的座位姓名Bob电话555-0202}], role: ROLE_USER, }, } response await a2a_agent.on_message_send(**message_data) task_id response[task][id] print(f任务ID: {task_id}) # 等待任务完成并获取结果 result await wait_for_task_completion(a2a_agent, task_id) print(f结果: {result}) if __name__ __main__: asyncio.run(main())6.2 测试结果验证运行测试脚本后你应该看到类似以下的输出 测试智能体卡片 代理名称: 预订代理 技能列表: [餐厅预订管理] 测试创建预订 任务ID: f7f7004d-cfea-49c2-b57d-5bca9959e193 结果: 您的预订已确认Bob2人周六晚上6点电话555-0202这个测试验证了A2A代理能够正确暴露其功能描述代理能够处理预订请求并返回正确结果整个A2A协议流程正常工作7. 部署到Agent运行时环境7.1 部署脚本实现将代理部署到生产环境需要处理依赖管理、环境配置和部署流程。# scripts/deploy_a2a_agent_runtime.py import os from pathlib import Path import vertexai from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor def deploy_reservation_agent(): 部署预订代理到Agent运行时 # 环境配置 project_id os.environ[GOOGLE_CLOUD_PROJECT] region os.environ[REGION] staging_bucket os.environ.get(STAGING_BUCKET) vertexai.init(projectproject_id, locationregion, staging_bucketstaging_bucket) # 创建A2A代理实例 a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor, ) print(开始部署预订代理到Agent运行时...) print(预计需要3-5分钟...) # 执行部署 client vertexai.Client(projectproject_id, locationregion) 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], }, ) print(f部署完成资源名称: {remote_agent.api_resource.name}) return remote_agent.api_resource.name7.2 部署后的验证部署完成后我们需要验证代理在生产环境的正常运行# scripts/test_deployed_agent.py import asyncio import vertexai from dotenv import load_dotenv load_dotenv() async def test_deployed_agent(): 测试已部署的代理 project_id os.environ[GOOGLE_CLOUD_PROJECT] region os.environ[REGION] resource_name os.environ[RESERVATION_AGENT_RESOURCE_NAME] vertexai.init(projectproject_id, locationregion) client vertexai.Client(projectproject_id, locationregion) # 获取已部署的代理 agent client.agent_engines.get(nameresource_name) # 测试智能体卡片 card await agent.handle_authenticated_agent_card() print(f已部署代理: {card.name}) print(f访问URL: {card.url}) # 测试预订功能 response await agent.on_message_send( messageIdprod-test-001, roleuser, parts[{kind: text, text: 预订3人周日下午12点的座位}] ) print(生产环境测试通过) if __name__ __main__: asyncio.run(test_deployed_agent())8. 集成多智能体系统8.1 升级餐厅主代理现在我们需要将预订代理集成到餐厅主代理中实现真正的多智能体协作。# 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 toolbox_adk import ToolboxToolset 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(httpx.Request(POST, https://oauth2.googleapis.com/token)) request.headers[Authorization] fBearer {self.credentials.token} yield request # 配置工具和代理 TOOLBOX_URL os.environ.get(TOOLBOX_URL, http://127.0.0.1:5000) RESERVATION_AGENT_CARD_URL os.environ.get(RESERVATION_AGENT_CARD_URL, ) toolbox ToolboxToolset(TOOLBOX_URL) # 创建远程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, instruction你是Foodie Finds餐厅的知识渊博的礼宾员。 你的工作 - 帮助食客按分类或菜系浏览菜单 - 提供菜品的完整详细信息包括成分、价格和饮食信息 - 根据食客渴望的描述推荐菜品 - 当用户想要预订、查询或取消餐桌时委托给reservation_agent 当食客按名称或菜系询问特定菜品时使用get-item-details工具。 当食客要求特定分类或菜系类型时使用search-menu工具。 当食客描述他们想要的食物类型时使用search-menu-by-description工具进行语义搜索。 对于预订请求预订、查询或取消餐桌委托给reservation_agent。 保持对话性、知识渊博且简洁。, tools[toolbox], sub_agents[reservation_remote_agent], )8.2 多智能体协作流程这个集成实现了以下协作流程用户请求分析主代理首先分析用户意图任务路由如果是菜单相关请求路由到MCP工具箱如果是预订相关请求路由到A2A预订代理结果整合主代理将子代理的结果整合后返回给用户9. 完整系统测试与验证9.1 端到端测试场景我们需要测试整个多智能体系统的各种使用场景# scripts/test_full_system.py import asyncio import requests async def test_full_system(): 测试完整的多智能体系统 agent_url os.environ.get(AGENT_URL, http://localhost:8080) test_scenarios [ { name: 菜单查询测试, message: 你们有哪些意大利菜, expected_keywords: [pasta, pizza, 意大利] }, { name: 预订创建测试, message: 我想预订周五晚上7点4个人的座位姓名Eve电话555-0505, expected_keywords: [确认, 预订, Eve] }, { name: 预订查询测试, message: 查询电话555-0505的预订, expected_keywords: [预订, Eve, 周五] } ] for scenario in test_scenarios: print(f\n {scenario[name]} ) # 发送请求到代理 response requests.post( f{agent_url}/api/chat, json{message: scenario[message]} ) if response.status_code 200: result response.json() print(f响应: {result[response]}) # 验证响应包含预期关键词 for keyword in scenario[expected_keywords]: if keyword in result[response]: print(f✓ 包含预期关键词: {keyword}) else: print(f✗ 缺少预期关键词: {keyword}) else: print(f请求失败: {response.status_code})9.2 性能与稳定性考量在生产环境中我们还需要考虑错误处理网络超时、代理不可用等情况重试机制临时故障的自动恢复限流控制防止单个用户过度使用资源监控告警实时监控系统健康状态10. 生产环境最佳实践10.1 安全配置# 生产环境安全配置示例 security: # API密钥管理 api_keys: rotation_period: 90 days emergency_rotation: enabled # 访问控制 access_control: ip_whitelist: [10.0.0.0/8] rate_limiting: 100 requests/minute # 数据保护 data_protection: encryption: enabled retention_period: 30 days10.2 监控与日志建立完整的可观测性体系# 监控配置示例 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider # 设置分布式追踪 tracer_provider TracerProvider() trace.set_tracer_provider(tracer_provider) # 关键指标监控 monitoring_metrics [ agent_response_time, a2a_request_success_rate, concurrent_users, error_rate_by_agent ]10.3 成本优化策略特别是使用DeepSeek等成本较低的模型时缓存策略对常见查询结果进行缓存请求批处理合并多个小请求模型选择根据任务复杂度选择合适的模型规格自动扩缩根据负载动态调整资源11. 常见问题与解决方案11.1 部署问题排查问题现象可能原因解决方案部署超时网络问题或依赖下载慢检查网络连接使用国内镜像源依赖冲突版本不兼容使用uv管理依赖确保版本一致权限错误服务账号权限不足配置正确的IAM角色11.2 运行时问题问题现象可能原因解决方案A2A调用失败身份验证令牌过期实现自动令牌刷新机制响应缓慢模型推理时间过长优化提示词启用流式响应状态丢失会话配置错误检查会话服务配置11.3 模型兼容性问题由于我们设计时考虑了多模型兼容性从Gemini切换到DeepSeek只需要修改配置# 修改模型配置 # 从Gemini切换到DeepSeek model_config { gemini: gemini-3.5-flash, deepseek: deepseek-chat # 根据实际模型名称调整 } # 在agent初始化时选择模型 selected_model model_config[os.environ.get(MODEL_PROVIDER, deepseek)]12. 扩展与进阶方向12.1 添加新的专业智能体基于这个架构你可以轻松添加新的智能体支付处理代理处理订单支付和退款推荐代理基于用户历史提供个性化推荐客服代理处理投诉和复杂咨询12.2 性能优化建议异步处理对耗时操作使用异步模式连接池重用数据库和API连接预加载提前加载常用数据到内存压缩传输对大量数据使用压缩算法12.3 企业级特性对于大型企业应用可以考虑多租户支持隔离不同客户的数据和配置审计日志记录所有操作用于合规审查备份恢复定期备份智能体状态和配置蓝绿部署实现零停机更新通过本文的实战指南你应该已经掌握了构建生产级多智能体系统的核心技能。从单智能体到多智能体协作的转变不仅仅是技术架构的升级更是开发思维的转变。这种架构为构建复杂、可扩展的AI应用提供了坚实的基础。记住最好的学习方式是实践。建议你按照本文的步骤亲手实现这个系统然后在基础上进行扩展和优化。随着经验的积累你将能够设计出更加复杂和强大的多智能体解决方案。
