企业官网建设流程全解析

1. 先搞清楚这个项目到底解决什么实际问题

如果你正在研究AI Agent开发，特别是多智能体协作场景，这个"手撕AI Agent框架"项目值得重点关注。它不是一个简单的Demo演示，而是完整展示了从单智能体到多智能体系统的全链路实现。

核心解决的是生产级AI Agent系统的架构问题：如何让不同的AI Agent既能各司其职，又能有效协作。传统的单智能体框架在处理复杂业务时往往力不从心，而这个项目通过A2A协议和MCP工具箱的集成，展示了真正的分布式智能体协作方案。

最实际的价值在于：用DeepSeek就能跑通。很多多智能体项目对硬件要求极高，但这个方案在普通开发环境下就能验证，特别适合想要深入理解Agent架构的开发者。

2. 环境准备：DeepSeek + 基础开发环境就能开始

2.1 硬件和软件要求

这个项目的优势在于对硬件要求相对友好：

最低配置：

• CPU: 4核以上（Intel i5或同等性能）
• 内存: 16GB（8GB勉强可跑但体验较差）
• 存储: 至少20GB可用空间
• 网络: 稳定互联网连接（用于模型调用）

推荐配置：

• CPU: 8核以上
• 内存: 32GB
• GPU: 可选，非必须（项目主要使用云端LLM）

软件环境：

• 操作系统: Ubuntu 20.04+ / macOS 12+ / Windows 10+（WSL2推荐）
• Python: 3.9-3.11
• 包管理: uv（Rust编写的快速Python包管理器）
• 版本控制: Git

2.2 DeepSeek API配置

项目使用DeepSeek作为主要LLM，需要提前准备：

# 获取DeepSeek API Key # 访问DeepSeek官网注册账号并获取API密钥 # 设置环境变量 export DEEPSEEK_API_KEY="your_actual_api_key_here" export DEEPSEEK_BASE_URL="https://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": f"Reservation 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": f"No 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": f"No reservation found for {phone_number}."} reservation["status"] = "cancelled" tool_context.state[key] = reservation return {"success": True, "message": f"Reservation for {reservation['name']} has been cancelled."} # 创建主智能体 root_agent = LlmAgent( name="reservation_agent", model="gemini-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( id="manage_reservations", name="Restaurant 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_name="Reservation 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( project=os.environ.get("GOOGLE_CLOUD_PROJECT"), location=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), agent_engine_id=engine_id, ) app_name = engine_id else: # 开发环境：使用内存会话 session_service = InMemorySessionService() app_name = self.agent.name self.runner = Runner( app_name=app_name, agent=self.agent, artifact_service=InMemoryArtifactService(), session_service=session_service, memory_service=InMemoryMemoryService(), ) 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(role="user", parts=[types.Part(text=query)]) # 通过ADK运行器处理请求 async for event in self.runner.run_async( session_id=session.id, user_id=user_id, new_message=content, ): 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(text=answer)], name="answer") await updater.complete() break except Exception as e: await updater.update_status( TaskState.failed, message=f"Error: {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_card=agent_card, agent_executor_builder=ReservationAgentExecutor ) 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-sdk==0.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"] = f"Bearer {self.credentials.token}" yield request # 创建远程A2A智能体代理 reservation_remote_agent = RemoteA2aAgent( name="reservation_agent", description="处理餐厅餐桌预订——创建、查询和取消预订。当用户想要预订、查询或取消餐桌时委托给此智能体。", agent_card=RESERVATION_AGENT_CARD_URL, httpx_client=httpx.AsyncClient(auth=GoogleCloudAuth(), timeout=60), ) # 主餐厅智能体 root_agent = LlmAgent( name="restaurant_agent", model="gemini-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 --config=tools.yaml > logs/toolbox.log 2>&1 & # 启动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(project=project_id, location=region) a2a_agent = A2aAgent( agent_card=agent_card, agent_executor_builder=ReservationAgentExecutor, ) client = vertexai.Client(project=project_id, location=region) print("开始部署预订智能体...") remote_agent = client.agent_engines.create( agent=a2a_agent, config={ "display_name": agent_card.name, "description": agent_card.description, "requirements": [ "google-cloud-aiplatform[agent_engines,adk]==1.149.0", "a2a-sdk==0.3.26", "google-adk==1.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(f"RESERVATION_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-vars="RESERVATION_AGENT_CARD_URL=$RESERVATION_AGENT_CARD_URL" \ --min-instances=0 \ --max-instances=1 \ --memory=1Gi

6.3 权限和网络配置

服务账号权限：

# 授予Cloud Run服务账号调用Agent Runtime的权限 PROJECT_NUMBER=$(gcloud projects describe $GOOGLE_CLOUD_PROJECT --format='value(projectNumber)') gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \ --member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \ --role="roles/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 wrapper

7.2 错误处理和重试机制

智能A2A调用重试：

class ResilientA2aClient: """具有重试机制的A2A客户端""" def __init__(self, max_retries=3, backoff_factor=1.0): self.max_retries = max_retries self.backoff_factor = backoff_factor async def call_with_retry(self, agent, message, context_id=None): """带重试的A2A调用""" for attempt in range(self.max_retries): try: response = await agent.on_message_send( messageId=f"msg-{attempt}", role="user", parts=[{"kind": "text", "text": message}], contextId=context_id ) return response except Exception as e: if attempt == self.max_retries - 1: raise # 指数退避 await asyncio.sleep(self.backoff_factor * (2 ** attempt))

7.3 扩展性设计

多智能体架构的扩展模式：

1. 垂直扩展：为现有智能体添加更多工具和能力
2. 水平扩展：增加新的专业智能体（如支付智能体、配送智能体）
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: 120

8. 实际踩坑经验和排查指南

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(level=logging.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的优化：

1. 批量处理：将多个相关请求合并为单个调用
2. 缓存策略：对频繁查询的结果进行缓存
3. 超时设置：根据操作复杂度设置合理的超时时间
4. 并发控制：限制同时进行的A2A调用数量

这个多智能体框架的真正价值在于它提供了一种可扩展的AI Agent架构模式。一旦理解了A2A协议和MCP工具箱的集成原理，你可以将其应用到各种复杂业务场景中，而不仅仅是餐厅预订这个示例。