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

三、标准项目结构

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 代码内联对比

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

参考资源

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

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