Semantic Kernel Python 进阶：Prompt 模板函数嵌套调用实战

Semantic Kernel Python 进阶：Prompt 模板函数嵌套调用实战

前言

Microsoft 的 Semantic Kernel (SK) 提供了一个强大的特性：允许在 Prompt 模板中直接调用其他函数。这意味着你可以在一个 Semantic Function 的 Prompt 中嵌套调用其他 Semantic Functions 或 Native Functions，实现真正的函数式编程范式。

本文将深入讲解 SK Python 中的嵌套调用机制，并通过大量实战示例展示如何构建模块化的 AI 应用。

---

一、核心概念：Prompt 模板语法

Semantic Kernel 使用双大括号 {{...}} 作为模板语法，支持两种主要操作：

语法	用途	示例
{{$variable}}	变量插值	{{$user_name}}
{{Plugin.Function}}	函数调用	{{TextUtils.Summarize $input}}

关键特性：

• 支持在 Prompt 中直接调用已注册的插件函数
• 支持链式调用（一个函数的输出作为另一个函数的输入）
• 支持混合调用 Semantic Functions 和 Native Functions

---

二、环境准备

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 Function 调用

4.1 定义 Native Function（Python 代码）

# 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")

4.2 创建 Semantic Function（文件分离方式）

文件 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
    }
  ]
}

4.3 加载并执行

# 从文件夹加载 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())

执行流程解析：

1. Kernel 解析 skprompt.txt，发现 {{TextUtils.count_words ...}} 和 {{TextUtils.extract_keywords ...}}
2. 先执行这两个 Native Function，获取字数和关键词
3. 将函数返回值填充到 Prompt 中对应位置
4. 最后将完整的 Prompt 发送给 LLM 生成最终回复

---

五、进阶：Semantic Function 嵌套调用

更有趣的是，你可以在 Prompt 中调用其他的 Semantic Function，构建复杂的处理管道。

5.1 创建可复用的 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
    }
  ]
}

5.2 嵌套调用：在 Prompt 中调用其他 Semantic Function

文件 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": "英文"
    }
  ]
}

5.3 加载并执行嵌套调用

# 加载两个插件
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 管道

以下是一个完整的 RAG（检索增强生成）示例，展示如何在 Prompt 中组合多个函数。

6.1 定义检索插件（Native Function）

# 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")

6.2 定义重排序插件（Native Function）

# 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")

6.3 组合成完整的 RAG Semantic Function

文件 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
    }
  ]
}

6.4 加载并执行 RAG

# 加载所有插件
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())

---

七、高级技巧与最佳实践

7.1 处理函数执行顺序

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}}
请生成完整的执行摘要。

7.2 错误处理与降级策略

# 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")

7.3 使用 Handlebars 模板实现复杂逻辑

对于需要循环、条件判断的复杂场景，可以使用 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]

---

八、调试与监控

8.1 查看渲染后的 Prompt

在开发阶段，你可能需要查看最终发送给 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())

8.2 函数调用追踪

# 监听函数调用事件
@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))}")

---

九、性能优化建议

1. 缓存常用结果：对于重复的函数调用，使用 SK 的 KernelMemory 或外部缓存
2. 并行执行：无依赖关系的函数会自动并行执行，合理利用这一点
3. 控制嵌套深度：建议不超过 3-5 层嵌套，避免延迟过高
4. 异步优化：确保所有 Native Function 都使用 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 和配置
• ✅ 构建复杂的 LLM 处理管道
• ✅ 实现 Semantic Function 与 Native Function 的无缝集成
• ✅ 创建可维护、可测试的模块化 AI 系统
• ✅ 实现 RAG、Agent 等高级应用场景

关键要点回顾：

• 使用 {{Plugin.Function}} 语法在 Prompt 中调用函数
• 函数参数通过 param=value 形式传递，支持变量 $var 和字面量
• SK 自动处理依赖关系和执行顺序
• 合理使用模板格式（semantic-kernel vs handlebars）应对不同复杂度

---

附录：文件分离 vs 代码内联对比

方式	适用场景	优点	缺点
文件分离 (skprompt.txt + config.json)	生产环境、复杂插件	• 清晰分离内容和配置 • 支持热更新 • 便于版本管理 • 非开发人员可编辑 Prompt	• 文件较多 • 需要管理目录结构
代码内联 (PromptTemplateConfig)	快速原型、简单测试	• 单文件完成 • 便于调试 • 动态生成 Prompt	• 代码和配置耦合 • 不易维护

生产环境推荐： 始终使用文件分离方式，将插件组织在独立的目录结构中。

---

参考资源

• Semantic Kernel Python 官方文档
• SK Prompt Template 语法详解
• GitHub: semantic-kernel/python

本文基于 Semantic Kernel Python v1.x 版本编写，不同版本 API 可能略有差异。