Last updated
from semantic_kernel.functions import kernel_function
class InventoryPlugin:
@kernel_function(
description="根据产品ID查询当前库存数量",
name="GetStockCount"
)
def get_stock_count(self, product_id: str) -> int:
# 这里调用旧系统的 API 或数据库
print(f"Checking database for {product_id}...")
return 42
@kernel_function(
description="当库存低于阈值时发送补货邮件",
name="SendRestockEmail"
)
def send_restock_email(self, product_id: str, quantity: int) -> str:
return f"Email sent for {quantity} units of {product_id}"# 定义一个写邮件的 Prompt 模板
email_prompt = """
请为产品 {{$product_name}} 写一封补货申请邮件。
当前库存: {{$current_stock}}
建议补货: {{$amount}}
语气: 正式且紧急
"""# 伪代码：初始化 Kernel 并注册模型服务
async def main():
# 1. 初始化 Kernel
kernel = Kernel()
# 2. 配置 AI 服务
service_id = "default"
kernel.add_model_service(ModelService(model="<MODEL>", api_key="<KEY>"))
# 3. 注册插件
# 导入原生插件
kernel.import_plugin_from_object(InventoryPlugin(), "Inventory")
# 导入语义插件 (Prompt)
kernel.create_function_from_prompt(
function_name="WriteEmail",
plugin_name="Office",
prompt=email_prompt
)
# 4. 自动规划 (Planner)
planner = Planner(model_service="default")
# 用户的模糊指令
goal = "查询某产品的库存，如果低于阈值，就生成一封补货申请邮件。"
# Planner 会自动生成计划并执行
# Step 1: Call Inventory.GetStockCount("<product>") -> Returns <count>
# Step 2: Logic Check (42 < 50)
# Step 3: Call Office.WriteEmail(name="<product>", stock=<count>, amount=<amount>)
result = await planner.invoke(kernel, goal)
print(result)
if __name__ == "__main__":
import asyncio
asyncio.run(main())