1. Semantic Kernel Python 进阶:Prompt 模板中的函数嵌套调用实战
作为一名长期从事AI应用开发的工程师,我一直在寻找能够提升大语言模型(LLM)开发效率的框架。Microsoft的Semantic Kernel(SK)无疑是我近年来发现的最具实用价值的工具之一。特别是在其Python版本中,Prompt模板的函数嵌套调用功能彻底改变了我的开发方式,让AI应用的模块化和复用性达到了全新高度。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 环境准备与基础配置
2.1 安装与初始化
首先确保你已经安装了最新版的Semantic Kernel Python包:
bash复制pip install semantic-kernel
对于需要Handlebars模板支持的情况,还需安装扩展:
bash复制pip install semantic-kernel[handlebars]
初始化Kernel是使用SK的第一步,这里我推荐使用Azure OpenAI服务,因为它提供了更稳定的企业级支持:
python复制import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
kernel = sk.Kernel()
# 配置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"
)
)
注意:在实际项目中,建议将API密钥等敏感信息存储在环境变量中,而不是直接写在代码里。
2.2 项目结构设计
良好的项目结构是可持续开发的基础。经过多个项目的实践,我总结出以下目录结构最为高效:
code复制sk_plugins/
├── CoreSkills/ # 核心技能插件
│ ├── Summarization/
│ │ ├── skprompt.txt
│ │ └── config.json
│ └── Translation/
│ ├── skprompt.txt
│ └── config.json
├── BusinessLogic/ # 业务逻辑插件
│ ├── CustomerSupport/
│ └── DataAnalysis/
└── Utilities/ # 工具类插件
├── TextProcessing/
└── DataValidation/
这种结构清晰地区分了不同层级的插件,便于团队协作和维护。
3. Prompt模板基础与函数调用
3.1 模板语法详解
SK的Prompt模板使用双大括号{{...}}作为插值语法,支持两种主要操作:
-
变量插值:
{{$variableName}}- 用于注入上下文中的变量值
- 示例:
你好,{{$userName}}!
-
函数调用:
{{PluginName.FunctionName arg1=value1 arg2=value2}}- 可以调用已注册的任何插件函数
- 支持链式调用(一个函数的输出作为另一个函数的输入)
- 示例:
{{TextUtils.CountWords text=$input}}
3.2 Native Function开发实践
Native Function是用Python代码实现的插件函数。下面是一个实用的文本处理插件示例:
python复制# plugins/text_utils.py
from semantic_kernel.functions import kernel_function
class TextUtilsPlugin:
"""文本处理工具集"""
@kernel_function(
name="count_words",
description="统计文本中的单词数量"
)
def count_words(self, text: str) -> int:
"""实际统计逻辑"""
return len(text.split())
@kernel_function(
name="extract_entities",
description="从文本中提取命名实体"
)
def extract_entities(self, text: str) -> str:
"""简单的实体提取实现"""
# 实际项目中可以使用spaCy等NLP库
entities = []
for word in text.split():
if word.istitle():
entities.append(word)
return ", ".join(entities) if entities else "未识别到实体"
注册插件到Kernel:
python复制kernel.import_plugin_from_object(TextUtilsPlugin(), "TextUtils")
4. Semantic Function开发进阶
4.1 文件分离式开发
SK推荐将Semantic Function分离到两个文件中:
skprompt.txt- 包含Prompt模板config.json- 包含配置参数
示例:文本总结功能
plugins/CoreSkills/Summarization/skprompt.txt:
code复制请将以下文本总结为3个要点,每个要点不超过20个单词:
{{$input}}
要求:
1. 保留核心信息
2. 使用简洁的语言
3. 输出格式:
- 要点1: ...
- 要点2: ...
- 要点3: ...
plugins/CoreSkills/Summarization/config.json:
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
}
]
}
加载插件:
python复制summarize_plugin = kernel.import_plugin_from_prompt_directory(
parent_directory="plugins",
plugin_directory="CoreSkills/Summarization"
)
4.2 函数嵌套调用实战
SK最强大的功能之一是在Prompt中嵌套调用其他函数。下面是一个多语言摘要生成器的实现:
plugins/CoreSkills/MultiLangSummary/skprompt.txt:
code复制请生成{{$lang}}语的文本摘要:
原始文本:{{$input}}
处理步骤:
1. 首先提取关键信息:{{CoreSkills.Summarize input=$input}}
2. 然后翻译为目标语言:{{CoreSkills.Translate text=$summary target_lang=$lang}}
最终摘要要求:
- 保持原意的准确性
- 符合目标语言的习惯表达
- 长度控制在100字以内
对应的config.json中需要定义两个输入变量:input和lang。
执行示例:
python复制async def generate_summary():
result = await kernel.invoke(
kernel.plugins["CoreSkills"]["MultiLangSummary"],
sk.KernelArguments(
input="""人工智能(AI)正在改变各行各业...""",
lang="法语"
)
)
print(result)
asyncio.run(generate_summary())
5. 复杂场景应用:构建RAG系统
5.1 检索插件实现
RAG(检索增强生成)系统通常需要三个核心组件:
python复制# plugins/retrieval.py
from semantic_kernel.functions import kernel_function
import your_vector_db_client # 替换为实际的向量数据库客户端
class RetrievalPlugin:
"""文档检索组件"""
@kernel_function(
name="vector_search",
description="从向量数据库检索相关文档"
)
async def vector_search(self, query: str, top_k: int = 3) -> str:
"""实际向量搜索实现"""
results = await your_vector_db_client.search(
query=query,
top_k=top_k
)
return "\n".join([f"[{i+1}] {doc.content}" for i, doc in enumerate(results)])
5.2 重排序插件
python复制# plugins/reranker.py
from semantic_kernel.functions import kernel_function
import your_reranking_model # 替换为实际的重排序模型
class RerankerPlugin:
"""结果重排序组件"""
@kernel_function(
name="rerank",
description="对检索结果进行相关性重排序"
)
async def rerank(self, documents: str, query: str) -> str:
"""实际重排序逻辑"""
docs_list = documents.split("\n")
reranked = await your_reranking_model.rerank(
query=query,
documents=docs_list
)
return "\n".join(reranked[:3]) # 返回前3个最相关结果
5.3 完整的RAG Prompt
plugins/RAG/system/skprompt.txt:
code复制你是一个专业的知识助手。请基于以下检索到的信息回答问题。
用户问题:{{$question}}
检索到的文档(已按相关性排序):
{{Retrieval.vector_search query=$question top_k=5}}
重排序后的最相关文档:
{{Reranker.rerank documents=$retrieved_docs query=$question}}
请根据以上信息回答问题:
1. 如果文档中有明确答案,直接回答并引用来源
2. 如果信息不完整,进行合理推断并说明
3. 如果完全没有相关信息,诚实地告知
5.4 RAG系统执行
python复制# 注册所有插件
kernel.import_plugin_from_object(RetrievalPlugin(), "Retrieval")
kernel.import_plugin_from_object(RerankerPlugin(), "Reranker")
rag_plugin = kernel.import_plugin_from_prompt_directory(
parent_directory="plugins",
plugin_directory="RAG/system"
)
# 执行RAG查询
async def rag_query(question: str):
result = await kernel.invoke(
rag_plugin["rag"],
sk.KernelArguments(question=question)
)
return str(result)
# 示例查询
question = "Semantic Kernel中如何处理函数嵌套调用?"
answer = asyncio.run(rag_query(question))
print(answer)
6. 高级技巧与最佳实践
6.1 错误处理机制
在实际应用中,健壮的错误处理至关重要:
python复制# plugins/safe_utils.py
from semantic_kernel.functions import kernel_function
class SafePlugin:
"""安全执行工具"""
@kernel_function(
name="try_invoke",
description="安全地调用函数并提供降级方案"
)
async def try_invoke(
self,
plugin_name: str,
function_name: str,
**kwargs
) -> str:
"""带错误处理的函数调用"""
try:
func = kernel.plugins[plugin_name][function_name]
result = await kernel.invoke(func, sk.KernelArguments(**kwargs))
return str(result)
except Exception as e:
print(f"调用失败: {plugin_name}.{function_name}, 错误: {str(e)}")
return f"[{function_name} 功能暂时不可用]"
6.2 性能优化技巧
-
并行执行:SK会自动并行执行无依赖关系的函数调用
-
结果缓存:对计算密集型操作实现缓存
python复制from functools import lru_cache
class OptimizedPlugin:
"""优化插件示例"""
@kernel_function
@lru_cache(maxsize=100)
def expensive_operation(self, param: str) -> str:
"""带缓存的耗时操作"""
# 模拟耗时处理
time.sleep(1)
return param.upper()
- 批量处理:设计支持批量处理的函数减少API调用
6.3 调试与监控
添加调试过滤器查看实际Prompt:
python复制from semantic_kernel.filters import PromptRenderFilter
class DebugFilter(PromptRenderFilter):
async def on_prompt_render(self, context, next):
print(f"=== 渲染 {context.function.name} ===")
await next(context)
print(f"最终Prompt:\n{context.rendered_prompt[:500]}...") # 限制输出长度
print("=" * 50)
kernel.add_filter(DebugFilter())
函数调用追踪:
python复制@kernel.on_function_invoking
def log_invoking(sender, args):
print(f"调用开始: {args.function.name}")
@kernel.on_function_invoked
def log_invoked(sender, args):
print(f"调用完成: {args.function.name}, 耗时: {args.metadata['time']}ms")
7. 项目实战:客户支持自动化系统
7.1 系统架构设计
一个完整的客户支持系统可能包含以下组件:
- 意图识别:分类用户问题的类型
- 知识检索:从文档库查找相关信息
- 情感分析:检测用户情绪状态
- 回答生成:生成友好且专业的回复
- 转接判断:决定是否需要人工介入
7.2 核心实现代码
plugins/SupportSystem/main.skprompt.txt:
code复制{{$user_input}}
处理流程:
1. 识别意图:{{Classification.DetectIntent input=$user_input}}
2. 分析情绪:{{Sentiment.Analyze text=$user_input}}
3. 检索知识:{{KnowledgeBase.Search query=$intent}}
4. 生成回答:{{Response.Generate
intent=$intent
knowledge=$knowledge
sentiment=$sentiment}}
5. 判断转接:{{Routing.Decide
intent=$intent
sentiment=$sentiment_score}}
最终回复:
{{if $needs_human}}
[转接说明] {{$response}}
{{else}}
[自动回复] {{$response}}
{{/if}}
7.3 部署建议
对于生产环境部署,我推荐:
-
容器化:使用Docker打包整个应用
dockerfile复制FROM python:3.10-slim WORKDIR /app COPY . . RUN pip install semantic-kernel[all] CMD ["python", "support_bot.py"] -
API服务化:使用FastAPI暴露为REST接口
python复制from fastapi import FastAPI import uvicorn app = FastAPI() @app.post("/support") async def handle_query(query: str): result = await kernel.invoke( support_plugin["main"], sk.KernelArguments(user_input=query) ) return {"response": str(result)} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000) -
监控集成:添加Prometheus指标和日志记录
8. 经验总结与避坑指南
经过多个项目的实践,我总结了以下关键经验:
-
模板设计原则:
- 保持Prompt简洁明确
- 为每个函数调用添加清晰的注释
- 限制嵌套深度(建议不超过3层)
-
常见问题解决:
-
问题:函数调用循环依赖
解决:使用SK的内置检测机制,合理设计调用链 -
问题:Prompt渲染性能瓶颈
解决:对复杂模板预编译,使用Handlebars格式 -
问题:大模型响应不一致
解决:设置适当的temperature(0.1-0.3)和top_p值
-
-
性能优化指标:
- 单个函数调用延迟应<500ms
- 复杂管道的总延迟控制在3s内
- 错误率<0.1%
-
团队协作建议:
- 使用Git管理Prompt模板
- 建立插件开发规范
- 定期进行Prompt评审
在实际项目中,我发现最有效的开发流程是:
- 先在Jupyter Notebook中快速原型设计
- 然后重构为模块化插件
- 最后进行集成测试和优化
这种渐进式的方法既能保证开发速度,又能确保最终代码质量。
