Semantic Kernel Python 进阶:Prompt 模板函数嵌套调用实战
前言
Microsoft 的 Semantic Kernel (SK) 提供了一个强大的特性:允许在 Prompt 模板中直接调用其他函数。这意味着你可以在一个 Semantic Function 的 Prompt 中嵌套调用其他 Semantic Functions 或 Native Functions,实现真正的函数式编程范式。
介绍 Microsoft Semantic Kernel 在 Python 中的高级用法,重点讲解 Prompt 模板内的函数嵌套调用机制。内容包括核心语法(变量插值与函数调用)、环境配置、项目结构规范。通过原生函数(Native Function)与语义函数(Semantic Function)的混合调用示例,展示了如何构建模块化 AI 应用。此外,还涵盖了 RAG 管道构建、Handlebars 模板逻辑、错误处理策略及性能优化建议。文章强调文件分离方式便于维护,并提供调试监控方法,帮助开发者实现复杂的 LLM 处理流程。
Microsoft 的 Semantic Kernel (SK) 提供了一个强大的特性:允许在 Prompt 模板中直接调用其他函数。这意味着你可以在一个 Semantic Function 的 Prompt 中嵌套调用其他 Semantic Functions 或 Native Functions,实现真正的函数式编程范式。
本文将深入讲解 SK Python 中的嵌套调用机制,并通过大量实战示例展示如何构建模块化的 AI 应用。
Semantic Kernel 使用双大括号 {{...}} 作为模板语法,支持两种主要操作:
| 语法 | 用途 | 示例 |
|---|---|---|
{{$variable}} | 变量插值 | {{$user_name}} |
{{Plugin.Function}} | 函数调用 | {{TextUtils.Summarize $input}} |
关键特性:
import asyncio
import semantic_kernel as sk
from semantic_kernel.kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion, OpenAIChatCompletion
# 初始化 Kernel
kernel = Kernel()
# 添加 AI 服务(以 Azure OpenAI 为例)
kernel.add_service(
AzureChatCompletion(
service_id="default",
deployment_name="gpt-4",
endpoint="https://your-resource.openai.azure.com/",
api_key="your-api-key",
api_version="2024-02-15-preview"
)
)
# 或者使用 OpenAI
# kernel.add_service(
# OpenAIChatCompletion(
# service_id="default",
# ai_model_id="gpt-4",
# api_key="your-openai-key"
# )
# )
Semantic Kernel 将 Semantic Function 分离到两个文件中:
my_plugins/
├── TextProcessing/ # 文本处理插件
│ ├── summarize/
│ │ ├── skprompt.txt # Prompt 模板
│ │ └── config.json # 配置参数(temperature、输入变量等)
│ └── analyze_sentiment/
│ ├── skprompt.txt
│ └── config.json
├── Translation/ # 翻译插件
│ └── translate_summary/
│ ├── skprompt.txt # 嵌套调用 summarize 的示例
│ └── config.json
└── RAG/ # RAG 系统插件
├── retrieve/
├── rerank/
└── generate_answer/
# native_plugins/TextUtilsPlugin.py
from semantic_kernel.functions import kernel_function
class TextUtilsPlugin:
"""文本处理工具插件"""
@kernel_function(description="统计文本字数")
def count_words(self, text: str) -> str:
"""统计文本中的字数"""
words = len(text.split())
return f"{words} 字"
@kernel_function(description="提取关键词")
def extract_keywords(self, text: str, top_n: int = 3) -> str:
"""简单提取关键词(实际应用中可使用 NLP 库)"""
words = text.split()
unique_words = list(dict.fromkeys(words))[:top_n]
return ", ".join(unique_words)
# 注册插件到 Kernel
kernel.import_plugin_from_object(TextUtilsPlugin(), plugin_name="TextUtils")
文件 1:my_plugins/Analysis/feedback_analyzer/skprompt.txt
请分析以下用户反馈:
反馈内容:{{$feedback}}
--- 自动分析结果:
- 字数统计:{{TextUtils.count_words text=$feedback}}
- 关键词:{{TextUtils.extract_keywords text=$feedback top_n=5}}
基于以上信息,请生成一份简要的分析报告。
文件 2:my_plugins/Analysis/feedback_analyzer/config.json
{
"schema": 1,
"type": "completion",
"description": "分析用户反馈并生成报告",
"execution_settings": {
"default": {
"max_tokens": 500,
"temperature": 0.3,
"top_p": 0.5
}
},
"input_variables": [
{
"name": "feedback",
"description": "用户反馈内容",
"is_required": true
}
]
}
# 从文件夹加载 Semantic Function(自动读取 skprompt.txt 和 config.json)
analysis_plugin = kernel.import_plugin_from_prompt_directory(
parent_directory="./my_plugins",
plugin_directory="Analysis"
)
# 执行
async def main():
result = await kernel.invoke(
analysis_plugin["feedback_analyzer"],
sk.KernelArguments(feedback="这个产品质量很好,但是物流速度太慢了,希望能改进配送服务。")
)
print(result)
asyncio.run(main())
执行流程解析:
skprompt.txt,发现 {{TextUtils.count_words ...}} 和 {{TextUtils.extract_keywords ...}}更有趣的是,你可以在 Prompt 中调用其他的 Semantic Function,构建复杂的处理管道。
文件 1:my_plugins/TextProcessing/summarize/skprompt.txt
请将以下文本总结为 3 个要点,每个要点不超过 20 字:
{{$input}}
输出格式:
1. [要点 1]
2. [要点 2]
3. [要点 3]
文件 2:my_plugins/TextProcessing/summarize/config.json
{
"schema": 1,
"type": "completion",
"description": "将长文本总结为 3 个要点",
"execution_settings": {
"default": {
"max_tokens": 200,
"temperature": 0.3,
"top_p": 0.5
}
},
"input_variables": [
{
"name": "input",
"description": "需要总结的原始文本",
"is_required": true
}
]
}
文件 3:my_plugins/Translation/translate_summary/skprompt.txt
请将以下内容翻译成{{$target_lang}}:
{{TextProcessing.summarize input=$long_text}}
注意:保持 3 个要点的格式,使用专业术语。
文件 4:my_plugins/Translation/translate_summary/config.json
{
"schema": 1,
"type": "completion",
"description": "先总结后翻译的管道函数",
"execution_settings": {
"default": {
"max_tokens": 300,
"temperature": 0.3
}
},
"input_variables": [
{
"name": "long_text",
"description": "需要处理的长文本",
"is_required": true
},
{
"name": "target_lang",
"description": "目标语言(如:英文、日文、法文)",
"is_required": true,
"default": "英文"
}
]
}
# 加载两个插件
text_processing = kernel.import_plugin_from_prompt_directory(
parent_directory="./my_plugins",
plugin_directory="TextProcessing"
)
translation = kernel.import_plugin_from_prompt_directory(
parent_directory="./my_plugins",
plugin_directory="Translation"
)
# 执行:translate_summary 会自动调用 summarize
async def nested_demo():
long_article = """
人工智能技术在近年来取得了突破性进展。深度学习模型在自然语言处理、
计算机视觉和语音识别等领域展现出强大能力。各大科技公司纷纷投入巨资
研发大语言模型,如 GPT-4、Claude 等。这些模型不仅能理解和生成自然语言,
还能进行复杂的推理和代码生成。然而,AI 发展也面临伦理、安全和就业等
挑战,需要社会各界共同探讨治理框架。
"""
result = await kernel.invoke(
translation["translate_summary"],
sk.KernelArguments(
long_text=long_article,
target_lang="英文"
)
)
print("=== 翻译后的摘要 ===")
print(result)
asyncio.run(nested_demo())
输出效果:
=== 翻译后的摘要 ===
1. AI technology has made breakthrough progress recently
2. Deep learning shows strong capabilities in NLP and computer vision
3. Major tech companies invest heavily in large language models
以下是一个完整的 RAG(检索增强生成)示例,展示如何在 Prompt 中组合多个函数。
# native_plugins/SearchPlugin.py
from semantic_kernel.functions import kernel_function
class SearchPlugin:
"""模拟文档检索插件(实际可连接向量数据库)"""
@kernel_function(description="搜索相关文档")
def search_documents(self, query: str, top_k: int = 3) -> str:
"""模拟检索结果"""
docs = {
"python": "Python 是一种高级编程语言,以简洁和易读性著称。",
"semantic kernel": "Semantic Kernel 是微软开源的 LLM 应用开发框架。",
"RAG": "RAG(检索增强生成)结合检索系统和生成模型,提高回答准确性。"
}
results = []
for key, value in docs.items():
if any(q in key or q in value for q in query.lower().split()):
results.append(f"[文档] {key}: {value}")
return "\n".join(results[:top_k]) if results else "未找到相关文档"
kernel.import_plugin_from_object(SearchPlugin(), plugin_name="Search")
# native_plugins/RerankPlugin.py
from semantic_kernel.functions import kernel_function
class RerankPlugin:
"""文档重排序插件"""
@kernel_function(description="对检索结果进行重排序")
def rerank(self, documents: str, query: str) -> str:
"""简单模拟重排序(实际可使用 Cross-Encoder 等模型)"""
docs = documents.split("\n")
return "\n".join(docs[:2]) # 简化处理:只返回前 2 个
kernel.import_plugin_from_object(RerankPlugin(), plugin_name="Rerank")
文件 1:my_plugins/RAG/rag_qa/skprompt.txt
你是一个专业的 AI 助手。请基于以下检索到的信息回答用户问题。
用户问题:{{$question}}
步骤 1 - 文档检索:{{Search.search_documents query=$question top_k=5}}
步骤 2 - 结果重排序(取最相关的 2 个):{{Rerank.rerank documents=$search_results query=$question}}
--- 请基于以上重排序后的文档,回答用户问题。如果文档中没有相关信息,请明确告知。
答案:
文件 2:my_plugins/RAG/rag_qa/config.json
{
"schema": 1,
"type": "completion",
"description": "RAG 问答系统",
"execution_settings": {
"default": {
"max_tokens": 500,
"temperature": 0.1
}
},
"input_variables": [
{
"name": "question",
"description": "用户问题",
"is_required": true
}
]
}
# 加载所有插件
search_plugin = kernel.import_plugin_from_object(SearchPlugin(), plugin_name="Search")
rerank_plugin = kernel.import_plugin_from_object(RerankPlugin(), plugin_name="Rerank")
rag_plugin = kernel.import_plugin_from_prompt_directory(
parent_directory="./my_plugins",
plugin_directory="RAG"
)
# 执行 RAG
async def run_rag():
result = await kernel.invoke(
rag_plugin["rag_qa"],
sk.KernelArguments(question="什么是 Semantic Kernel?")
)
print(result)
asyncio.run(run_rag())
SK 会自动解析依赖关系并确定执行顺序。以下是一个复杂的多层嵌套示例:
文件:my_plugins/DataPipeline/process_report/skprompt.txt
数据分析报告生成流程:
1. 原始数据:{{$raw_data}}
2. 数据清洗:{{DataPipeline.clean data=$raw_data}}
3. 特征提取:{{DataPipeline.extract_features cleaned_data=$cleaned_data}}
4. 异常检测:{{MLModel.detect_anomalies features=$features}}
5. 可视化描述:{{ChartGenerator.describe chart_type="line" data=$anomalies}}
6. 最终报告:{{ReportFormatter.format sections=$sections}}
请生成完整的执行摘要。
# native_plugins/SafePlugin.py
from semantic_kernel.functions import kernel_function
class SafePlugin:
"""带错误处理的插件"""
@kernel_function(description="安全调用外部 API")
def safe_search(self, query: str) -> str:
try:
# 实际 API 调用
return external_api_call(query)
except Exception as e:
# 返回友好错误信息,避免整个 Pipeline 失败
return f"[搜索服务暂时不可用,使用本地知识回答] 相关主题:{query}"
kernel.import_plugin_from_object(SafePlugin(), plugin_name="Safe")
对于需要循环、条件判断的复杂场景,可以使用 Handlebars 模板格式:
文件:my_plugins/Advanced/conditional_process/skprompt.txt
{{#if (TextUtils.is_long text=$input)}}
检测到长文本,执行总结流程:
{{TextProcessing.summarize input=$input}}
{{else}}
短文本直接分析情感:
{{TextProcessing.analyze_sentiment text=$input}}
{{/if}}
文件:my_plugins/Advanced/conditional_process/config.json
{
"schema": 1,
"type": "completion",
"description": "根据文本长度选择不同处理流程",
"template_format": "handlebars",
"execution_settings": {
"default": {
"max_tokens": 500,
"temperature": 0.3
}
},
"input_variables": [
{
"name": "input",
"description": "输入文本",
"is_required": true
}
]
}
注意: 使用 Handlebars 需要安装扩展:
pip install semantic-kernel[handlebars]
在开发阶段,你可能需要查看最终发送给 LLM 的完整 Prompt:
from semantic_kernel.filters import PromptRenderFilter
class DebugFilter(PromptRenderFilter):
async def on_prompt_render(self, context, next):
print(f"=== 渲染 Prompt: {context.function.name} ===")
await next(context)
print(f"最终 Prompt:\n{context.rendered_prompt}")
print("=" * 50)
kernel.add_filter(DebugFilter())
# 监听函数调用事件
@kernel.on_function_invoking
def on_invoking(sender, args):
print(f"🔄 正在调用:{args.function.name}")
@kernel.on_function_invoked
def on_invoked(sender, args):
print(f"✅ 调用完成:{args.function.name}, 结果长度:{len(str(args.result))}")
KernelMemory 或外部缓存async/await# 优化示例:带缓存的 Native Function
from functools import lru_cache
class OptimizedPlugin:
@kernel_function
@lru_cache(maxsize=100)
def expensive_operation(self, param: str) -> str:
# 耗时操作
pass
Semantic Kernel 的 Prompt 模板函数嵌套调用机制为构建模块化、可复用的 AI 应用提供了强大支持。通过本文介绍的技术,你可以:
skprompt.txt + config.json)分离 Prompt 和配置关键要点回顾:
{{Plugin.Function}} 语法在 Prompt 中调用函数param=value 形式传递,支持变量 $var 和字面量semantic-kernel vs handlebars)应对不同复杂度| 方式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 文件分离 (skprompt.txt + config.json) | 生产环境、复杂插件 | • 清晰分离内容和配置 • 支持热更新 • 便于版本管理 • 非开发人员可编辑 Prompt | • 文件较多 • 需要管理目录结构 |
| 代码内联 (PromptTemplateConfig) | 快速原型、简单测试 | • 单文件完成 • 便于调试 • 动态生成 Prompt | • 代码和配置耦合 • 不易维护 |
生产环境推荐: 始终使用文件分离方式,将插件组织在独立的目录结构中。
本文基于 Semantic Kernel Python v1.x 版本编写,不同版本 API 可能略有差异。

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online