手撕AI Agent框架多智能体MCPA2A全链路实战DeepSeek可跑最近在AI Agent开发中踩了不少坑特别是多智能体协作的场景下不同框架的兼容性和部署复杂度让人头疼。本文将从零开始构建一个完整的餐厅多智能体系统涵盖MCP工具集成、A2A协议通信、智能体部署等核心环节所有代码都兼容DeepSeek模型可直接运行。1. AI Agent技术栈选型与架构设计1.1 为什么需要多智能体架构在真实业务场景中单个AI智能体往往难以胜任复杂任务。比如餐厅服务场景需要菜单查询、预订管理、客户服务等多个专业能力。多智能体架构的优势在于职责分离每个智能体专注特定领域便于维护和迭代独立部署不同智能体可以有独立的更新周期和部署策略弹性扩展可根据业务负载单独扩缩容特定功能模块技术异构不同智能体可以采用最适合的技术栈1.2 核心技术组件介绍MCPModel Context ProtocolMCP是连接AI智能体与工具数据的标准协议让智能体能够安全地访问外部系统和数据源。在我们的餐厅场景中MCP用于菜单数据的查询和管理。A2AAgent-to-Agent协议A2A解决了智能体间的通信问题标准化了智能体发现、能力描述和任务协作的机制。与MCP主要区别在于MCP智能体↔工具无状态函数调用A2A智能体↔智能体有状态多轮对话ADKAgent Development KitGoogle开源的AI智能体开发框架提供统一的开发模式和部署工具。1.3 系统架构设计我们的多智能体餐厅系统包含以下组件用户请求 ↓ 餐厅主智能体编排器 ├── MCP工具箱菜单查询 └── A2A预订智能体预订管理 ↓ Agent Runtime托管服务主智能体负责请求路由菜单查询走MCP协议预订操作走A2A协议。2. 开发环境准备与依赖配置2.1 基础环境要求# 检查Python版本 python --version # 需要Python 3.9 uv --version # 包管理工具 # 创建项目目录 mkdir ai-agent-restaurant cd ai-agent-restaurant2.2 项目依赖配置创建pyproject.toml文件[project] name ai-agent-restaurant version 0.1.0 description Multi-agent restaurant system with MCP and A2A requires-python 3.9 dependencies [ google-cloud-aiplatform[agent_engines,adk]1.149.0, a2a-sdk0.3.26, google-adk1.29.0, httpx0.25.0, pydantic2.0.0, cloudpickle2.0.0, ] [build-system] requires [hatchling] build-backend hatchling.build创建环境配置文件.env.example# 项目配置 GOOGLE_CLOUD_PROJECTyour-project-id GOOGLE_CLOUD_LOCATIONus-central1 REGIONus-central1 # 服务配置 TOOLBOX_URLhttp://127.0.0.1:5000 RESERVATION_AGENT_CARD_URL # 部署配置 STAGING_BUCKETyour-project-adk-a2a-agent-runtime2.3 初始化项目结构# 创建项目目录结构 mkdir -p {reservation_agent,restaurant_agent,scripts,logs,tools} # 初始化Python环境 uv sync项目最终结构如下ai-agent-restaurant/ ├── reservation_agent/ # 预订智能体 │ ├── agent.py │ ├── a2a_config.py │ ├── executor.py │ └── __init__.py ├── restaurant_agent/ # 餐厅主智能体 │ ├── agent.py │ └── __init__.py ├── scripts/ # 部署和测试脚本 ├── tools/ # MCP工具配置 ├── logs/ # 日志文件 └── pyproject.toml3. 构建预订智能体A2A服务端3.1 预订智能体核心逻辑创建reservation_agent/agent.pyimport os from functools import cached_property from typing import Any from google.adk.agents import LlmAgent from google.adk.models.google_llm import Gemini from google.adk.tools import ToolContext from google.genai import Client, types # 应用级状态前缀确保预订数据在所有会话间持久化 STATE_PREFIX app:reservation: class DeepSeekCompatibleGemini(Gemini): 兼容DeepSeek的Gemini封装类 cached_property def api_client(self) - Client: project os.getenv(GOOGLE_CLOUD_PROJECT) return Client( projectproject, locationglobal, http_optionstypes.HttpOptions( headersself._tracking_headers(), retry_optionsself.retry_options, ), ) 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}. Phone: {phone_number}., } 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}., } if reservation.get(status) cancelled: return { success: False, message: fReservation for {phone_number} is already cancelled., } reservation[status] cancelled tool_context.state[key] reservation return { success: True, message: fReservation for {reservation[name]} ({phone_number}) has been cancelled., } # 创建主智能体实例 root_agent LlmAgent( namereservation_agent, modelDeepSeekCompatibleGemini(modelgemini-3.5-flash), instructionYou are a friendly reservation assistant for Foodie Finds restaurant. You help diners create, check, and cancel table reservations. When a diner wants to make a reservation, collect these details: - Name for the reservation - Phone number (used as the reservation ID) - Party size (number of guests) - Date - Time Always confirm the details before creating the reservation. When checking or cancelling, ask for the phone number if not provided. Be concise and professional., tools[create_reservation, check_reservation, cancel_reservation], )3.2 A2A智能体卡片配置创建reservation_agent/a2a_config.pyfrom a2a.types import AgentSkill from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card # 定义预订技能 reservation_skill AgentSkill( idmanage_reservations, nameRestaurant Reservations, descriptionCreate, check, and cancel table reservations at Foodie Finds restaurant, tags[reservations, restaurant, booking], examples[ Book a table for 4 on Friday at 7pm, Check reservation for 555-0101, Cancel my reservation, phone number 555-0101, ], input_modes[text/plain], output_modes[text/plain], ) # 创建智能体卡片 agent_card create_agent_card( agent_nameReservation Agent, descriptionHandles restaurant table reservations — create, check, and cancel bookings for Foodie Finds restaurant., skills[reservation_skill], )3.3 A2A执行器实现创建reservation_agent/executor.pyimport os from typing import NoReturn import vertexai 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, UnsupportedOperationError from a2a.utils import new_agent_text_message from a2a.utils.errors import ServerError 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: # 生产环境使用持久化会话 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, artifact_serviceInMemoryArtifactService(), session_servicesession_service, memory_serviceInMemoryMemoryService(), ) async def execute(self, context: RequestContext, event_queue: EventQueue) - None: 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)]) 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, messagenew_agent_text_message(fError: {e!s}), ) raise async def _get_or_create_session(self, context_id: str, user_id: str): app_name self.runner.app_name if context_id: session await self.runner.session_service.get_session( app_nameapp_name, session_idcontext_id, user_iduser_id, ) if session: return session session await self.runner.session_service.create_session( app_nameapp_name, user_iduser_id, session_idcontext_id, ) return session async def cancel(self, context: RequestContext, event_queue: EventQueue) - NoReturn: raise ServerError(errorUnsupportedOperationError())4. 本地测试A2A智能体4.1 创建本地测试脚本创建scripts/test_a2a_agent_local.pyimport asyncio import json import os from pprint import pprint from dotenv import load_dotenv from starlette.requests import Request from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor load_dotenv() def build_post_request(data: dict None, path_params: dict None) - Request: 构建模拟POST请求 scope { type: http, http_version: 1.1, headers: [(bcontent-type, bapplication/json)], app: None } if path_params: scope[path_params] path_params async def receive(): byte_data json.dumps(data).encode(utf-8) if data else b return {type: http.request, body: byte_data, more_body: False} return Request(scope, receive) async def wait_for_task(a2a_agent, task_id, max_retries30): 轮询任务直到完成 for _ in range(max_retries): request build_post_request() result await a2a_agent.on_get_task(requestrequest, contextNone) state result.get(status, {}).get(state, ) if state in [completed, failed]: return result await asyncio.sleep(1) return result async def main(): # 创建本地A2A智能体 a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor ) a2a_agent.set_up() print( * 50) print(1. 测试智能体卡片检索...) print( * 50) # 测试智能体卡片 request build_post_request() card_response await a2a_agent.handle_authenticated_agent_card(requestrequest, contextNone) print(f智能体名称: {card_response.get(name)}) print(f可用技能: {[s.get(name) for s in card_response.get(skills, [])]}) # 测试预订创建 print(\n * 50) print(2. 测试预订创建...) print( * 50) message_data { message: { messageId: fmsg-test-001, content: [{text: Book a table for 2 on Saturday at 6pm. Name: Bob, Phone: 555-0202}], role: ROLE_USER, }, } request build_post_request(message_data) response await a2a_agent.on_message_send(requestrequest, contextNone) task_id response[task][id] context_id response[task].get(contextId) print(f任务ID: {task_id}) # 等待任务完成 result await wait_for_task(a2a_agent, task_id) print(f任务状态: {result.get(status, {}).get(state)}) if result.get(artifacts): for artifact in result.get(artifacts, []): if artifact.get(parts) and text in artifact[parts][0]: print(f智能体回复: {artifact[parts][0][text]}) print(\n * 50) print(本地A2A测试完成!) print( * 50) if __name__ __main__: asyncio.run(main())4.2 运行本地测试# 设置环境变量 source .env # 运行测试 PYTHONPATH. uv run python scripts/test_a2a_agent_local.py预期输出 1. 测试智能体卡片检索... 智能体名称: Reservation Agent 可用技能: [Restaurant Reservations] 2. 测试预订创建... 任务ID: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx 任务状态: completed 智能体回复: Your reservation for Bob, party of 2, on Saturday at 6:00 PM has been confirmed.5. 部署A2A智能体到Agent Runtime5.1 创建部署脚本创建scripts/deploy_a2a_agent_runtime.pyimport os from pathlib import Path import vertexai from dotenv import load_dotenv from google.genai import types from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor load_dotenv() def main(): PROJECT_ID os.environ[GOOGLE_CLOUD_PROJECT] REGION os.environ[REGION] STAGING_BUCKET os.environ.get(STAGING_BUCKET, f{PROJECT_ID}-adk-a2a-agent-runtime) BUCKET_URI fgs://{STAGING_BUCKET} # 初始化A2A智能体 a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor, ) # 初始化Vertex AI vertexai.init(projectPROJECT_ID, locationREGION, staging_bucketBUCKET_URI) client vertexai.Client( projectPROJECT_ID, locationREGION, http_optionstypes.HttpOptions(api_versionv1beta1), ) print(正在部署预订智能体到Agent Runtime...) print(这可能需要3-5分钟...) # 部署到Agent Runtime 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, cloudpickle2.0.0, pydantic2.0.0 ], extra_packages: [./reservation_agent], http_options: {api_version: v1beta1}, staging_bucket: BUCKET_URI, }, ) resource_name remote_agent.api_resource.name print(f\n部署完成!) print(f资源名称: {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(已将RESERVATION_AGENT_RESOURCE_NAME写入.env文件) if __name__ __main__: main()5.2 执行部署# 创建存储桶用于暂存 STAGING_BUCKET${GOOGLE_CLOUD_PROJECT}-adk-a2a-agent-runtime gsutil mb -l $REGION -p $GOOGLE_CLOUD_PROJECT gs://$STAGING_BUCKET 2/dev/null || echo 存储桶已存在 echo STAGING_BUCKET$STAGING_BUCKET .env # 运行部署 source .env PYTHONPATH. uv run python scripts/deploy_a2a_agent_runtime.py5.3 验证部署创建测试脚本scripts/test_deployed_agent.pyimport asyncio import os import time import vertexai from a2a.types import TaskState from dotenv import load_dotenv from google.genai import types load_dotenv() async def main(): 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, http_optionstypes.HttpOptions(api_versionv1beta1), ) # 获取已部署的智能体 agent client.agent_engines.get(nameRESOURCE_NAME) print( * 50) print(测试已部署的A2A智能体...) print( * 50) # 发送测试消息 message_data { messageId: msg-deploy-test-001, role: user, parts: [{kind: text, text: Book a table for 3 on Sunday at noon. Name: Carol, Phone: 555-0303}], } response await agent.on_message_send(**message_data) # 处理响应 task_object None for chunk in response: if isinstance(chunk, tuple) and len(chunk) 0 and hasattr(chunk[0], id): task_object chunk[0] break if task_object: print(f任务ID: {task_object.id}) print(f初始状态: {task_object.status.state}) # 等待任务完成 for _ in range(30): try: result await agent.on_get_task(idtask_object.id) if result.status.state in [TaskState.completed, TaskState.failed]: break except Exception: pass time.sleep(1) print(f最终状态: {result.status.state}) if result.artifacts: for artifact in result.artifacts: if artifact.parts and hasattr(artifact.parts[0], text): print(f智能体回复: {artifact.parts[0].text}) if __name__ __main__: asyncio.run(main())6. 构建餐厅主智能体MCP A2A集成6.1 主智能体实现创建restaurant_agent/agent.pyimport 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 # MCP工具箱集成模拟 class MockToolboxToolset: 模拟MCP工具箱用于菜单查询 def __init__(self, toolbox_url: str): self.toolbox_url toolbox_url async def search_menu(self, query: str) - dict: 模拟菜单搜索 # 在实际项目中这里会调用真实的MCP服务 menu_items [ {name: Margherita Pizza, price: 12.99, category: Italian}, {name: Caesar Salad, price: 8.99, category: Salads}, {name: Spaghetti Carbonara, price: 14.99, category: Italian}, ] return { results: [item for item in menu_items if query.lower() in item[name].lower() or query.lower() in item[category].lower()], count: len(menu_items) } # Google Cloud认证处理 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 # 配置环境变量 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 MockToolboxToolset(TOOLBOX_URL) # 创建远程A2A智能体预订服务 reservation_remote_agent RemoteA2aAgent( namereservation_agent, descriptionHandles restaurant table reservations — create, check, and cancel bookings., agent_cardRESERVATION_AGENT_CARD_URL, httpx_clienthttpx.AsyncClient(authGoogleCloudAuth(), timeout60), ) # 创建主智能体 root_agent LlmAgent( namerestaurant_agent, modelgemini-3.5-flash, instructionYou are a friendly and knowledgeable concierge at Foodie Finds restaurant. 核心职责 - 通过MCP工具箱帮助顾客浏览菜单按类别或菜系 - 提供菜品详细信息成分、价格、饮食信息 - 根据顾客描述推荐菜品 - 将预订相关请求委托给专门的预订智能体 请求路由逻辑 1. 菜单查询 → MCP工具箱 2. 预订操作 → A2A预订智能体 对话风格专业、友好、简洁确保顾客获得准确信息。, tools[toolbox.search_menu], # 实际项目中会有更多MCP工具 sub_agents[reservation_remote_agent], )6.2 智能体卡片解析脚本创建scripts/resolve_agent_card_url.pyimport asyncio import os from pathlib import Path import vertexai from dotenv import load_dotenv from google.genai import types load_dotenv() async def main(): 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, http_optionstypes.HttpOptions(api_versionv1beta1), ) # 获取已部署的智能体 agent client.agent_engines.get(nameRESOURCE_NAME) # 获取智能体卡片 card await agent.handle_authenticated_agent_card() card_url f{card.url}/v1/card print(f智能体: {card.name}) print(f卡片URL: {card_url}) # 更新环境配置文件 for env_path in [Path(restaurant_agent/.env), 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_CARD_URL)] lines.append(fRESERVATION_AGENT_CARD_URL{card_url}) env_path.write_text(\n.join(lines) \n) print(f已更新 {env_path}) if __name__ __main__: asyncio.run(main())7. 完整系统集成测试7.1 集成测试脚本创建scripts/test_integrated_system.pyimport asyncio import os from dotenv import load_dotenv from restaurant_agent.agent import root_agent as restaurant_agent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.adk.artifacts import InMemoryArtifactService from google.adk.memory.in_memory_memory_service import InMemoryMemoryService from google.genai import types load_dotenv() async def test_integrated_system(): 测试完整的多智能体系统 # 初始化餐厅智能体运行器 runner Runner( app_namerestaurant_test, agentrestaurant_agent, artifact_serviceInMemoryArtifactService(), session_serviceInMemorySessionService(), memory_serviceInMemoryMemoryService(), ) test_cases [ { name: 菜单查询测试, query: What Italian dishes do you have?, expected_keywords: [pizza, pasta, Italian] }, { name: 预订创建测试, query: I want to book a table for 4 people this Friday at 7 PM, expected_keywords: [reservation, book, table] }, { name: 预订查询测试, query: Can you check my reservation for phone number 555-0202?, expected_keywords: [reservation, check, found] } ] for i, test_case in enumerate(test_cases): print(f\n{*50}) print(f测试 {i1}: {test_case[name]}) print(f{*50}) print(f用户查询: {test_case[query]}) # 运行智能体 session await runner.session_service.create_session( app_namerestaurant_test, user_idftest_user_{i}, session_idftest_session_{i} ) content types.Content(roleuser, parts[types.Part(texttest_case[query])]) async for event in runner.run_async( session_idsession.id, user_idftest_user_{i}, new_messagecontent, ): if event.is_final_response(): response_text .join(p.text for p in event.content.parts if p.text) print(f智能体回复: {response_text}) # 验证回复包含预期关键词 keywords_found [ keyword for keyword in test_case[expected_keywords] if keyword.lower() in response_text.lower() ] print(f找到关键词: {keywords_found}) break async def main(): print(开始集成系统测试...) await test_integrated_system() print(\n测试完成!) if __name__ __main__: asyncio.run(main())7.2 运行集成测试# 解析智能体卡片URL uv run python scripts/resolve_agent_card_url.py # 运行集成测试 source .env PYTHONPATH. uv run python scripts/test_integrated_system.py8. 生产环境部署与优化8.1 Cloud Run部署配置创建scripts/deploy_to_cloud_run.pyimport os import subprocess from dotenv import load_dotenv load_dotenv() def deploy_restaurant_agent(): 部署餐厅主智能体到Cloud Run PROJECT_ID os.environ[GOOGLE_CLOUD_PROJECT] REGION os.environ[REGION] RESERVATION_AGENT_CARD_URL os.environ[RESERVATION_AGENT_CARD_URL] # 授予Cloud Run服务账号访问Agent Runtime的权限 project_number_cmd fgcloud projects describe {PROJECT_ID} --formatvalue(projectNumber) project_number subprocess.check_output(project_number_cmd, shellTrue).decode().strip() # 设置IAM权限 iam_cmd f gcloud projects add-iam-policy-binding {PROJECT_ID} \ --memberserviceAccount:{project_number}-computedeveloper.gserviceaccount.com \ --roleroles/aiplatform.user subprocess.run(iam_cmd, shellTrue, checkTrue) # 部署到Cloud Run deploy_cmd f gcloud run deploy restaurant-agent \ --source . \ --region{REGION} \ --allow-unauthenticated \ --update-env-varsRESERVATION_AGENT_CARD_URL{RESERVATION_AGENT_CARD_URL} \ --min-instances0 \ --max-instances1 \ --memory1Gi \ --port8080 print(正在部署餐厅智能体到Cloud Run...) subprocess.run(deploy_cmd, shellTrue, checkTrue) # 获取服务URL url_cmd fgcloud run services describe restaurant-agent --region{REGION} --formatvalue(status.url) service_url subprocess.check_output(url_cmd, shellTrue).decode().strip() print(f部署完成! 服务URL: {service_url}) return service_url if __name__ __main__: deploy_restaurant_agent()8.2 性能优化配置创建optimization_config.yaml# 智能体性能优化配置 optimization: # 会话管理 session: timeout: 3600 # 会话超时时间秒 max_size: 1000 # 最大会话数 # 缓存配置 cache: enabled: true ttl: 300 # 缓存存活时间 # 并发控制 concurrency: max_workers: 10 queue_size: 100 # 监控配置 monitoring: enabled: true metrics: - response_time - error_rate - throughput9. 常见问题与解决方案9.