手撕AI Agent框架！多智能体+MCP+A2A全链路实战（DeepSeek可跑）

如果你还在为AI Agent开发感到困惑，觉得多智能体协作、MCP协议、A2A通信这些概念听起来高大上但实际操作无从下手，那么这篇文章就是为你准备的。今天我们不谈空洞的理论，直接手把手带你从零构建一个完整的餐厅多智能体系统，让你真正理解AI Agent开发的完整链路。

很多人以为AI Agent开发就是调用API那么简单，但实际上真正的挑战在于如何让多个智能体协同工作、如何管理状态、如何实现跨框架通信。本文将以餐厅预订场景为例，带你实战MCP工具集成、A2A多智能体通信、状态管理等核心概念，所有代码都兼容DeepSeek模型，确保你可以实际运行和修改。

1. 这篇文章真正要解决的问题

为什么你需要关注多智能体开发？

传统的单智能体系统在处理复杂任务时存在明显瓶颈：一个智能体要同时处理菜单查询、预订管理、用户对话等多个功能，导致系统臃肿、维护困难、更新周期长。而多智能体架构通过职责分离，让专业的人做专业的事：

• 菜单查询智能体 ：专注于菜品搜索和推荐
• 预订管理智能体 ：专门处理餐厅预订逻辑
• 对话协调智能体 ：负责任务分发和用户交互

这种架构的优势在于：

• 单个智能体功能单一，易于开发和测试
• 不同智能体可以由不同团队独立开发
• 部署和扩缩容更加灵活
• 系统容错性更强

本文要解决的核心痛点：

1. 如何让多个AI智能体发现彼此并协同工作？
2. 如何实现智能体间的标准化通信？
3. 如何管理智能体的状态和会话？
4. 如何将整个系统部署到生产环境？

2. 基础概念与核心原理

2.1 MCP（Model Context Protocol）是什么？

MCP是连接AI智能体与工具数据的协议标准。简单来说，它让智能体能够安全、标准化地访问外部工具和数据源。在我们的餐厅场景中，MCP Toolbox就是通过MCP协议为智能体提供菜单查询能力的。

MCP vs 传统API调用：

• MCP提供了统一的工具发现和调用机制
• 支持工具描述和参数验证
• 具备更好的安全控制和权限管理

2.2 A2A（Agent-to-Agent）协议的核心价值

A2A协议解决了智能体间的通信问题，就像人类工作中的团队协作一样：

# A2A通信的基本流程
用户请求 → 协调智能体 → A2A协议 → 专业智能体 → 返回结果

A2A三大核心组件：

1. 智能体卡片（Agent Card） ：相当于智能体的"名片"，描述其功能和调用方式
2. 消息（Message） ：智能体间的通信内容
3. 任务（Task） ：具有生命周期的执行单元

2.3 多智能体架构的优势对比

架构类型	开发复杂度	维护成本	扩展性	适用场景
单智能体	低	高	差	简单任务
多智能体	中	低	优秀	复杂业务流程

3. 环境准备与前置条件

3.1 基础环境要求

确保你的开发环境满足以下要求：

# 检查Python版本
python --version  # 需要Python 3.9+
uv --version     # 需要uv包管理器

# 检查DeepSeek API访问
# 确保你有可用的DeepSeek API密钥

3.2 项目结构准备

创建项目目录结构：

# 创建项目目录
mkdir multi-agent-restaurant && cd multi-agent-restaurant

# 创建核心目录结构
mkdir -p reservation_agent restaurant_agent scripts logs
touch .env.example .env README.md

# 初始化uv环境
uv venv
source .venv/bin/activate  # Linux/Mac
# 或 .venv\Scripts\activate  # Windows

3.3 依赖包安装

创建 pyproject.toml 文件：

[project]
name = "multi-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-sdk>=0.3.26",
    "google-adk>=1.29.0",
    "httpx>=0.27.0",
    "python-dotenv>=1.0.0",
    "pydantic>=2.0.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

安装依赖：

uv sync

4. 构建预订管理智能体

4.1 创建基础智能体框架

首先构建专门处理餐厅预订的智能体：

# reservation_agent/agent.py
import os
from typing import Dict, Any
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[str, Any]:
    """创建新的餐厅预订"""
    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"预订成功：{name}，{party_size}人，{date} {time}，电话：{phone_number}",
    }

def check_reservation(phone_number: str, tool_context: ToolContext) -> Dict[str, Any]:
    """通过电话号码查询预订"""
    reservation = tool_context.state.get(f"{STATE_PREFIX}{phone_number}")
    if reservation:
        return {"found": True, "reservation": reservation}
    return {"found": False, "message": f"未找到电话{phone_number}的预订"}

def cancel_reservation(phone_number: str, tool_context: ToolContext) -> Dict[str, Any]:
    """取消现有预订"""
    key = f"{STATE_PREFIX}{phone_number}"
    reservation = tool_context.state.get(key)
    
    if not reservation:
        return {"success": False, "message": f"未找到电话{phone_number}的预订"}
    
    if reservation.get("status") == "cancelled":
        return {"success": False, "message": f"电话{phone_number}的预订已取消"}
    
    reservation["status"] = "cancelled"
    tool_context.state[key] = reservation
    
    return {
        "success": True, 
        "message": f"{reservation['name']}（{phone_number}）的预订已取消"
    }

# 创建预订智能体
reservation_agent = LlmAgent(
    name="reservation_agent",
    model="gemini-3.5-flash",  # 可替换为DeepSeek模型
    instruction="""你是Foodie Finds餐厅的友好预订助手。
帮助顾客创建、查询和取消餐桌预订。
当顾客要预订时，请收集以下信息：
- 预订姓名
- 电话号码（作为预订ID）
- 用餐人数
- 日期
- 时间

创建预订前请确认详细信息。
查询或取消时，如果未提供电话号码请询问。
保持简洁专业。""",
    tools=[create_reservation, check_reservation, cancel_reservation],
)

4.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="餐厅预订管理",
    description="为Foodie Finds餐厅创建、查询和取消餐桌预订",
    tags=["reservations", "restaurant", "booking"],
    examples=[
        "预订周五晚上7点4人桌",
        "查询电话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],
)

4.3 实现A2A执行器

执行器是A2A协议与ADK智能体之间的桥梁：

# reservation_agent/executor.py
import 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 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实现持久会话
            project = os.environ.get("GOOGLE_CLOUD_PROJECT")
            location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
            vertexai.init(project=project, location=location)
            session_service = VertexAiSessionService(
                project=project, location=location, 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:
        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)])
            
            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 "无响应"
                    await updater.add_artifact([TextPart(text=answer)], name="answer")
                    await updater.complete()
                    break
                    
        except Exception as e:
            await updater.update_status(
                TaskState.failed, message=new_agent_text_message(f"错误：{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_name=app_name, session_id=context_id, user_id=user_id,
            )
            if session:
                return session
                
        session = await self.runner.session_service.create_session(
            app_name=app_name, user_id=user_id, session_id=context_id,
        )
        return session
    
    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> NoReturn:
        raise ServerError(error=UnsupportedOperationError())

5. 本地测试A2A智能体

在部署之前，我们需要在本地测试A2A智能体的功能：

# scripts/test_a2a_agent_local.py
import asyncio
import json
import os
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": [(b"content-type", b"application/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_retries=30):
    """等待任务完成"""
    for _ in range(max_retries):
        request = build_post_request()
        result = await a2a_agent.on_get_task(request=request, context=None)
        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_card=agent_card, 
        agent_executor_builder=ReservationAgentExecutor
    )
    a2a_agent.set_up()
    
    print("=" * 50)
    print("1. 测试智能体卡片检索...")
    print("=" * 50)
    
    # 测试智能体卡片
    request = build_post_request()
    card_response = await a2a_agent.handle_authenticated_agent_card(
        request=request, context=None
    )
    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": f"msg-{os.urandom(4).hex()}",
            "content": [{"text": "预订周六晚上6点2人桌，姓名：Bob，电话：555-0202"}],
            "role": "ROLE_USER",
        },
    }
    request = build_post_request(message_data)
    response = await a2a_agent.on_message_send(request=request, context=None)
    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')}")
    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("本地测试完成！")
    print("=" * 50)

if __name__ == "__main__":
    asyncio.run(main())

运行测试：

uv run python scripts/test_a2a_agent_local.py

6. 构建餐厅协调智能体

现在创建主协调智能体，负责将用户请求路由到相应的专业智能体：

# 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

# 配置环境变量
TOOLBOX_URL = os.environ.get("TOOLBOX_URL", "http://127.0.0.1:5000")
RESERVATION_AGENT_CARD_URL = os.environ.get("RESERVATION_AGENT_CARD_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),
)

# 创建主餐厅智能体
restaurant_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=[],  # 这里可以添加MCP工具箱工具
    sub_agents=[reservation_remote_agent],
)

7. 系统集成与部署

7.1 配置MCP工具箱集成

创建MCP工具箱配置文件：

# tools.yaml
toolsets:
  - name: "restaurant-tools"
    description: "Tools for restaurant menu management"
    tools:
      - name: "search-menu"
        description: "Search menu items by category or cuisine"
        inputSchema:
          type: "object"
          properties:
            category:
              type: "string"
              description: "Category or cuisine to search for"
          required: ["category"]
      
      - name: "get-item-details"  
        description: "Get detailed information about a specific menu item"
        inputSchema:
          type: "object"
          properties:
            item_id:
              type: "string"
              description: "ID of the menu item"
          required: ["item_id"]
      
      - name: "search-menu-by-description"
        description: "Semantic search for menu items based on natural language description"
        inputSchema:
          type: "object"
          properties:
            description:
              type: "string"
              description: "Natural language description of desired food"
          required: ["description"]

7.2 部署脚本

创建完整的部署脚本：

# scripts/deploy_system.py
import os
import asyncio
from pathlib import Path
import vertexai
from dotenv import load_dotenv
from google.genai import types

load_dotenv()

async def deploy_reservation_agent():
    """部署预订智能体到Agent Runtime"""
    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")
    
    vertexai.init(project=PROJECT_ID, location=REGION, staging_bucket=f"gs://{STAGING_BUCKET}")
    
    client = vertexai.Client(
        project=PROJECT_ID,
        location=REGION,
        http_options=types.HttpOptions(api_version="v1beta1"),
    )
    
    print("部署预订智能体到Agent Runtime...")
    print("这可能需要3-5分钟...")
    
    # 这里需要导入实际的A2A智能体配置
    from reservation_agent.a2a_config import agent_card
    from reservation_agent.executor import ReservationAgentExecutor
    from vertexai.preview.reasoning_engines import A2aAgent
    
    a2a_agent = A2aAgent(
        agent_card=agent_card,
        agent_executor_builder=ReservationAgentExecutor,
    )
    
    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
    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(f"RESERVATION_AGENT_RESOURCE_NAME={resource_name}")
    env_path.write_text("\n".join(lines) + "\n")
    
    return resource_name

async def resolve_agent_card_url(resource_name: str):
    """解析智能体卡片URL"""
    PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
    REGION = os.environ["REGION"]
    
    vertexai.init(project=PROJECT_ID, location=REGION)
    client = vertexai.Client(
        project=PROJECT_ID, location=REGION,
        http_options=types.HttpOptions(api_version="v1beta1"),
    )
    
    agent = client.agent_engines.get(name=resource_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(f"RESERVATION_AGENT_CARD_URL={card_url}")
        env_path.write_text("\n".join(lines) + "\n")
    
    return card_url

async def main():
    # 1. 部署预订智能体
    resource_name = await deploy_reservation_agent()
    
    # 2. 解析卡片URL
    card_url = await resolve_agent_card_url(resource_name)
    
    print("\n" + "="*50)
    print("部署完成！")
    print("="*50)
    print(f"预订智能体资源: {resource_name}")
    print(f"A2A卡片URL: {card_url}")

if __name__ == "__main__":
    asyncio.run(main())

8. 完整系统测试

创建端到端测试脚本验证整个多智能体系统：

# scripts/test_full_system.py
import asyncio
import os
from dotenv import load_dotenv

load_dotenv()

async def test_menu_query():
    """测试菜单查询功能"""
    print("测试菜单查询...")
    # 这里实现菜单查询测试逻辑
    print("菜单查询测试完成")

async def test_reservation_flow():
    """测试完整预订流程"""
    print("测试预订流程...")
    
    # 1. 创建预订
    print("1. 创建预订...")
    # 实现预订创建逻辑
    
    # 2. 查询预订
    print("2. 查询预订...")
    # 实现预订查询逻辑
    
    # 3. 取消预订
    print("3. 取消预订...")
    # 实现预订取消逻辑
    
    print("预订流程测试完成")

async def test_multi_agent_coordination():
    """测试多智能体协调"""
    print("测试多智能体协调...")
    
    # 测试协调智能体如何路由请求
    test_cases = [
        "你们有什么意大利菜？",  # 应该路由到菜单查询
        "我想预订明天晚上7点的4人桌",  # 应该路由到预订智能体
        "查询我电话123456的预订",  # 应该路由到预订智能体
        "推荐一些辣的菜品"  # 应该路由到菜单查询
    ]
    
    for query in test_cases:
        print(f"测试查询: {query}")
        # 实现路由测试逻辑
    
    print("多智能体协调测试完成")

async def main():
    print("开始完整系统测试...")
    print("=" * 50)
    
    await test_menu_query()
    print("-" * 30)
    
    await test_reservation_flow()  
    print("-" * 30)
    
    await test_multi_agent_coordination()
    print("-" * 30)
    
    print("所有测试完成！")
    print("=" * 50)

if __name__ == "__main__":
    asyncio.run(main())

9. 常见问题与排查思路

9.1 部署问题排查

问题现象	可能原因	排查方式	解决方案
部署失败，提示权限不足	服务账号缺少必要权限	检查IAM权限设置	授予aiplatform.user角色
智能体卡片无法访问	网络配置问题	检查VPC和防火墙规则	配置正确的网络规则
会话状态不持久	SessionService配置错误	检查环境变量和配置	确保使用VertexAiSessionService

9.2 通信问题排查

# 通信诊断脚本
import asyncio
import aiohttp
import os
from dotenv import load_dotenv

load_dotenv()

async def diagnose_communication():
    """诊断A2A通信问题"""
    card_url = os.environ.get("RESERVATION_AGENT_CARD_URL")
    
    if not card_url:
        print("错误：未找到RESERVATION_AGENT_CARD_URL环境变量")
        return
    
    try:
        async with aiohttp.ClientSession() as session:
            # 测试卡片访问
            async with session.get(card_url) as response:
                if response.status == 200:
                    print("✓ 智能体卡片可访问")
                    card_data = await response.json()
                    print(f"智能体名称: {card_data.get('name')}")
                else:
                    print(f"✗ 卡片访问失败: {response.status}")
                    
    except Exception as e:
        print(f"通信诊断失败: {e}")

if __name__ == "__main__":
    asyncio.run(diagnose_communication())

9.3 状态管理问题

常见状态问题：

1. 状态丢失 ：检查SessionService配置和权限
2. 状态冲突 ：确保使用正确的状态前缀和范围
3. 状态同步延迟 ：验证网络连接和超时设置

10. 最佳实践与工程建议

10.1 智能体设计原则

1. 单一职责原则 ：每个智能体只负责一个明确的业务领域
2. 接口标准化 ：使用A2A协议确保智能体间的标准化通信
3. 状态隔离 ：不同的智能体应该管理独立的状态
4. 错误处理 ：实现完善的错误处理和重试机制

10.2 部署和运维建议

# 生产环境配置示例
监控配置:
  指标监控:
    - 请求延迟
    - 错误率  
    - 并发连接数
  日志记录:
    - 访问日志
    - 错误日志
    - 性能日志
  告警规则:
    - 错误率 > 1%
    - 平均延迟 > 2秒
    - 服务不可用

10.3 安全最佳实践

1. 身份验证 ：使用Google Cloud IAM进行访问控制
2. 网络隔离 ：在VPC内部署敏感服务
3. 数据加密 ：启用CMEK进行数据加密
4. 审计日志 ：启用完整的操作审计

11. 扩展和优化方向

11.1 性能优化

# 智能体性能优化示例
from google.adk.agents import LlmAgent
from google.adk.caching import DiskCache

# 启用响应缓存
optimized_agent = LlmAgent(
    name="optimized_agent",
    model="gemini-3.5-flash",
    cache=DiskCache(),  # 磁盘缓存
    instruction="...",
    tools=[...],
)

11.2 扩展更多智能体

基于现有架构，可以轻松扩展更多专业智能体：

• 支付处理智能体 ：处理订单支付
• 配送跟踪智能体 ：跟踪外卖配送状态
• 客户服务智能体 ：处理客户投诉和咨询

11.3 与DeepSeek模型集成

要将系统适配DeepSeek模型，只需修改模型配置：

# 配置DeepSeek模型
deepseek_agent = LlmAgent(
    name="deepseek_agent",
    model=CustomLLM(  # 自定义DeepSeek模型适配器
        model="deepseek-chat",
        api_key=os.environ.get("DEEPSEEK_API_KEY")
    ),
    instruction="...",
    tools=[...],
)

通过本文的实战教程，你应该已经掌握了多智能体系统开发的核心技能。从MCP工具集成到A2A智能体通信，从本地测试到生产部署，这套架构可以应用于各种复杂的业务场景。记住，良好的智能体设计就像组建一个高效的团队，让每个成员专注自己最擅长的工作，通过标准化协议协同合作，才能发挥最大的效能。

建议将本文代码作为起点，根据你的具体业务需求进行定制和扩展。多智能体架构的未来充满可能性，现在正是深入实践的最佳时机。