版本说明:本文基于 Semantic Kernel Python SDK v1.0+(beta 0.9.6b1 及更高版本)编写。若您使用 v0.x 版本,请将 @kernel_function 替换为 @sk_functionKernelArguments 替换为 SKContext

一、什么是 Native Function?

Native Function(原生函数)是 Semantic Kernel 中允许你将现有的 Python 代码封装为可被 AI 调用的函数单元。与 Semantic Function(基于 Prompt 的函数)不同,Native Function 直接执行 Python 代码逻辑,适用于:

  • 数学计算
  • 数据库查询
  • API 调用(REST、数据库等)
  • 文件操作
  • 任何需要精确控制的业务逻辑

二、单参数场景:极简主义

核心原则

当函数只需要一个输入时,不需要使用 KernelArguments,也不需要额外的注解——默认参数名就是 input

代码示例

from semantic_kernel.functions import kernel_function

class TextPlugin:
    """文本处理插件"""
    
    @kernel_function(
        name="uppercase",
        description="Convert a string to uppercase"
    )
    def uppercase(self, input: str) -> str:
        """最简单的形式:一个参数,自动映射为 input"""
        return input.upper()
    
    @kernel_function(
        name="reverse",
        description="Reverse the input string"
    )
    def reverse(self, text: str) -> str:
        """参数名可以自定义,但语义上 input 最清晰"""
        return text[::-1]

    @kernel_function(
        name="word_count",
        description="Count words in text"
    )
    def word_count(self, content: str) -> int:
        """单参数支持任意类型返回值"""
        return len(content.split())

调用方式

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.functions import KernelArguments

async def main():
    # 1. 初始化 Kernel
    kernel = Kernel()
    
    # 2. 添加插件
    text_plugin = kernel.add_plugin(TextPlugin(), plugin_name="TextPlugin")
    
    # 3. 单参数调用 - 直接传递字符串(旧版方式,仍兼容)
    result = await kernel.invoke(
        function=text_plugin["uppercase"],
        arguments=KernelArguments(input="hello world")
    )
    print(f"Uppercase: {result}")  # HELLO WORLD
    
    # 4. 更简洁的调用方式(推荐)
    result = await kernel.invoke(
        function=text_plugin["reverse"],
        arguments=KernelArguments(text="hello")
    )
    print(f"Reverse: {result}")  # olleh

if __name__ == "__main__":
    asyncio.run(main())

关键点解析

特性 说明
参数名 可以是 inputtext 或任意名称,建议保持语义清晰
无需 Annotated 单参数时不需要 Annotated 类型注解,但多参数时需要
自动映射 KernelArguments 中的键自动映射到方法参数名
返回值 支持 strintfloatbool 等基础类型

三、多参数场景:使用 Annotated 和 KernelArguments

何时需要切换模式?

当函数需要两个或更多输入参数时,必须使用 Annotated 类型注解为每个参数添加描述,并通过 KernelArguments 传递参数。

代码示例

from typing import Annotated
from semantic_kernel.functions import kernel_function

class MathPlugin:
    """数学计算插件"""
    
    @kernel_function(
        name="add",
        description="Add two numbers together"
    )
    def add(
        self,
        number1: Annotated[float, "The first number to add"],
        number2: Annotated[float, "The second number to add"]
    ) -> Annotated[float, "The sum of the two numbers"]:
        """多参数必须使用 Annotated 添加描述"""
        return float(number1) + float(number2)
    
    @kernel_function(
        name="calculate_bmi",
        description="Calculate BMI index"
    )
    def calculate_bmi(
        self,
        weight: Annotated[float, "Weight in kg"],
        height: Annotated[float, "Height in meters"],
        precision: Annotated[int, "Decimal places for result"] = 2
    ) -> Annotated[float, "BMI value"]:
        """支持默认参数(可选参数)"""
        bmi = weight / (height ** 2)
        return round(bmi, precision)

class EmailPlugin:
    """邮件处理插件"""
    
    @kernel_function(
        name="send_email",
        description="Send email to recipient"
    )
    async def send_email(
        self,
        content: Annotated[str, "Email content body"],
        receiver: Annotated[str, "Receiver email address"],
        subject: Annotated[str, "Email subject line"],
        cc: Annotated[str, "CC recipients (comma separated)"] = ""
    ) -> Annotated[str, "Send status message"]:
        """异步多参数函数"""
        # 模拟异步操作
        await asyncio.sleep(0.1)
        
        print(f"Sending to: {receiver}")
        print(f"Subject: {subject}")
        print(f"Content: {content[:50]}...")
        
        return f"Email sent successfully to {receiver}"

调用方式

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.functions import KernelArguments

async def main():
    kernel = Kernel()
    
    # 添加插件
    math_plugin = kernel.add_plugin(MathPlugin(), plugin_name="Math")
    email_plugin = kernel.add_plugin(EmailPlugin(), plugin_name="Email")
    
    # ========== 多参数调用方式 1:使用 KernelArguments ==========
    result = await kernel.invoke(
        function=math_plugin["add"],
        arguments=KernelArguments(
            number1=10.5,
            number2=20.3
        )
    )
    print(f"Add result: {result}")  # 30.8
    
    # ========== 多参数调用方式 2:关键字参数(更 Pythonic) ==========
    # 注意:v1.0+ 支持直接传递关键字参数作为 KernelArguments
    result = await kernel.invoke(
        function=math_plugin["calculate_bmi"],
        weight=70,
        height=1.75,
        precision=1
    )
    print(f"BMI: {result}")  # 22.9
    
    # ========== 异步函数调用 ==========
    result = await kernel.invoke(
        function=email_plugin["send_email"],
        content="Hello, this is a test email from Semantic Kernel!",
        receiver="user@example.com",
        subject="Test Email"
        # cc 参数使用默认值 ""
    )
    print(f"Email status: {result}")

if __name__ == "__main__":
    asyncio.run(main())

四、版本演进对比(重要!)

v0.x(旧版) vs v1.0+(新版)

特性 v0.x(旧版) v1.0+(当前)
装饰器 @sk_function @kernel_function
上下文对象 SKContext KernelArguments
参数描述 @sk_function_context_parameter Annotated[type, "description"]
多参数声明 通过 SKContext 字典 直接声明多个参数 + Annotated
方法命名 xxx_async xxx(移除 async 后缀)
调用方式 kernel.run_async() kernel.invoke()

迁移示例

# ========== v0.x 旧版写法 ==========
from semantic_kernel.skill_definition import sk_function, sk_function_context_parameter
from semantic_kernel.orchestration.sk_context import SKContext

class OldPlugin:
    @sk_function(
        description="Add two numbers",
        name="add"
    )
    @sk_function_context_parameter(
        name="number1",
        description="First number"
    )
    @sk_function_context_parameter(
        name="number2",
        description="Second number"
    )
    async def add_async(self, context: SKContext) -> str:
        number1 = float(context["number1"])
        number2 = float(context["number2"])
        return str(number1 + number2)

# ========== v1.0+ 新版写法 ==========
from typing import Annotated
from semantic_kernel.functions import kernel_function

class NewPlugin:
    @kernel_function(
        name="add",
        description="Add two numbers"
    )
    def add(
        self,
        number1: Annotated[float, "First number"],
        number2: Annotated[float, "Second number"]
    ) -> Annotated[float, "Sum of numbers"]:
        # 直接使用方法参数,更清晰!
        return number1 + number2

五、实战运用:构建智能代理

场景:股票分析助手

import yfinance as yf
import pandas as pd
from typing import Annotated
from semantic_kernel.functions import kernel_function

class StockPlugin:
    """股票数据插件"""
    
    @kernel_function(
        name="get_stock_price",
        description="Get current stock price for a ticker symbol"
    )
    def get_stock_price(
        self,
        ticker: Annotated[str, "Stock ticker symbol (e.g., AAPL, MSFT)"]
    ) -> Annotated[str, "Current stock price as string"]:
        """单参数:获取股票价格"""
        stock = yf.Ticker(ticker)
        price = stock.info.get('currentPrice', 'N/A')
        return f"${price}"
    
    @kernel_function(
        name="get_historical_data",
        description="Get historical stock data for a period"
    )
    def get_historical_data(
        self,
        ticker: Annotated[str, "Stock ticker symbol"],
        period: Annotated[str, "Time period (1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max)"] = "1mo",
        interval: Annotated[str, "Data interval (1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo)"] = "1d"
    ) -> Annotated[str, "JSON string of historical data"]:
        """多参数:获取历史数据(含默认值)"""
        stock = yf.Ticker(ticker)
        hist = stock.history(period=period, interval=interval)
        
        # 格式化数据
        hist.index = hist.index.strftime('%Y-%m-%d')
        return hist.head(10).to_json()

# 使用 Function Calling 让 LLM 自动选择
async def analyze_stock():
    from semantic_kernel import Kernel
    from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
    from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings import OpenAIPromptExecutionSettings
    
    kernel = Kernel()
    
    # 配置 OpenAI 服务
    kernel.add_service(OpenAIChatCompletion(
        service_id="default",
        ai_model_id="gpt-4",
        api_key="your-api-key"
    ))
    
    # 添加股票插件
    stock_plugin = kernel.add_plugin(StockPlugin(), plugin_name="Stock")
    
    # 配置自动函数调用
    settings = OpenAIPromptExecutionSettings(
        service_id="default",
        function_call_behavior="AutoInvokeKernelFunctions"  # 自动调用
    )
    
    # LLM 会自动识别需要调用哪个函数
    result = await kernel.invoke_prompt(
        prompt="What's the current price of Apple stock? Also show me the last 5 days of data.",
        settings=settings
    )
    
    print(result)

if __name__ == "__main__":
    import asyncio
    asyncio.run(analyze_stock())

场景:与 Semantic Function 混用

from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function, KernelArguments

# Native Function:数据获取
class DataPlugin:
    @kernel_function(
        name="fetch_user_data",
        description="Fetch user data from database"
    )
    def fetch_user_data(
        self,
        user_id: Annotated[str, "User ID to fetch"]
    ) -> Annotated[str, "User data as JSON"]:
        # 模拟数据库查询
        return f'{{"id": "{user_id}", "name": "John Doe", "tier": "premium"}}'

# Semantic Function:文本生成(通过 YAML 或代码创建)
# config.yaml:
# name: summarize_user
# template: |
#   Based on the user data: {{$user_data}}
#   Write a brief summary of this user's status.

async def mixed_workflow():
    kernel = Kernel()
    
    # 添加插件
    data_plugin = kernel.add_plugin(DataPlugin(), plugin_name="Data")
    
    # 1. 先调用 Native Function 获取数据
    user_data = await kernel.invoke(
        function=data_plugin["fetch_user_data"],
        user_id="user_123"
    )
    
    # 2. 再调用 Semantic Function 生成总结
    summarize_func = kernel.create_function_from_prompt(
        prompt="Based on the user data: {{$user_data}}\nWrite a brief summary.",
        function_name="summarize_user",
        plugin_name="Summary"
    )
    
    summary = await kernel.invoke(
        function=summarize_func,
        user_data=str(user_data)
    )
    
    print(f"Summary: {summary}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(mixed_workflow())

六、最佳实践总结

✅ 应该做的

  1. 单参数保持简单:直接 def func(self, input: str),无需 Annotated
  2. 多参数使用 Annotated:每个参数添加 Annotated[type, "description"],帮助 LLM 理解
  3. 使用具体类型strintfloatbool 等,避免模糊的 Any
  4. 提供默认值:可选参数给默认值,提高灵活性
  5. 添加返回值注解-> Annotated[type, "description"] 让 LLM 理解输出
  6. 异步支持:需要 I/O 操作时使用 async def,SK 会自动处理

❌ 避免做的

  1. 单参数过度使用 KernelArguments:直接传递即可
  2. 参数无描述:多参数时必须用 Annotated,否则 LLM 无法理解
  3. 混合新旧语法:统一使用 v1.0+ 新语法(@kernel_function + Annotated
  4. 在 Native Function 中写 Prompt:那是 Semantic Function 的职责
  5. 使用可变默认参数:避免 def func(self, arg=[]),使用 None 代替

七、完整示例代码

import asyncio
from typing import Annotated
from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function, KernelArguments
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

# ========== 定义 Plugins ==========
class TextPlugin:
    @kernel_function(name="uppercase", description="Convert to uppercase")
    def uppercase(self, input: str) -> str:
        return input.upper()

class MathPlugin:
    @kernel_function(name="add", description="Add two numbers")
    def add(
        self,
        a: Annotated[int, "First number"],
        b: Annotated[int, "Second number"]
    ) -> Annotated[int, "Sum"]:
        return a + b
    
    @kernel_function(name="multiply", description="Multiply two numbers")
    def multiply(
        self,
        x: Annotated[float, "First factor"],
        y: Annotated[float, "Second factor"]
    ) -> Annotated[float, "Product"]:
        return x * y

class UserPlugin:
    @kernel_function(name="format_name", description="Format full name")
    def format_name(
        self,
        first_name: Annotated[str, "First name"],
        last_name: Annotated[str, "Last name"],
        title: Annotated[str, "Optional title"] = ""
    ) -> Annotated[str, "Formatted full name"]:
        if title:
            return f"{title} {first_name} {last_name}"
        return f"{first_name} {last_name}"

# ========== 主程序 ==========
async def main():
    # 初始化 Kernel
    kernel = Kernel()
    
    # 可选:添加 AI 服务(用于自动函数调用)
    # kernel.add_service(OpenAIChatCompletion(
    #     service_id="default",
    #     ai_model_id="gpt-4",
    #     api_key="your-api-key"
    # ))
    
    # 导入所有 Plugins
    text_plugin = kernel.add_plugin(TextPlugin(), plugin_name="Text")
    math_plugin = kernel.add_plugin(MathPlugin(), plugin_name="Math")
    user_plugin = kernel.add_plugin(UserPlugin(), plugin_name="User")
    
    # ========== 单参数调用 ==========
    result = await kernel.invoke(
        function=text_plugin["uppercase"],
        arguments=KernelArguments(input="hello semantic kernel")
    )
    print(f"Uppercase: {result}")
    
    # ========== 多参数调用 ==========
    result = await kernel.invoke(
        function=math_plugin["add"],
        arguments=KernelArguments(a=10, b=25)
    )
    print(f"Sum: {result}")
    
    # ========== 带默认值的调用 ==========
    result = await kernel.invoke(
        function=user_plugin["format_name"],
        first_name="John",
        last_name="Doe",
        title="Dr"
    )
    print(f"Formatted: {result}")
    
    # ========== 自动函数调用示例(需配置 OpenAI) ==========
    # settings = kernel.get_prompt_execution_settings_from_service("default")
    # settings.function_call_behavior = "AutoInvokeKernelFunctions"
    # 
    # result = await kernel.invoke_prompt(
    #     prompt="Calculate (15 * 8) + 23, then convert the result to uppercase",
    #     settings=settings
    # )
    # print(f"AI Result: {result}")

if __name__ == "__main__":
    asyncio.run(main())

结语

Semantic Kernel Python SDK 的 Native Function 设计遵循**“简单场景简单处理,复杂场景灵活处理”**的原则:

  • 单参数:保持极简,直接方法参数,无需 Annotated
  • 多参数:使用 Annotated[type, "description"] 为每个参数添加描述,通过 KernelArguments 传递

v1.0+ 版本的 Python SDK 更加 Pythonic,移除了繁琐的装饰器参数,改用类型注解,与 Python 生态更好地融合。这种设计既保证了日常使用的便捷性,又为复杂业务场景提供了足够的灵活性。


参考资源

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐