AI大模型全栈工程师课程笔记 - Semantic Kernel

Michael阿明

发布于 2023-12-18 13:38:18

88300

代码可运行

文章被收录于专栏：Michael阿明学习之路

运行总次数：0

代码可运行

如果侵权，请联系删除，感谢！

文章目录
1. Semantic Kernel
2. 创建SK
3. 创建 `semantic Function`
4. 调用 Semantic Function
5. Native Functions
6. 用 SKContext 实现多参数 Functions
7. Plugins/Skills
8. 调用 Pipeline
9. 函数嵌套调用
10. Memory
11. Planner
12. SK 调试插件

1. Semantic Kernel

大语言模型开发框架（SDK）
封装了一系列工具（提示词模板、链式调用、规划能力等），像调用函数一样
支持（C# Python Java）基于 OpenAI / Azure OpenAI / Huggingface 开发 LLM 应用

https://learn.microsoft.com/en-us/semantic-kernel/overview/

https://github.com/microsoft/semantic-kernel

在 SK 中，一组 Function 组成一个技能（Skill/Plugin）。
运行 Skill/Plugin，需要有一个配置和管理的单元，这个组织管理单元就是 Kernel
Kernel 负责管理底层接口与调用顺序，例如：OpenAI/Azure OpenAI 的授权信息、默认的 LLM 模型选择、对话上下文、技能参数的传递等

配置 .env，OpenAI 和 Azure，配置好一个就行，安装 pip install semantic-kernel

OPENAI_API_KEY="sk-***"
OPENAI_BASE_URL=""

AZURE_OPENAI_DEPLOYMENT_NAME=""
AZURE_OPENAI_ENDPOINT=""
AZURE_OPENAI_API_KEY=""

2. 创建SK

创建 kernel，添加 models

import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
import os

# 加载 .env 到环境变量
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv('../utils/.env'))

# 创建 semantic kernel
kernel = sk.Kernel()

# 配置 OpenAI 服务。OPENAI_BASE_URL 会被自动加载生效
api_key = os.getenv('OPENAI_API_KEY')
model = OpenAIChatCompletion(
    "gpt-3.5-turbo",
    api_key
)

# 把 LLM 服务加入 kernel
# 可以加多个。第一个加入的会被默认使用，非默认的要被指定使用
kernel.add_text_completion_service("michael_chatbot", [model])

3. 创建 `semantic Function`

由 prompt + 配置文件构成，语言无关
skFunction包含文件： skprompt.txt: 存放 prompt，可以包含参数，还可以调用其它函数 config.json: 存放配置，包括函数功能，参数的类型，以及调用大模型时的参数
skprompt.txt

将用户的指令转换成 Linux 命令
只输出命令本身，不要分析，不要评论。

{{$input}}

config.json ，其中的 type 只有 "completion" 和 "embedding" 两种

{
    "schema": 1,
    "type": "completion",
    "description": "将用户的指令转换成 Linux 命令",
    "completion": {
        "max_tokens": 256,
        "temperature": 0,
        "top_p": 0,
        "presence_penalty": 0,
        "frequency_penalty": 0
    },
    "input": {
        "parameters": [
            {
                "name": "input",
                "description": "用户的指令",
                "defaultValue": ""
            }
        ]
    }
}

4. 调用 Semantic Function

import asyncio
# 加载 semantic function。注意目录结构
my_skill = kernel.import_semantic_skill_from_directory(
    "./sk_samples/", "SamplePlugin")

# 运行 function 看结果


async def run_function():
    # 运行 skill 看结果
    result = await kernel.run_async(
        my_skill["GenerateCommand"],
        input_str="将系统日期设为2023-04-01",
    )
    return result

result = await run_function()
print(result)  # date -s "2023-04-01"

官方例子参考：https://github.com/microsoft/semantic-kernel/tree/main/samples/skills

5. Native Functions

在 SK 中，Semantic Function 和 Native Function 被 Kernel 平等对待

在 sk_samples/SamplePlugin/native_function.py 中编写原生的函数

from semantic_kernel.skill_definition import sk_function, sk_function_context_parameter
from semantic_kernel.orchestration.sk_context import SKContext
import json

class CommandVerifier:
    @sk_function(
        description="检查命令是否合法",
        name="verifyCommand",
    )
    def verify(self, command: str) -> str:
        if ">" in command:
            return "非法"
        parts = command.replace(';', '|').split('|')
        for cmd in parts:
            name = cmd.split(" ")[0]
            if name not in ["ls","cat","head","tail","echo"]:
                return "非法"
        return "合法"

import_native_skill_from_directory调用

my_skill = kernel.import_native_skill_from_directory(
    "./sk_samples/", "SamplePlugin")
async def run_function():
    # 运行 skill 看结果
    result = await kernel.run_async(
        my_skill["verifyCommand"],
        input_str='date -s "2023-04-01"',
    )
    return result

result = await run_function()
print(result)  # 非法

如果导入了sk_function装饰了的这个类，可以import_skill

# 加载 native function
verify_skill = kernel.import_skill(CommandVerifier(), "Verifier")

# 看结果
result = await kernel.run_async(
    verify_skill["verifyCommand"],
    input_str='date -s "2023-04-01"',
    # input_str="ls -l ./",
)

print(result)  # 非法

6. 用 SKContext 实现多参数 Functions

默认一个参数的话，可以使用 {{$input}}，会被默认赋值

多参数Semantic Function写法：

# Prompt 模板
sk_prompt = """
讲一个{{$topic1}}和{{$topic2}}的一句话笑话
"""

# 创建 Semantic Function
joke = kernel.create_semantic_function(sk_prompt)

# 创建 SKContext
context = kernel.create_new_context()

# 变量赋值
context["topic1"] = "农夫"
context["topic2"] = "蛇"

# 看结果
result = await kernel.run_async(
    joke,
    input_context=context
)
print(result)
# 农夫对蛇说：“你是唯一一个我不会踩死的蛇，因为我怕你会咬我！”

多参数 Native Function 的写法：@sk_function，@sk_function_context_parameter

from semantic_kernel.skill_definition import sk_function, sk_function_context_parameter
from semantic_kernel.orchestration.sk_context import SKContext


class Math:
    @sk_function(
        description="加法",
        name="add",
    )
    @sk_function_context_parameter(
        name="number1",
        description="被加数",
    )
    @sk_function_context_parameter(
        name="number2",
        description="加数",
    )
    def add(self, context: SKContext) -> str:
        return str(float(context["number1"]) + float(context["number2"]))

    @sk_function(
        description="减法",
        name="minus",
    )
    @sk_function_context_parameter(
        name="number1",
        description="被减数",
    )
    @sk_function_context_parameter(
        name="number2",
        description="减数",
    )
    def minus(self, context: SKContext) -> str:
        return str(float(context["number1"]) - float(context["number2"]))

# 加载 native function
math_skill = kernel.import_skill(Math(), "Math")


# 创建 SKContext
context = kernel.create_new_context()

# 变量赋值
context["number1"] = 1024
context["number2"] = 65536

# 看结果
result = await kernel.run_async(
    math_skill["add"],
    input_context=context
)
print(f"加法计算结果：{result}")

result = await kernel.run_async(
    math_skill["minus"],
    input_context=context
)
print(f"减法计算结果：{result}")

在 SK 中，向函数传递多个参数，需要创建一个 SKContext 对象

create_new_context，context 中的变量赋值
input_context=context，传入 run_async

7. Plugins/Skills

两个词是一个概念，意思是一组函数的集合，可包含 语义函数SF（提示工程）， 原生函数NF（function call）

SK 的 plugin 会和 ChatGPT、Bing、Microsoft 365 通用

一些内置的 plugin (skill)

from semantic_kernel.core_skills import SkillName

见 https://github.com/microsoft/semantic-kernel/blob/main/python/semantic_kernel/core_skills/

ConversationSummarySkill - 生成对话的摘要
FileIOSkill - 读写文件
HttpSkill - 发出 HTTP 请求，支持 GET、POST、PUT 和 DELETE
MathSkill - 加法和减法计算
TextMemorySkill - 保存文本到 memory 中，可以对其做向量检索
TextSkill - 把文本全部转为大写或小写，去掉头尾的空格（trim）
TimeSkill - 获取当前时间及用多种格式获取时间参数
WaitSkill - 等待指定的时间
WebSearchEngineSkill - 在互联网上搜索给定的文本

8. 调用 Pipeline

没有明确的定义，LLM，memory，functions，组合在一起就是 pipeline 了

# 加载 semantic function。注意目录结构
command_skill = kernel.import_semantic_skill_from_directory(
    "./sk_samples/", "SamplePlugin"
)

# 加载 native function
verify_skill = kernel.import_skill(CommandVerifier(), "Verifier")

# 看结果
result = await kernel.run_async(
    command_skill["GenerateCommand"],
    verify_skill["verifyCommand"],
    input_str="显示 key.txt 文件的内容",
)

print(result)  # 合法
# 输入 删除所有根目录文件， 输出 非法

可以看到，sk 顺序调用了两个 skill，先生成命令，然后检查是否非法

9. 函数嵌套调用

Semantic Function 嵌套调用

sk_samples\NestedSample\GenerateCommand\skprompt.txt

已知，判断用户指令是否为合法指令的结果是：
{{NestedSample.VerifyCommand $input}}

根据以上结果，执行下述动作之一：
如果用户指令为非法，向用户说明该指令不合法及原因；
否则，将用户的指令转换成 Linux 命令。输出命令本身，不要分析，不要评论。

用户指令：{{$input}}

# 加载 semantic function。注意目录结构
command_skill = kernel.import_semantic_skill_from_directory(
    "./sk_samples/", "NestedSample"
)

# 看结果
result = await kernel.run_async(
    command_skill["GenerateCommand"],
    input_str="显示 example.txt 文件的内容",
)

print(result)  # cat example.txt
# 输入：删除当前目录
# 输出：该指令为非法指令，因为删除当前目录会对系统状态和现有文件进行修改。请重新输入合法指令。

Native Function 嵌套调用

import json
from semantic_kernel.skill_definition import sk_function, sk_function_context_parameter
from semantic_kernel.orchestration.sk_context import SKContext
from semantic_kernel.orchestration.context_variables import ContextVariables


class Calculator:
    def __init__(self, kernel):
        self._kernel = kernel  # 初始化时传入 kernel

    @sk_function(
        description="加减计算器",
        name="calc",
    )
    async def calc(self, query: str) -> str:
        # 嵌套调用 Semantic Function
        q2f = self._kernel.skills.get_function(
            "NestedSample", "Query2Function")  # 获取函数
        json_str = (
            await self._kernel.run_async(q2f, input_str=query)
        ).result.strip()  # 获取调用的结果，Query2Function 要求输出的 json
        json_obj = json.loads(json_str)
        func_name = json_obj["name"]  # 获取输出的函数名
        context = self._kernel.create_new_context()
        context["number1"] = json_obj["number1"] # 添加输入
        context["number2"] = json_obj["number2"]
        # 嵌套调用 Native Function
        math_func = self._kernel.skills.get_function("Math", func_name)
        result = (
            await self._kernel.run_async(math_func, input_context=context)
        ).result.strip()  # 调用 Math 类定义的函数
        return result

import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAITextEmbedding
import os

# 加载 .env 到环境变量
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv('../utils/.env'))

# 创建 semantic kernel
kernel = sk.Kernel()

# 配置 OpenAI 服务
api_key = os.getenv('OPENAI_API_KEY')
model = OpenAIChatCompletion("gpt-3.5-turbo", api_key)

# 把 LLM 服务加入 kernel
# 可以加多个。第一个加入的会被默认使用，非默认的要被指定使用
kernel.add_text_completion_service("your-gpt3", model)

# 加载 math skill
kernel.import_skill(Math(), "Math")

# 加载 nested skills
kernel.import_semantic_skill_from_directory(
    "./sk_samples/", "NestedSample")

# 加载 calucator skill
# 初始化时传入 kernel
skill = kernel.import_skill(Calculator(kernel), "Calculator")

result = await kernel.run_async(
    skill["calc"],
    input_str="1024减去256等于多少"
)
print(result)  # 768.0

10. Memory

用 kernel.add_text_embedding_generation_service() 添加一个文本向量生成服务
用 kernel.register_memory_store() 注册一个 memory store，可以是内存、文件、向量数据库等
用 kernel.memory.save_information_async() 保存信息到 memory store
用 kernel.memory.search_async() 搜索信息

演示基于文档对话

添加embedding服务

import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAITextEmbedding
import os

# 加载 .env 到环境变量
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv('../utils/.env'))

# 创建 semantic kernel
kernel = sk.Kernel()

# 配置 OpenAI 服务
api_key = os.getenv('OPENAI_API_KEY')
model = OpenAIChatCompletion("gpt-3.5-turbo", api_key)

# 把 LLM 服务加入 kernel
# 可以加多个。第一个加入的会被默认使用，非默认的要被指定使用
kernel.add_text_completion_service("your-gpt3", model)

# 添加 embedding 服务
kernel.add_text_embedding_generation_service(
    "ada", OpenAITextEmbedding("text-embedding-ada-002", api_key))

文本向量化 https://learn.microsoft.com/en-us/semantic-kernel/memories/vector-db

from semantic_kernel.text import split_markdown_lines

# 使用内存做 memory store
kernel.register_memory_store(memory_store=sk.memory.VolatileMemoryStore())

# 读取文件内容
with open('ChatALL.md', 'r', encoding='utf-8') as f:
    content = f.read()

# 将文件内容分片，单片最大 100 token（注意：SK 的 text split 功能目前对中文支持不如对英文支持得好）
lines = split_markdown_lines(content, 100)

# 将分片后的内容，存入内存
for index, line in enumerate(lines):
    await kernel.memory.save_information_async("chatall", id=index, text=line)

向量搜索

result = await kernel.memory.search_async("chatall", "ChatALL怎么下载？")
print(result[0].text)

11. Planner

用于 Agent 智能体开发

https://github.com/microsoft/semantic-kernel/tree/main/python/semantic_kernel/planning

使用 planner 的步骤：

把 plugin 注册到 kernel
把 kernel 当参数，实例化某个 planner
调用 planner 的 create_plan_async() 方法获得 plan
调用 plan 的 invoke_async() 方法执行 plan (注意，不同 planner 接口并不一致)

做一个例子 周杰伦的生日是星期几：先申请 bingAPI https://portal.azure.com

创建SK，添加模型

from semantic_kernel.core_skills import WebSearchEngineSkill
from semantic_kernel.connectors.search_engine import BingConnector
from semantic_kernel.planning import SequentialPlanner
import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
import os

# 加载 .env 到环境变量
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv('../utils/.env'))

# 创建 semantic kernel
kernel = sk.Kernel()

# 配置 OpenAI 服务
api_key = os.getenv('OPENAI_API_KEY')
model = OpenAIChatCompletion("gpt-4", api_key)

# 把 LLM 服务加入 kernel
# 可以加多个。第一个加入的会被默认使用，非默认的要被指定使用
kernel.add_text_completion_service("your-gpt4", model)

编写SK函数

import calendar
import dateutil.parser as parser
from datetime import date
from semantic_kernel.skill_definition import sk_function


class DayOfWeek:
    @sk_function(
        description="计算输入日期是星期几",
        name="weekday",
    )
    def weekday(self, date_str: str) -> str:
        """Convert date to weekday name"""
        d = parser.parse(date_str)
        return calendar.day_name[d.weekday()]

编写SK函数

sk_prompt = """
抽取下述文本中第一个出现的日期。
以yyyy-MM-dd格式输出。只输出日期，不要输出其他字符

{{$input}}
"""
kernel.create_semantic_function(
    sk_prompt,
    function_name="parseDate",
    skill_name="DateParser",
    description="抽取输入文本中的日期"
)

导入技能

# 导入搜索 plugin
connector = BingConnector(api_key=os.getenv("BING_API_KEY"))
kernel.import_skill(WebSearchEngineSkill(connector), "WebSearch")

kernel.import_skill(
    DayOfWeek(), "DayOfWeek"
)

生成plan，并执行

# 创建 planner
planner = SequentialPlanner(kernel)

# 开始
query = "周杰伦的生日是星期几？"
plan = await planner.create_plan_async(goal=query)

result = await plan.invoke_async()

# 打印步骤用来调试
for index, step in enumerate(plan._steps):
    print("Step:", index)
    print("Description:", step.description)
    print("Function:", step.skill_name + "." + step._function.name)
    if len(step._outputs) > 0:
        print("  Output:\n", str.replace(
            result[step._outputs[0]], "\n", "\n  "))


print(result)

输出：

Step: 0
Description: Performs a web search for a given query
Function: WebSearch.searchAsync
  Output:
 ['0 周杰伦（Jay Chou），1979年1月18日出生于台湾省新北市，祖籍福建省泉州市永春县，华语流行乐男歌手、音乐人、演员、导演、编剧，毕业于 淡江中学 。 2000年，发行个人首张音乐专辑《 Jay 》 [27] 。 2001年，凭借专辑《 范特西 》奠定其融合中西方音乐的风格 [16] 。 2002年，举行“The One”世界巡回演唱会 [1] 。 2003年，成为美国《 时代周刊 》封面人物 [2] ；同年，发行音乐专辑《 叶惠美 》 [21] ，该专辑亚洲首月销量达到200万张 [22] ，并获得 第15届台湾金曲奖 最佳流行音乐演唱专辑奖 [23] 。 2004年，发行音乐专辑《 七里香 》 [30] ，该专辑亚洲首月销量达到300万张 [329] 。']
Step: 1
Description: 抽取输入文本中的日期
Function: DateParser.parseDate
  Output:
 1979-01-18
Step: 2
Description: 计算输入日期是星期几
Function: DayOfWeek.weekday
  Output:
 Thursday
Thursday

先进行搜索，再解析日期，再求星期几