当前位置：首页 > news >正文

技术速递｜使用 Semantic Kernel 与 A2A 协议构建多智能体解决方案

news 2025/7/25 9:33:13

作者：卢建晖 - 微软高级云技术布道师

翻译/排版：Alan Wang

在快速发展的 AI 应用开发领域，能够协调多个智能体已成为构建复杂企业级解决方案的关键。虽然单个 AI 智能体擅长特定任务，但复杂的业务场景往往需要跨平台、跨框架甚至跨组织边界的专业智能体协同工作。这正是微软的 Semantic Kernel 编排功能与 A2A（Agent-to-Agent）协议相结合，构建真正互操作多智能体系统的重要基础。

理解 A2A 协议：超越传统工具集成

A2A 协议由 Google 于 2025 年 4 月推出，并得到了超过 50 家技术合作伙伴的支持，旨在解决 AI 生态系统中的一个核心难题：让智能体能够以对等的身份进行沟通与协作，而不仅仅是作为工具被调用。与侧重连接智能体与外部工具和数据源的 MCP（模型上下文协议）不同，A2A 构建了一个专门为智能体间交互设计的标准化通信层。

A2A 的核心能力

通过 Agent 卡片发现智能体：每个支持 A2A 的智能体都会公开一个机器可读的 “Agent 卡片”，这是一个 JSON 文档，用于展示其能力、接入端点、支持的消息类型、认证要求及运行时元数据。这种发现机制使客户端智能体能够动态识别并选择最适合执行特定任务的远程智能体。
通过生命周期跟踪进行任务管理：A2A 的所有交互都围绕有着明确生命周期的独立任务展开，这些任务可以即时完成，也可以持续较长时间，并提供实时状态更新，使该协议既适用于快速的 API 调用，也适用于可能耗时数小时或数天的复杂研究操作。
丰富的消息交换：通信通过结构化消息进行，消息包含具有特定内容类型的“部分”。这使得智能体能够协商合适的交互方式，并交换多种类型的数据，包括文本、结构化 JSON、文件，甚至多媒体流。
企业级安全保障：A2A 构建在常见的 Web 标准之上（如 HTTP、JSON-RPC 2.0、Server-Sent Events），并结合了企业级的身份验证与权限控制，支持 OpenAPI 的认证机制，在不暴露智能体内部状态或专有工具的前提下，实现安全的协作。

A2A 与 MCP：互为补充，而非竞争关系

一个常见的误解是 A2A 与 Anthropic 的模型上下文协议（MCP）存在竞争关系。实际上，这两个协议服务于智能体 AI 技术栈中的不同层面：

MCP 连接智能体与工具和数据 —— 使其能够访问外部 API、数据库、文件系统等结构化资源
A2A 连接智能体与其他智能体 —— 实现对等协作、任务委派和分布式问题解决

你可以把它理解为：MCP 是给智能体一把锤子，而 A2A 是教它如何与一个施工队协同工作。大多数复杂的应用程序都会同时利用这两种协议。

Semantic Kernel：编排引擎

微软的 Semantic Kernel 为构建支持 A2A 的多智能体系统提供了理想的基础。作为一个开源 SDK，Semantic Kernel 在以下方面表现出色：

插件化架构：通过可复用插件轻松扩展智能体能力
多模型支持：协调不同的 AI 模型以执行专门任务
企业集成：无缝连接现有的企业系统与 API
智能体框架：提供实验性但强大的多智能体协同能力

为什么要结合 Semantic Kernel 与 A2A？

框架无关互操作性：Semantic Kernel 智能体可与 LangGraph、CrewAI、Google ADK 或其他 A2A 兼容框架智能体进行通信
保留 Semantic Kernel 的优势：在享受 Semantic Kernel 插件生态与提示工程、企业能力的同时，实现跨平台兼容
渐进迁移路径：现有 Semantic Kernel 应用可逐步采用 A2A，无需大规模重构
云原生设计：支持认证、日志、可观测性，满足企业场景需求

A2A 支持系统的架构模式

在使用 Semantic Kernel 和 A2A 设计多智能体系统时，我们的实现架构展示了几个关键模式：

使用 Azure AI Foundry 的集中路由

我们采用的主要模式是使用由 Azure AI Foundry 驱动的中央路由智能体，智能地将任务委派给各个专业的远程智能体：
在这里插入图片描述

关键组件

主机智能体：基于 Azure AI Agents 的中央路由系统，负责智能决策
A2A 协议：标准化的智能体间通信机制
Semantic Kernel：集成 MCP 的高级智能体框架
远程智能体：使用不同通信协议的专业任务执行者

优势

通过 Azure AI Foundry 的线程实现集中式对话状态管理
基于智能体能力和用户意图进行智能任务委派
在不同智能体交互中提供一致的用户体验
明确的审计记录和完善的错误处理机制

多协议智能体通信

系统展示了不同通信协议如何实现共存：
在这里插入图片描述

通信模式：

A2A HTTP/JSON-RPC：适用于具备标准化发现能力的通用工具型智能体
STDIO：用于基于进程的智能体，如 Playwright 自动化
Server-Sent Events（SSE）：用于 Azure 中的无服务器 MCP 函数

MCP 与 A2A 的混合集成

我们的架构展示了 MCP 协议与 A2A 协议如何相互补充：
在这里插入图片描述

面向工具的 MCP：直接集成 Azure Functions、开发工具和数据源
面向智能体的 A2A：实现智能体间通信与任务委派
Semantic Kernel：作为编排层，桥接两种协议

实施详解

项目结构与组件

我们的多智能体系统采用模块化架构：
在这里插入图片描述

开发环境设置

要构建这个支持 A2A 的 Semantic Kernel 系统，我们需要：

# Initialize Python project with uv (recommended)
uv init multi_agent_system
cd multi_agent_system
# Core dependencies for Semantic Kernel and Azure integration
uv add semantic-kernel[azure]
uv add azure-identity
uv add azure-ai-agents
uv add python-dotenv
# A2A protocol dependencies
uv add a2a-client
uv add httpx
# MCP integration dependencies
uv add semantic-kernel[mcp]
# Web interface dependencies
uv add gradio
# Development dependencies
uv add --dev pytest pytest-asyncio

环境配置

配置以下环境变量：

# Azure AI Foundry configuration
AZURE_AI_AGENT_ENDPOINT=https://your-ai-foundry-endpoint.azure.com
AZURE_AI_AGENT_MODEL_DEPLOYMENT_NAME=Your AI Foundry Model Deployment Name
# Remote agent endpoints
PLAYWRIGHT_AGENT_URL=http://localhost:10001
TOOL_AGENT_URL=http://localhost:10002
# Optional: MCP server configuration
MCP_SSE_URL=http://localhost:7071/runtime/webhooks/mcp/sse

创建中央路由智能体

我们系统的核心是利用 Azure AI Foundry 实现智能任务委派的路由智能体：

import json
import os
import time
import uuid
from typing import Any, Dict, List
import httpx
from a2a.client import A2ACardResolver
from azure.ai.agents import AgentsClient
from azure.identity import DefaultAzureCredential
from dotenv import load_dotenv
class RoutingAgent:"""Central routing agent powered by Azure AI Foundry."""def __init__(self):self.remote_agent_connections = {}self.cards = {}self.agents_client = AgentsClient(endpoint=os.environ["AZURE_AI_AGENT_ENDPOINT"],credential=DefaultAzureCredential(),)self.azure_agent = Noneself.current_thread = Noneasync def initialize(self, remote_agent_addresses: list[str]):"""Initialize with A2A agent discovery."""# Discover remote agents via A2A protocolasync with httpx.AsyncClient(timeout=30) as client:for address in remote_agent_addresses:try:card_resolver= A2ACardResolver(client, address)card = await card_resolver.get_agent_card()from remote_agent_connection import RemoteAgentConnectionsremote_connection = RemoteAgentConnections(agent_card=card, agent_url=address)self.remote_agent_connections[card.name] = remote_connectionself.cards[card.name] = cardexcept Exception as e:print(f'Failed to connect to agent at {address}: {e}')# Create Azure AI agent for intelligent routingawait self._create_azure_agent()async def _create_azure_agent(self):"""Create Azure AI agent with function calling capabilities."""instructions = self._get_routing_instructions()# Define function for task delegationtools = [{"type": "function","function": {"name": "send_message","description": "Delegate task to specialized remote agent","parameters": {"type": "object","properties": {"agent_name": {"type": "string"},"task": {"type": "string"}},"required": ["agent_name", "task"]}}}]model_name = os.environ.get("AZURE_AI_AGENT_MODEL_DEPLOYMENT_NAME", "gpt-4")self.azure_agent = self.agents_client.create_agent(model=model_name,name="routing-agent",instructions=instructions,tools=tools)self.current_thread = self.agents_client.threads.create()print(f"Routing agent initialized: {self.azure_agent.id}")def _get_routing_instructions(self) -> str:"""Generate context-aware routing instructions."""agent_info = [{'name': card.name, 'description': card.description}for card in self.cards.values()]return f"""You are an intelligent routing agent for a multi-agent system.
Available Specialist Agents:
{json.dumps(agent_info, indent=2)}

路由指引：

网页自动化、截图、浏览器任务 → Playwright 智能体
开发任务、文件操作、代码库管理 → 工具智能体
委派任务时务必提供完整的任务上下文
向用户说明哪个智能体正在处理他们的请求

构建集成 MCP 的专业智能体

我们的远程智能体利用 Semantic Kernel 的 MCP 集成，实现可扩展能力：

from semantic_kernel.agents import AzureAIAgent, AzureAIAgentSettings
from semantic_kernel.connectors.mcp import MCPStdioPlugin, MCPSsePlugin
from azure.identity.aio import DefaultAzureCredential
class SemanticKernelMCPAgent:"""Specialized agent with MCP plugin integration."""def __init__(self):self.agent = Noneself.client = Noneself.credential = Noneself.plugins = []async def initialize_playwright_agent(self):"""Initialize with Playwright automation via MCP STDIO."""try:self.credential= DefaultAzureCredential()self.client = await AzureAIAgent.create_client(credential=self.credential).__aenter__()# Create Playwright MCP pluginplaywright_plugin = MCPStdioPlugin(name="Playwright",command="npx",args=["@playwright/mcp@latest"],)await playwright_plugin.__aenter__()self.plugins.append(playwright_plugin)# Create specialized agentagent_definition = await self.client.agents.create_agent(model=AzureAIAgentSettings().model_deployment_name,name="PlaywrightAgent",instructions=("You are a web automation specialist. Use Playwright to ""navigate websites, take screenshots, interact with elements, ""and perform browser automation tasks."),)self.agent = AzureAIAgent(client=self.client,definition=agent_definition,plugins=self.plugins,)except Exception as e:await self.cleanup()raiseasync def initialize_tools_agent(self, mcp_url: str):"""Initialize with development tools via MCP SSE."""try:self.credential= DefaultAzureCredential()self.client = AzureAIAgent.create_client(credential=self.credential)# Create development tools MCP plugintools_plugin = MCPSsePlugin(name="DevTools",url=mcp_url,)await tools_plugin.__aenter__()self.plugins.append(tools_plugin)agent_definition = await self.client.agents.create_agent(model=AzureAIAgentSettings().model_deployment_name,name="DevAssistant",instructions=("You are a development assistant. Help with repository ""management, file operations, opening projects in VS Code, ""and other development tasks."),)self.agent = AzureAIAgent(client=self.client,definition=agent_definition,plugins=self.plugins,)except Exception as e:await self.cleanup()raiseasync def invoke(self, user_input: str) -> dict[str, Any]:"""Process tasks through the specialized agent."""if not self.agent:return{'is_task_complete': False,'content': 'Agent not initialized.',}try:responses = []async for response in self.agent.invoke(messages=user_input,thread=self.thread,):responses.append(str(response))self.thread = response.threadreturn {'is_task_complete': True,'content': "\n".join(responses) or "No response received.",}except Exception as e:return {'is_task_complete': False,'content': f'Error: {str(e)}',}

基于 Gradio 的 Web 界面

系统提供了由 Gradio 驱动的现代聊天界面：

import asyncio
import gradio as gr
from routing_agent import RoutingAgent
async def get_response_from_agent(message: str, history: list[gr.ChatMessage]
) -> gr.ChatMessage:"""Process user messages through the routing system."""global ROUTING_AGENTtry:response= 
awaitROUTING_AGENT.process_user_message(message)return gr.ChatMessage(role="assistant", content=response)except Exception as e:return gr.ChatMessage(role="assistant",content=f"❌ Error: {str(e)}")
async def main():"""Launch the multi-agent system."""# Initialize routing agentglobal ROUTING_AGENTROUTING_AGENT= 
awaitRoutingAgent.create([os.getenv('PLAYWRIGHT_AGENT_URL', 'http://localhost:10001'),os.getenv('TOOL_AGENT_URL', 'http://localhost:10002'),])ROUTING_AGENT.create_agent()# Create Gradio interfacewith gr.Blocks(theme=gr.themes.Ocean()) as demo:gr.Markdown("# 🤖 Azure AI Multi-Agent System")gr.ChatInterface(get_response_from_agent,title="Chat with AI Agents",examples=["Navigate to github.com/microsoft and take a screenshot","Clone repository https://github.com/microsoft/semantic-kernel","Open the cloned project in VS Code",])demo.launch(server_name="0.0.0.0", server_port=8083)

部署与运维

本地开发环境：

# Start MCP server (Azure Functions)
cd mcp_sse_server/MCPAzureFunc
func start

# Start remote agents in separate terminals
cd remote_agents/playwright_agent && uv run .
cd remote_agents/tool_agent && uv run .

# Start the host agent with web interface
cd host_agent && uv run .

生产环境注意事项：

将每个智能体作为独立微服务部署，推荐使用 Azure Container Apps
利用 Azure Service Bus 实现可靠的智能体发现与通信
使用 Azure Application Insights 实施全面的日志记录
配置合适的身份验证和网络安全措施

实际应用示例

基于我们的实现，以下是多智能体系统处理不同类型请求的实用示例：

示例一：网页自动化任务

User: "Navigate to github.com/microsoft and take a screenshot"
Flow:
1. Host Agent (Azure AI) analyzes the request
2. Identifies this as a web automation task  
3. Delegates to Playwright Agent via A2A protocol
4. Playwright Agent uses MCP STDIO to execute browser automation
5. Returns screenshot and navigation details to user

示例二：开发工作流

User: "Clone https://github.com/microsoft/semantic-kernel and open it in VS Code"
Flow:
1. Host Agent recognizes repository management + IDE operation
2. Delegates to Tool Agent via A2A protocol  
3. Tool Agent uses MCP SSE connection to Azure Functions
4. Executes git clone and VS Code launch commands
5. Reports success status back to user