入门指南:Semantic Kernel新手上路完整教程

【免费下载链接】semantic-kernel Integrate cutting-edge LLM technology quickly and easily into your apps 【免费下载链接】semantic-kernel 项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel

前言:为什么选择Semantic Kernel?

还在为AI应用开发中的复杂集成而头疼吗?面对众多LLM模型、插件系统和多智能体协作需求,传统开发方式往往力不从心。Semantic Kernel(语义内核)作为微软开源的AI编排框架,正是为解决这些痛点而生。

通过本教程,你将掌握:

  • ✅ Semantic Kernel核心概念与架构设计
  • ✅ 快速搭建第一个AI智能体应用
  • ✅ 插件系统开发与函数调用技巧
  • ✅ 多智能体协作系统构建方法
  • ✅ 生产环境最佳实践与调试技巧

一、Semantic Kernel核心架构解析

1.1 核心组件关系图

mermaid

1.2 核心概念解析

概念 描述 示例
Kernel(内核) 核心编排引擎,管理所有组件 Kernel.CreateBuilder().Build()
Plugin(插件) 功能模块集合,包含多个函数 天气插件、数据库插件
Function(函数) 具体功能实现单元 获取天气、查询数据
Memory(内存) 语义知识存储与检索系统 向量数据库集成
Planner(规划器) 目标分解与执行规划 多步骤任务自动化

二、环境准备与快速开始

2.1 环境要求与安装

系统要求:

  • Python 3.10+ 或 .NET 8.0+ 或 Java 17+
  • 支持Windows、macOS、Linux系统

安装方式:

# Python安装
pip install semantic-kernel

# .NET安装  
dotnet add package Microsoft.SemanticKernel
dotnet add package Microsoft.SemanticKernel.Agents.Core

# 环境变量配置(二选一)
export AZURE_OPENAI_API_KEY="你的Azure密钥"
export OPENAI_API_KEY="sk-你的OpenAI密钥"

2.2 第一个Hello World应用

Python版本:

import asyncio
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion

async def main():
    # 初始化聊天智能体
    agent = ChatCompletionAgent(
        service=AzureChatCompletion(),
        name="SK助手",
        instructions="你是一个有帮助的AI助手",
    )

    # 获取响应
    response = await agent.get_response("用中文介绍Semantic Kernel")
    print(response.content)

asyncio.run(main())

.NET版本:

using Microsoft.SemanticKernel;
using Microsoft.SemanticKernel.Agents;

var builder = Kernel.CreateBuilder();
builder.AddAzureOpenAIChatCompletion(
    Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT"),
    Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT"), 
    Environment.GetEnvironmentVariable("AZURE_OPENAI_API_KEY")
);
var kernel = builder.Build();

ChatCompletionAgent agent = new()
{
    Name = "SK助手",
    Instructions = "你是一个有帮助的AI助手",
    Kernel = kernel,
};

await foreach (var response in agent.InvokeAsync("用中文介绍Semantic Kernel"))
{
    Console.WriteLine(response.Message);
}

三、插件系统深度解析

3.1 创建自定义插件

from typing import Annotated
from pydantic import BaseModel
from semantic_kernel.functions import kernel_function

class 电商插件:
    @kernel_function(description="查询商品库存信息")
    def 查询库存(self, 商品ID: Annotated[str, "商品编号"]) -> Annotated[str, "库存数量"]:
        # 这里可以连接数据库或API
        库存数据 = {
            "P001": 50,
            "P002": 120, 
            "P003": 0
        }
        return f"商品 {商品ID} 库存: {库存数据.get(商品ID, 0)}件"
    
    @kernel_function(description="获取商品价格")
    def 查询价格(self, 商品ID: Annotated[str, "商品编号"]) -> Annotated[str, "商品价格"]:
        价格数据 = {
            "P001": 299.99,
            "P002": 599.99,
            "P003": 199.99
        }
        return f"商品 {商品ID} 价格: ¥{价格数据.get(商品ID, 0)}"

# 结构化输出模型
class 订单信息(BaseModel):
    订单号: str
    商品列表: list[str]
    总金额: float
    状态: str

3.2 插件集成与使用

from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion, OpenAIChatPromptExecutionSettings
from semantic_kernel.functions import KernelArguments

async def 使用电商插件():
    # 配置结构化输出
    settings = OpenAIChatPromptExecutionSettings()
    settings.response_format = 订单信息
    
    # 创建带插件的智能体
    agent = ChatCompletionAgent(
        service=AzureChatCompletion(),
        name="电商客服",
        instructions="你是专业的电商客服助手,帮助用户查询商品信息和处理订单",
        plugins=[电商插件()],
        arguments=KernelArguments(settings)
    )
    
    # 查询商品信息
    response = await agent.get_response("查询商品P001的库存和价格")
    print(response.content)
    
    # 处理订单查询
    order_response = await agent.get_response("生成一个包含P001和P002的测试订单")
    print(order_response.content)

四、多智能体协作系统

4.1 构建专业智能体团队

from semantic_kernel.agents import ChatCompletionAgent, ChatHistoryAgentThread

# 专业客服智能体
客服智能体 = ChatCompletionAgent(
    service=AzureChatCompletion(),
    name="专业客服",
    instructions="处理客户咨询、订单查询、售后服务等客户服务问题"
)

# 技术支持智能体  
技术支持智能体 = ChatCompletionAgent(
    service=AzureChatCompletion(),
    name="技术支持",
    instructions="解决技术问题、产品使用指导、故障排查等技术支持工作"
)

# 销售智能体
销售智能体 = ChatCompletionAgent(
    service=AzureChatCompletion(),
    name="销售顾问", 
    instructions="产品推荐、价格咨询、促销活动介绍等销售相关工作"
)

# 总调度智能体
总调度智能体 = ChatCompletionAgent(
    service=AzureChatCompletion(),
    name="智能调度",
    instructions="分析用户问题并分发给合适的专业智能体处理,最后汇总结果给用户",
    plugins=[客服智能体, 技术支持智能体, 销售智能体]
)

async def 多智能体协作示例():
    print("欢迎使用智能客服系统!输入'退出'结束对话")
    
    while True:
        用户输入 = input("用户:> ")
        
        if 用户输入.lower() == "退出":
            break
            
        响应 = await 总调度智能体.get_response(用户输入)
        print(f"客服:> {响应.content}")

4.2 智能体协作流程

mermaid

五、实战案例:智能电商客服系统

5.1 完整系统架构

import asyncio
from datetime import datetime
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion

class 电商数据库插件:
    def __init__(self):
        self.订单数据 = {
            "ORD001": {"商品": ["P001", "P002"], "金额": 899.98, "状态": "已发货"},
            "ORD002": {"商品": ["P003"], "金额": 199.99, "状态": "处理中"},
            "ORD003": {"商品": ["P001", "P003"], "金额": 499.98, "状态": "已完成"}
        }
    
    @kernel_function(description="根据订单号查询订单状态")
    def 查询订单状态(self, 订单号: str) -> str:
        订单 = self.订单数据.get(订单号)
        if 订单:
            return f"订单 {订单号}: 金额¥{订单['金额']}, 状态: {订单['状态']}"
        return f"未找到订单 {订单号}"
    
    @kernel_function(description="查询用户的所有订单")
    def 查询用户订单(self, 用户ID: str) -> str:
        # 模拟用户订单查询
        return "用户订单:\n- ORD001: 已发货\n- ORD003: 已完成"

class 物流查询插件:
    @kernel_function(description="查询物流信息")
    def 查询物流(self, 订单号: str) -> str:
        物流信息 = {
            "ORD001": "2024-01-15 已发货,预计1月18日送达",
            "ORD002": "2024-01-16 仓库处理中",
            "ORD003": "2024-01-10 已签收"
        }
        return 物流信息.get(订单号, "暂无物流信息")

# 创建电商客服系统
电商客服系统 = ChatCompletionAgent(
    service=AzureChatCompletion(),
    name="电商智能客服",
    instructions="你是专业的电商客服助手,帮助用户查询订单、物流、商品信息,处理售后问题",
    plugins=[电商数据库插件(), 物流查询插件(), 电商插件()],
    arguments=KernelArguments(
        temperature=0.1,  # 低随机性,保证回答一致性
        max_tokens=1000
    )
)

5.2 系统功能测试

async def 测试电商系统():
    测试用例 = [
        "我的订单ORD001现在到哪里了?",
        "我想查询商品P002的价格和库存",
        "用户U12345有哪些订单?",
        "订单ORD002的状态是什么?",
        "推荐一款性价比高的商品"
    ]
    
    for 用例 in 测试用例:
        print(f"\n用户询问: {用例}")
        响应 = await 电商客服系统.get_response(用例)
        print(f"客服回复: {响应.content}")
        print("-" * 50)

六、最佳实践与性能优化

6.1 配置管理最佳实践

import os
from dotenv import load_dotenv
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion

# 环境变量配置
load_dotenv()

class 配置管理:
    @staticmethod
    def 获取AI服务():
        """获取配置好的AI服务实例"""
        return AzureChatCompletion(
            deployment_name=os.getenv("AZURE_DEPLOYMENT_NAME"),
            endpoint=os.getenv("AZURE_ENDPOINT"),
            api_key=os.getenv("AZURE_API_KEY")
        )
    
    @staticmethod
    def 创建智能体(名称: str, 指令: str, 插件列表: list = None):
        """标准化智能体创建流程"""
        return ChatCompletionAgent(
            service=配置管理.获取AI服务(),
            name=名称,
            instructions=指令,
            plugins=插件列表 or [],
            arguments=KernelArguments(
                temperature=0.1,
                max_tokens=800
            )
        )

6.2 错误处理与重试机制

from tenacity import retry, stop_after_attempt, wait_exponential
from semantic_kernel.exceptions import ServiceResponseException

class 增强型智能体:
    def __init__(self, 基础智能体):
        self.基础智能体 = 基础智能体
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
        retry=retry_if_exception_type(ServiceResponseException)
    )
    async def 安全获取响应(self, 消息: str, 重试次数: int = 3):
        """带重试机制的响应获取"""
        try:
            return await self.基础智能体.get_response(消息)
        except ServiceResponseException as e:
            if 重试次数 <= 0:
                raise e
            print(f"请求失败,剩余重试次数: {重试次数}")
            return await self.安全获取响应(消息, 重试次数 - 1)

6.3 性能监控与日志记录

import logging
from datetime import datetime

# 配置日志系统
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("SemanticKernel")

class 监控装饰器:
    @staticmethod
    def 监控性能(func):
        async def 包装函数(*args, **kwargs):
            开始时间 = datetime.now()
            try:
                结果 = await func(*args, **kwargs)
                耗时 = (datetime.now() - 开始时间).total_seconds()
                logger.info(f"{func.__name__} 执行成功,耗时: {耗时:.2f}s")
                return 结果
            except Exception as e:
                耗时 = (datetime.now() - 开始时间).total_seconds()
                logger.error(f"{func.__name__} 执行失败,耗时: {耗时:.2f}s, 错误: {str(e)}")
                raise
        return 包装函数

# 使用监控装饰器
@监控装饰器.监控性能
async def 安全业务调用(智能体, 消息):
    return await 智能体.get_response(消息)

七、常见问题排查指南

7.1 故障排查表格

问题现象 可能原因 解决方案
认证失败 API密钥错误或过期 检查环境变量,重新生成密钥
模型不可用 部署名称错误或配额不足 验证部署名称,检查配额使用情况
响应超时 网络问题或模型负载高 增加超时时间,添加重试机制
内存溢出 处理数据量过大 优化插件逻辑,分批处理数据
插件加载失败 依赖缺失或版本冲突 检查依赖安装,统一版本

7.2 调试技巧

# 启用详细调试日志
import logging
logging.basicConfig(level=logging.DEBUG)

# 请求追踪装饰器
def 追踪请求(func):
    async def 包装函数(*args, **kwargs):
        print(f"→ 开始请求: {func.__name__}")
        print(f"  参数: {kwargs}")
        结果 = await func(*args, **kwargs)
        print(f"← 请求完成: {func.__name__}")
        print(f"  结果: {结果.content[:100]}...")
        return 结果
    return 包装函数

总结与后续学习路径

通过本教程,你已经掌握了Semantic Kernel的核心概念和实战技能。接下来建议:

  1. 深入插件开发 - 探索更多插件类型和高级功能
  2. 内存系统优化 - 学习向量数据库集成和语义检索
  3. 多模态支持 - 尝试图像和音频处理能力
  4. 生产部署 - 了解容器化部署和性能调优
  5. 社区参与 - 加入开发者社区,贡献代码和案例

记住,Semantic Kernel的真正威力在于将复杂的AI能力变得简单易用。开始你的第一个项目,在实践中不断探索和成长!

提示:在实际项目中,建议从简单用例开始,逐步增加复杂度,并始终关注性能监控和错误处理。

【免费下载链接】semantic-kernel Integrate cutting-edge LLM technology quickly and easily into your apps 【免费下载链接】semantic-kernel 项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel

Logo

更多推荐