# 智能体框架深度对比:谷歌ADK与微软Semantic Kernel技术解析
在人工智能智能体开发领域,谷歌的Action Development Kit (ADK) 和微软的Semantic Kernel (SK) 代表了两种不同的技术路线。这两个框架都在帮助开发者构建能够理解和执行复杂任务的AI智能体,但它们在设计哲学、架构实现和应用场景上存在显著差异。
## 架构设计哲学对比
谷歌ADK采用基于对话流的智能体设计,强调自然语言交互和上下文管理;而Semantic Kernel则更注重将传统编程与AI能力结合,提供插件化的技能组合。
```python
# 架构对比示例
from abc import ABC, abstractmethod
from typing import Dict, List, Any
class AgentFramework(ABC):
"""智能体框架基类"""
@abstractmethod
async def process_request(self, user_input: str, context: Dict) -> Dict[str, Any]:
pass
@abstractmethod
def register_tool(self, tool_name: str, tool_func: callable):
pass
class GoogleADKApproach(AgentFramework):
"""谷歌ADK风格的对话流架构"""
def __init__(self):
self.conversation_flows = {}
self.context_manager = ContextManager()
self.nlu_engine = NaturalLanguageUnderstanding()
async def process_request(self, user_input: str, context: Dict) -> Dict[str, Any]:
# ADK强调对话状态管理
current_state = self.context_manager.get_current_state(context)
# 意图识别和槽位填充
intent_result = await self.nlu_engine.parse_intent(user_input, current_state)
# 对话流路由
flow_handler = self.conversation_flows.get(intent_result.intent)
if flow_handler:
return await flow_handler.handle(intent_result, context)
return await self.default_flow_handler(intent_result, context)
def register_conversation_flow(self, intent: str, flow_handler):
"""注册对话流处理器"""
self.conversation_flows[intent] = flow_handler
class SemanticKernelApproach(AgentFramework):
<"www.hezhou.gov.cn.felli.cn">
<"www.hechi.gov.cn.felli.cn">
<"www.laibin.gov.cn.felli.cn">
"""Semantic Kernel风格的技能组合架构"""
def __init__(self):
self.skills = {}
self.planner = ActionPlanner()
self.memory = VolatileMemory()
async def process_request(self, user_input: str, context: Dict) -> Dict[str, Any]:
# SK强调技能规划和组合
plan = await self.planner.create_plan(user_input, context)
results = []
for step in plan.steps:
skill = self.skills.get(step.skill_name)
if skill:
result = await skill.execute(step.parameters, self.memory)
results.append(result)
self.memory.update(step.skill_name, result)
return await self.planner.aggregate_results(results, plan)
def register_tool(self, skill_name: str, skill_func: callable):
"""注册技能函数"""
self.skills[skill_name] = Skill(skill_name, skill_func)
```
## 谷歌ADK实战开发
ADK专注于构建基于对话的智能体,特别适合客服、助手等场景。
```python
# google_adk_implementation.py
from dataclasses import dataclass
from typing import Dict, List, Optional
import asyncio
<"www.chongzuo.gov.cn.felli.cn">
<"www.cenxi.gov.cn.felli.cn">
<"www.dongxing.gov.cn.felli.cn">
@dataclass
class Intent:
"""意图识别结果"""
name: str
confidence: float
parameters: Dict[str, Any]
fulfillment_text: Optional[str] = None
@dataclass
class Scene:
"""对话场景"""
name: str
slots: Dict[str, Any]
next_scenes: List[str]
class GoogleADKAgent:
"""谷歌ADK风格智能体实现"""
def __init__(self):
self.scenes = {}
self.intent_handlers = {}
self.session_storage = {}
def define_scene(self, scene: Scene):
"""定义对话场景"""
self.scenes[scene.name] = scene
async def handle_user_input(self, session_id: str, user_input: str) -> Dict[str, Any]:
"""处理用户输入"""
# 获取会话状态
session_state = self.session_storage.get(session_id, {})
current_scene = session_state.get('current_scene', 'welcome')
# 意图识别
intent = await self._detect_intent(user_input, current_scene)
# 场景处理
result = await self._process_scene(session_id, current_scene, intent)
# 更新会话状态
self._update_session_state(session_id, result)
return {
'response': result.fulfillment_text,
'next_scene': result.next_scene,
'session_state': self.session_storage[session_id]
<"www.guiping.gov.cn.felli.cn">
<"www.beiliu.gov.cn.felli.cn">
<"www.jingxi.gov.cn.felli.cn">
}
async def _detect_intent(self, user_input: str, current_scene: str) -> Intent:
"""意图检测"""
# 模拟意图识别 - 实际使用Dialogflow等服务
user_input_lower = user_input.lower()
if '预订' in user_input_lower or '预约' in user_input_lower:
return Intent(
name='book_appointment',
confidence=0.9,
parameters={'service_type': self._extract_service_type(user_input)}
)
elif '取消' in user_input_lower:
return Intent(
name='cancel_booking',
confidence=0.85,
parameters={}
)
else:
return Intent(
name='general_inquiry',
confidence=0.7,
parameters={}
)
async def _process_scene(self, session_id: str, scene_name: str, intent: Intent) -> Dict[str, Any]:
"""处理场景"""
scene = self.scenes.get(scene_name)
if not scene:
return await self._handle_unknown_scene(intent)
# 根据意图选择处理逻辑
handler = self.intent_handlers.get(intent.name)
if handler:
return await handler(session_id, scene, intent)
return await self._default_scene_handler(scene, intent)
# ADK对话流配置示例
def setup_booking_flow(agent: GoogleADKAgent):
<"www.pingxiang.gov.cn.felli.cn">
<"www.tongliang.gov.cn.felli.cn">
<"www.lipu.gov.cn.felli.cn">
"""设置预订流程"""
# 定义场景
welcome_scene = Scene(
name="welcome",
slots={},
next_scenes=["service_selection", "help"]
)
service_selection_scene = Scene(
name="service_selection",
slots={"service_type": None, "preferred_time": None},
next_scenes=["time_selection", "back_to_welcome"]
)
time_selection_scene = Scene(
name="time_selection",
slots={"selected_time": None, "customer_name": None},
next_scenes=["confirmation", "change_service"]
)
agent.define_scene(welcome_scene)
agent.define_scene(service_selection_scene)
agent.define_scene(time_selection_scene)
# 注册意图处理器
agent.intent_handlers['book_appointment'] = handle_booking_intent
agent.intent_handlers['cancel_booking'] = handle_cancellation_intent
async def handle_booking_intent(session_id: str, scene: Scene, intent: Intent):
"""处理预订意图"""
service_type = intent.parameters.get('service_type')
if scene.name == "welcome":
return {
'fulfillment_text': f"您想预订什么{service_type}服务?",
'next_scene': 'service_selection',
'updated_slots': {'service_type': service_type}
}
elif scene.name == "service_selection":
return {
'fulfillment_text': f"好的{service_type}服务,请选择预约时间",
'next_scene': 'time_selection',
'updated_slots': {'service_type': service_type}
}
```
## Semantic Kernel核心概念
Semantic Kernel采用技能(Skills)、规划器(Planner)和内存(Memory)的核心架构。
```python
# semantic_kernel_implementation.py
from typing import Dict, List, Any, Callable
from dataclasses import dataclass
import asyncio
<"www.pingguo.gov.cn.felli.cn">
<"www.hengzhou.gov.cn.felli.cn">
<"www.xiuying.gov.cn.felli.cn">
@dataclass
class Skill:
"""技能定义"""
name: str
description: str
function: Callable
parameters: Dict[str, str]
@dataclass
class PlanStep:
"""计划步骤"""
skill_name: str
parameters: Dict[str, Any]
description: str
@dataclass
class ExecutionPlan:
"""执行计划"""
goal: str
steps: List[PlanStep]
context: Dict[str, Any]
class SemanticKernel:
"""Semantic Kernel核心实现"""
def __init__(self):
self.skills = {}
self.memory = {}
self.plugins = {}
def import_skill(self, skill: Skill, plugin_name: str = "default"):
"""导入技能"""
if plugin_name not in self.plugins:
self.plugins[plugin_name] = {}
self.plugins[plugin_name][skill.name] = skill
self.skills[skill.name] = skill
async def create_plan(self, goal: str, context: Dict[str, Any] = None) -> ExecutionPlan:
"""创建执行计划"""
# 使用AI模型进行规划
plan_steps = await self._plan_with_ai(goal, context)
return ExecutionPlan(
goal=goal,
steps=plan_steps,
context=context or {}
)
async def execute_plan(self, plan: ExecutionPlan) -> Dict[str, Any]:
"""执行计划"""
results = {}
for step in plan.steps:
skill = self.skills.get(step.skill_name)
if skill:
try:
# 执行技能
result = await skill.function(**step.parameters)
results[step.skill_name] = result
# 更新记忆
self.memory[step.skill_name] = result
except Exception as e:
results[step.skill_name] = f"Error: {str(e)}"
return {
'goal': plan.goal,
'results': results,
'success': all(not isinstance(r, str) or not r.startswith('Error') for r in results.values())
}
async def _plan_with_ai(self, goal: str, context: Dict[str, Any]) -> List[PlanStep]:
"""使用AI进行规划"""
# 模拟AI规划过程
available_skills = list(self.skills.keys())
# 基于目标分析需要哪些技能
if "天气" in goal:
<"www.longhua.gov.cn.felli.cn">
<"www.qiongshan.gov.cn.felli.cn">
<"www.meilan.gov.cn.felli.cn">
return [
PlanStep(
skill_name="get_location",
parameters={"query": goal},
description="从用户输入中提取位置信息"
),
PlanStep(
skill_name="get_weather",
parameters={"location": "$get_location.result"},
description="获取指定位置的天气信息"
)
]
elif "计算" in goal:
return [
PlanStep(
skill_name="calculate_expression",
parameters={"expression": goal},
description="计算数学表达式"
)
]
return []
# SK技能定义示例
def setup_basic_skills(kernel: SemanticKernel):
"""设置基础技能"""
# 数学计算技能
math_skill = Skill(
name="calculate_expression",
description="计算数学表达式",
function=calculate_math_expression,
parameters={"expression": "数学表达式"}
)
# 天气查询技能
weather_skill = Skill(
name="get_weather",
description="获取天气信息",
function=get_weather_info,
parameters={"location": "地理位置"}
)
# 位置提取技能
location_skill = Skill(
name="get_location",
description="从文本中提取位置信息",
function=extract_location,
parameters={"query": "输入文本"}
)
kernel.import_skill(math_skill, "math_plugin")
kernel.import_skill(weather_skill, "weather_plugin")
kernel.import_skill(location_skill, "utils_plugin")
async def calculate_math_expression(expression: str) -> float:
"""计算数学表达式"""
try:
# 安全地计算表达式
allowed_chars = set('0123456789+-*/.() ')
if all(c in allowed_chars for c in expression):
return eval(expression)
else:
raise ValueError("表达式包含不安全字符")
except Exception as e:
return f"计算错误: {str(e)}"
async def get_weather_info(location: str) -> Dict[str, Any]:
"""获取天气信息"""
<"www.haitang.gov.cn.felli.cn">
<"www.xinbei.gov.cn.felli.cn">
<"www.taoyuan.gov.cn.felli.cn">
# 模拟天气API调用
return {
"location": location,
"temperature": 22.5,
"condition": "晴朗",
"humidity": 65
}
async def extract_location(query: str) -> str:
"""从文本中提取位置"""
# 简单的位置提取逻辑
import re
location_patterns = [
r'在(.+?)的天气',
r'(.+?)的天气',
r'位于(.+?)的'
]
for pattern in location_patterns:
match = re.search(pattern, query)
if match:
return match.group(1).strip()
return "北京" # 默认位置
```
## 功能特性深度对比
从多个维度对比两个框架的核心特性。
```python
# framework_comparison.py
from typing import Dict, List
from dataclasses import dataclass
from enum import Enum
class FrameworkFeature(Enum):
"""框架特性枚举"""
NATURAL_LANGUAGE = "natural_language"
PLANNING = "planning"
MEMORY_MANAGEMENT = "memory_management"
TOOL_INTEGRATION = "tool_integration"
MULTI_TURN_DIALOGUE = "multi_turn_dialogue"
CONTEXT_MANAGEMENT = "context_management"
@dataclass
class FeatureComparison:
"""特性对比"""
feature: FrameworkFeature
google_adk_score: int # 1-5分
semantic_kernel_score: int
description: str
class FrameworkComparator:
"""框架对比器"""
def __init__(self):
self.comparisons = self._initialize_comparisons()
def _initialize_comparisons(self) -> List[FeatureComparison]:
"""初始化对比数据"""
return [
FeatureComparison(
feature=FrameworkFeature.NATURAL_LANGUAGE,
google_adk_score=5,
semantic_kernel_score=3,
description="ADK专为对话设计,SK更关注技能组合"
),
FeatureComparison(
feature=FrameworkFeature.PLANNING,
google_adk_score=2,
semantic_kernel_score=5,
description="SK具备强大的AI规划能力,ADK依赖预定义流程"
),
FeatureComparison(
feature=FrameworkFeature.MEMORY_MANAGEMENT,
google_adk_score=4,
semantic_kernel_score=4,
description="两者都提供会话记忆,实现方式不同"
),
FeatureComparison(
feature=FrameworkFeature.TOOL_INTEGRATION,
google_adk_score=3,
semantic_kernel_score=5,
description="SK的插件化架构更易于工具集成"
),
FeatureComparison(
feature=FrameworkFeature.MULTI_TURN_DIALOGUE,
google_adk_score=5,
semantic_kernel_score=2,
description="ADK专长多轮对话,SK更适合单次任务执行"
),
<"www.taizhong.gov.cn.felli.cn">
<"www.tainan.gov.cn.felli.cn">
<"www.gaoxiong.gov.cn.felli.cn">
FeatureComparison(
feature=FrameworkFeature.CONTEXT_MANAGEMENT,
google_adk_score=5,
semantic_kernel_score=3,
description="ADK提供完整的上下文管理,SK上下文较简单"
)
]
def generate_comparison_report(self) -> Dict[str, Any]:
"""生成对比报告"""
total_adk = sum(comp.google_adk_score for comp in self.comparisons)
total_sk = sum(comp.semantic_kernel_score for comp in self.comparisons)
return {
'total_scores': {
'google_adk': total_adk,
'semantic_kernel': total_sk
},
'detailed_comparisons': [
{
'feature': comp.feature.value,
'adk_score': comp.google_adk_score,
'sk_score': comp.semantic_kernel_score,
'description': comp.description,
'recommendation': self._get_recommendation(comp)
}
for comp in self.comparisons
],
'recommendations': {
'choose_adk_when': [
"构建对话式助手",
"需要复杂的多轮对话管理",
"项目基于谷歌生态系统"
],
'choose_sk_when': [
"需要AI驱动的任务规划",
"集成多种工具和API",
"项目基于微软技术栈",
"需要灵活的技能组合"
]
}
}
def _get_recommendation(self, comparison: FeatureComparison) -> str:
"""获取特性推荐"""
if comparison.google_adk_score > comparison.semantic_kernel_score:
return "更适合谷歌ADK"
elif comparison.google_adk_score < comparison.semantic_kernel_score:
return "更适合Semantic Kernel"
else:
return "两者相当"
```
## 实际应用场景示例
展示两个框架在不同场景下的具体应用。
```python
# use_case_examples.py
import asyncio
from typing import Dict, Any
async def customer_service_scenario():
"""客服场景 - 更适合ADK"""
adk_agent = GoogleADKAgent()
setup_booking_flow(adk_agent)
# 模拟客服对话
conversation = [
"我想预订理发服务",
"这周五下午3点",
"我叫张三",
"确认预订"
]
session_id = "customer_123"
for user_input in conversation:
response = await adk_agent.handle_user_input(session_id, user_input)
print(f"用户: {user_input}")
print(f"助手: {response['response']}")
print("---")
async def task_automation_scenario():
"""任务自动化场景 - 更适合Semantic Kernel"""
<"www.jilong.gov.cn.felli.cn">
<"www.xinzhu.gov.cn.felli.cn">
<"www.jiayi.gov.cn.felli.cn">
kernel = SemanticKernel()
setup_basic_skills(kernel)
# 复杂任务规划执行
tasks = [
"计算北京和上海的平均气温,然后告诉我哪边更暖和",
"帮我规划明天的行程:上午计算项目预算,下午查询天气决定是否外出"
]
for task in tasks:
print(f"任务: {task}")
plan = await kernel.create_plan(task)
result = await kernel.execute_plan(plan)
print(f"执行结果: {result}")
print("---")
class HybridApproach:
"""混合方法 - 结合两者优势"""
def __init__(self):
self.adk_agent = GoogleADKAgent()
self.sk_kernel = SemanticKernel()
self._setup_hybrid_system()
def _setup_hybrid_system(self):
"""设置混合系统"""
# ADK处理对话流程
setup_booking_flow(self.adk_agent)
# SK提供技能支持
setup_basic_skills(self.sk_kernel)
async def process_hybrid_request(self, session_id: str, user_input: str) -> Dict[str, Any]:
"""处理混合请求"""
# 先用ADK进行对话管理
adk_response = await self.adk_agent.handle_user_input(session_id, user_input)
# 如果涉及复杂计算或外部工具,使用SK
if self._needs_complex_processing(user_input):
sk_plan = await self.sk_kernel.create_plan(user_input)
sk_result = await self.sk_kernel.execute_plan(sk_plan)
# 整合结果
return {
'dialogue_response': adk_response['response'],
'task_results': sk_result,
'next_scene': adk_response['next_scene']
}
return {
'dialogue_response': adk_response['response'],
'next_scene': adk_response['next_scene']
}
def _needs_complex_processing(self, user_input: str) -> bool:
"""判断是否需要复杂处理"""
complex_keywords = ['计算', '分析', '查询', '规划', '比较']
return any(keyword in user_input for keyword in complex_keywords)
# 性能基准测试
async def benchmark_frameworks():
"""框架性能基准测试"""
import time
adk_agent = GoogleADKAgent()
sk_kernel = SemanticKernel()
setup_booking_flow(adk_agent)
setup_basic_skills(sk_kernel)
test_cases = [
("简单的对话请求", "你好", "adk"),
("复杂任务规划", "计算(25+37)*2的结果然后查询北京天气", "sk"),
("混合场景", "我想预订服务并计算费用", "hybrid")
]
results = []
for desc, input_text, framework in test_cases:
start_time = time.time()
if framework == "adk":
await adk_agent.handle_user_input("test_session", input_text)
elif framework == "sk":
plan = await sk_kernel.create_plan(input_text)
await sk_kernel.execute_plan(plan)
else:
hybrid = HybridApproach()
await hybrid.process_hybrid_request("test_session", input_text)
execution_time = time.time() - start_time
results.append({
'description': desc,
'framework': framework,
'execution_time': execution_time
})
return results
```
谷歌ADK和微软Semantic Kernel代表了智能体开发的两种重要范式。ADK在对话管理和用户体验方面表现突出,特别适合构建面向消费者的对话式应用;而Semantic Kernel在任务规划、工具集成和复杂问题解决方面更具优势,适合企业级应用和开发人员工具。选择哪个框架取决于具体的使用场景、技术栈偏好和功能需求,在某些情况下,结合两者优势的混合方法可能是最佳选择。