智能体框架深度对比:谷歌ADK与微软Semantic Kernel技术解析

# 智能体框架深度对比:谷歌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在任务规划、工具集成和复杂问题解决方面更具优势,适合企业级应用和开发人员工具。选择哪个框架取决于具体的使用场景、技术栈偏好和功能需求,在某些情况下,结合两者优势的混合方法可能是最佳选择。


请使用浏览器的分享功能分享到微信等