【Semantic Kernel核心组件】planner：大模型时代的智能任务编排引擎-EW帮帮网

层级	功能组件	技术实现
意图解析层	自然语言理解	GPT-4等LLM进行意图分类与实体提取
技能发现层	插件元数据	扫描注册插件的函数描述、输入输出参数
计划生成层	XML规划器	生成符合Handlebars模板的XML执行计划
执行优化层	运行时引擎	动态绑定参数、并行执行与结果聚合

2.2 关键技术特性

多级缓存策略：对高频调用的插件函数进行结果缓存（参考网页8的Redis集成案例）
自适应重试机制：当函数执行失败时，自动尝试备用插件或调整参数组合
执行追踪系统：通过OpenTelemetry输出详细日志，支持性能分析与错误溯源

三、典型应用场景

3.1 复杂计算任务

# 用户请求："小王月薪3000元，半年后涨薪15%，计算一年后的总储蓄"
plan = await planner.create_plan(question)
# 生成计划：
# 1. 计算涨薪后薪资 math.multiply(3000, 1.15)
# 2. 计算年度总收入 math.add(新薪资*12, 绩效奖金)
# 3. 扣除生活开支 math.subtract(总收入, 开销总额)

3.2 智能服务编排

在天气服务案例中，Planner成功协调了三个插件：

意图识别插件：解析用户自然语言请求
天气API插件：获取实时气象数据
邮件发送插件：生成并发送穿衣建议

3.3 企业级工作流

某金融客户使用Planner构建的贷款审批系统：

用户咨询 -> 资质验证 -> 风险模型计算 -> 文档生成 -> 通知发送

通过插件组合实现全流程自动化，处理效率提升400%

四、开发最佳实践

4.1 插件设计规范

1.语义化描述：每个函数需包含<description>和<input_description>元数据

@sk_function(
    description="计算复利终值",
    input_description="本金,年利率,投资年限"
)
def compound_interest(context: SKContext):
    ...

2.参数验证：在函数入口处添加类型检查与范围校验（参考网页6的CommandVerifier）

3.版本兼容：通过[PluginName]_vX命名规范管理插件迭代

4.2 提示工程优化

采用分级提示策略提升计划质量：

<!-- Handlebars模板片段 -->
{{~! 步骤1: 验证输入参数合法性}}
{{set "initialInvestment" 2130.23}}
{{~! 步骤2: 调用增长率计算函数}}
{{MathPlugin.Multiply input=initialInvestment factor=1.23}}

通过添加以下约束条件减少LLM幻觉：

强制使用<seeAlso>标签引用已知插件
限制每个步骤最多调用两个函数
添加输入参数类型校验声明

4.3 性能调优技巧

计划缓存：对高频请求的MD5哈希值进行缓存，有效期设置5-10分钟
并行执行：对无依赖关系的步骤启用Parallel.ForEach（参考网页12）
资源隔离：通过Kubernetes Namespace隔离CPU密集型插件

五、演进方向与挑战

尽管Planner大幅提升了开发效率，但仍面临以下挑战：

长链依赖问题：超过7个步骤的计划准确率下降至68%
实时性约束：金融等场景要求亚秒级响应，需要引入预编译计划机制
安全边界：防止恶意提示注入导致插件滥用（

未来演进将聚焦：

计划验证器：通过形式化验证确保执行路径安全性
自适应优化器：基于历史执行数据动态调整计划策略
多模态扩展：支持图像处理、语音识别等新型插件类型

六、实例代码

运行流程图

manthpluginmore.py

说明：该代码主要是定义不同的计算方法，方便调用

import math
from semantic_kernel.skill_definition import (
    sk_function,
    sk_function_context_parameter,
)
from semantic_kernel.orchestration.sk_context import SKContext

class MathPlugin:
    @sk_function(
        description="计算一个数的平方根",
        name="square_root",
        input_description="要计算平方根的数值",
    )
    def square_root(self, number: str) -> str:
        print(f"[DEBUG] square_root 输入: {number}")
        return str(math.sqrt(float(number)))

    @sk_function(
        description="将两个数相加",
        name="add",
    )
    @sk_function_context_parameter(
        name="input",
        description="第一个加数",
    )
    @sk_function_context_parameter(
        name="number2",
        description="第二个加数",
    )
    def add(self, context: SKContext) -> str:
        input_val = context["input"]
        number2_val = context["number2"]
        print(f"[DEBUG] add 输入: {input_val} + {number2_val}")
        return str(float(input_val) + float(number2_val))
    
    @sk_function(
        description="执行两个数的减法运算",
        name="subtract",
    )
    @sk_function_context_parameter(
        name="input",
        description="被减数",
    )
    @sk_function_context_parameter(
        name="number2",
        description="减数",
    )
    def subtract(self, context: SKContext) -> str:
        input_val = context["input"]
        number2_val = context["number2"]
        print(f"[DEBUG] subtract 输入: {input_val} - {number2_val}")
        return str(float(input_val) - float(number2_val))

    @sk_function(
        description="计算两个数的乘积",
        name="multiply",
    )
    @sk_function_context_parameter(
        name="input",
        description="第一个乘数",
    )
    @sk_function_context_parameter(
        name="number2",
        description="第二个乘数",
    )
    def multiply(self, context: SKContext) -> str:
        input_val = context["input"]
        number2_val = context["number2"]
        print(f"[DEBUG] multiply 输入: {input_val} * {number2_val}")
        return str(float(input_val) * float(number2_val))

    @sk_function(
        description="将两个数相除",
        name="divide",
    )
    @sk_function_context_parameter(
        name="input",
        description="被除数",
    )
    @sk_function_context_parameter(
        name="number2",
        description="除数",
    )
    def divide(self, context: SKContext) -> str:
        input_val = context["input"]
        number2_val = context["number2"]
        print(f"[DEBUG] divide 输入: {input_val} / {number2_val}")
        return str(float(input_val) / float(number2_val))
    
    
'''
使用了准确的中文数学术语：
加数（addend）
被减数（minuend）
减数（subtrahend）
乘数（multiplier）
被除数（dividend）
除数（divisor）
'''

message.py

该代码为运行程序，通过planner，

import os
from dotenv import load_dotenv
import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.planning.sequential_planner import SequentialPlanner
import asyncio
from mathpluginmore import MathPlugin

#1、加载配置环境
load_dotenv()

#2、加载sk框架
kernel=sk.Kernel()

#3、将千问模型注册到sk框架中
kernel.add_chat_service(
        "qwen-max",
        OpenAIChatCompletion(
                model_id="qwen-max",
                api_key=os.getenv("DASHSCOPE_API_KEY"),
                endpoint="https://dashscope.aliyuncs.com/compatible-mode/v1"
                )
        )

#4、将mathpluginmore.py中的MathPlugin注册到sk中
math_plugin=kernel.import_skill(MathPlugin(),skill_name="MathPlugin")

#5、创建plan实例
planner=SequentialPlanner(kernel)

#6、定义问题
ask="""如果小王一个月底薪是3000元，半年后底薪会上涨15%，提成绩效为5000元，有2000生活开销，半年后小王一个月可以存多少钱"""

#7、主程序
async def main():
        #①创建计划
        planing=await planner.create_plan_async(ask)
        
        #②将存储计划的分步执行逻辑遍历
        for step in planing._steps:
                print(step.description,":",step._state.__dict__)
        
        #③异步调用执行
        result=await planing.invoke_async()
        print(result)

#8、异步执行
asyncio.run(main())

运行结果

计算两个数的乘积 : {'variables': {'input': ''}}
将两个数相加 : {'variables': {'input': ''}}
执行两个数的减法运算 : {'variables': {'input': ''}}
[DEBUG] multiply 输入: 3000 * 1.15
[DEBUG] add 输入: 3449.9999999999995 + 5000
[DEBUG] subtract 输入: 8450.0 - 2000
6450.0

七、 `message.py` 文件解析

库名及作用表格

库/模块名	作用
`os`	提供操作系统接口功能，用于读取环境变量（如 `DASHSCOPE_API_KEY`）。
`semantic_kernel` (缩写为 `sk`)	Semantic Kernel 核心库，用于构建和管理 AI 应用的工作流和插件。
`mathpluginmore`	用户自定义的数学插件模块，封装数学计算逻辑（如投资增长、金额扣除等操作）。
`semantic_kernel.connectors.ai.open_ai`	提供与 OpenAI 兼容的接口，用于连接阿里云 DashScope 的模型服务（如 `qwen-max`）。
`dotenv`	从 `.env` 文件中加载环境变量（如 API 密钥），简化配置管理。
`semantic_kernel.planning.sequential_planner`	提供顺序计划器（`SequentialPlanner`），将复杂问题拆解为分步执行计划。
`asyncio`	支持异步编程，用于管理异步任务（如计划生成、执行）。

方法名及作用表格

方法名	所属库/模块/类	作用
`os.getenv`	`os`	从环境变量中读取指定键的值（如 `DASHSCOPE_API_KEY`）。
`sk.Kernel()`	`semantic_kernel`	创建 Semantic Kernel 的核心实例，用于整合服务和插件。
`load_dotenv`	`dotenv`	加载 `.env` 文件中的环境变量到当前运行环境。
`add_chat_service`	`kernel` 对象（`semantic_kernel`）	为 Kernel 添加一个聊天服务（如阿里云 DashScope 的 `qwen-max` 模型）。
`import_skill`	`kernel` 对象（`semantic_kernel`）	将自定义插件（如 `MathPlugin`）导入为 Kernel 的技能，供后续调用。
`SequentialPlanner`	`semantic_kernel.planning.sequential_planner`	创建顺序计划器实例，用于将用户问题拆解为可执行的步骤。
`create_plan_async(ask)`	`SequentialPlanner` 实例	根据用户输入（`ask`）生成分步执行计划（`Plan` 对象）。
`invoke_async()`	`Plan` 实例	异步执行计划中的所有步骤，并返回最终结果。
`asyncio.run(main)`	`asyncio`	运行异步函数（如 `main()`），管理事件循环和任务调度。
`MathPlugin`	`mathpluginmore`	用户自定义的数学插件类，封装计算逻辑（如 `calculate_investment`、`subtract_cost` 等方法）。

`plan._steps` 的含义

plan._steps 是 Semantic Kernel 中 Plan 对象的一个内部属性（以下划线开头表示私有），用于存储计划的分步执行逻辑。
当用户提问时，SequentialPlanner 会将问题拆解为多个可执行的步骤，每个步骤对应一个已注册的技能函数（例如 MathPlugin 中的数学运算方法）。
_steps 的具体内容示例如下：

属性	作用
`step.description`	步骤的描述（例如“调用 MathPlugin 的 add 方法”）
`step._state`	步骤的输入参数和上下文信息（例如 `{"input": "2130.23", "number2": "5"}`）
`step.skill_name`	所属技能的名称（例如 `math_plugin`，对应 `import_skill` 时定义的名称）
`step.function`	调用的具体方法（例如 `MathPlugin.add`）

八、 `mathpluginmore.py` 文件解析

该文件定义了一个 MathPlugin 类，包含多个数学运算方法，并通过 Semantic Kernel 的装饰器声明为可调用的技能函数。以下是关键内容解析：

(1) 核心功能

方法名	数学运算	参数传递方式	示例调用
`square_root`	平方根计算	直接输入数值（`number` 参数）	`square_root("100")` → 返回 `"10.0"`
`add`	加法	通过上下文对象 `SKContext` 获取参数	`add(input="2", number2="3")` → `"5.0"`
`subtract`	减法	同上	`subtract(input="5", number2="3")` → `"2.0"`
`multiply`	乘法	同上	`multiply(input="2", number2="3")` → `"6.0"`
`divide`	除法	同上	`divide(input="6", number2="3")` → `"2.0"`

(2) 装饰器说明

装饰器	作用
`@sk_function`	将方法标记为 Semantic Kernel 可调用的技能函数，定义名称、描述和输入参数。
`@sk_function_context_parameter`	声明方法需要从上下文（`SKContext`）中获取的参数，包括参数名称和描述（用于 Planner 自动匹配用户输入）。

(3) 参数传递逻辑

直接参数传递（如 square_root）：
直接通过方法参数接收输入（例如 number: str）。
上下文参数传递（如 add）：
通过 SKContext 对象获取参数，需使用 @sk_function_context_parameter 声明参数名称和描述，例如：

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

九、代码运行流程示例

假设用户提问：“如果我有 100 元，先增加 20%，再减去 30 元，还剩多少钱？”
SequentialPlanner 可能生成如下 _steps：

Step 1: 调用 multiply(input="100", number2="1.2") → 得到 120.0
Step 2: 调用 subtract(input="120", number2="30") → 最终结果 90.0

补充说明

私有方法/属性：
代码中出现的 plan._steps 和 step._state 是类或对象的私有属性（以下划线开头），通常不推荐直接调用，此处用于调试目的。
自定义插件逻辑：
MathPlugin 是 calculate_investment_growth（计算投资增长）和 subtract_cost（扣除花费）的方法。
异步编程：
await 关键字和 async 函数用于非阻塞式异步操作（如调用外部 API 或执行复杂计算），提升性能。

十、总结

Semantic Kernel的Planner组件正在重塑AI应用开发范式。通过将自然语言转化为可执行计划，开发者可以专注业务逻辑创新，而无需深究底层实现细节。随着1.0正式版的发布，该组件已在实际生产环境经受住日均百万级调用的考验（微软技术博客数据）。对于渴望拥抱智能编排技术的开发者而言，现在正是深入探索Planner组件的最佳时机。

【Semantic Kernel核心组件】planner：大模型时代的智能任务编排引擎

一、组件定位与核心价值

二、核心工作原理

2.1 计划生成机制

2.2 关键技术特性

三、典型应用场景

3.1 复杂计算任务

3.2 智能服务编排

3.3 企业级工作流

四、开发最佳实践

4.1 插件设计规范

4.2 提示工程优化

4.3 性能调优技巧

五、演进方向与挑战

六、实例代码

运行流程图

manthpluginmore.py

message.py

七、 `message.py` 文件解析

库名及作用表格

方法名及作用表格

`plan._steps` 的含义

八、 `mathpluginmore.py` 文件解析

(1) 核心功能

(2) 装饰器说明

(3) 参数传递逻辑

九、代码运行流程示例

补充说明

十、总结

网站公告

今日签到

热门文章

最新发布

【Semantic Kernel核心组件】planner：大模型时代的智能任务编排引擎

一、组件定位与核心价值

二、核心工作原理

2.1 计划生成机制

2.2 关键技术特性

三、典型应用场景

3.1 复杂计算任务

3.2 智能服务编排

3.3 企业级工作流

四、开发最佳实践

4.1 插件设计规范

4.2 提示工程优化

4.3 性能调优技巧

五、演进方向与挑战

六、实例代码

运行流程图

manthpluginmore.py

message.py

七、 message.py 文件解析

库名及作用表格

方法名及作用表格

plan._steps 的含义

八、 mathpluginmore.py 文件解析

(1) 核心功能

(2) 装饰器说明

(3) 参数传递逻辑

九、 代码运行流程示例

补充说明

十、总结

网站公告

今日签到

热门文章

最新发布

七、 `message.py` 文件解析

`plan._steps` 的含义

八、 `mathpluginmore.py` 文件解析

九、代码运行流程示例