mirror of
https://github.com/microsoft/semantic-kernel.git
synced 2026-10-12 02:19:06 +00:00
Evan Mattson
6b01468efb
### Motivation and Context The current kernel function decorator class parses function params, but had a miss related to how the underlying param's type is parsed if not using Annotations. <!-- Thank you for your contribution to the semantic-kernel repo! Please help reviewers and future users, providing the following information: 1. Why is this change required? 2. What problem does it solve? 3. What scenario does it contribute to? 4. If it fixes an open issue, please link to the issue here. --> ### Description This PR properly handles the type_object for params that don't using typing.Annotations. - Adds unit tests to exercise the code. - Adds more kernel.py unit tests for near 100% test coverage. - Adds some planner tests to improve the test coverage. - Closes #7300 <!-- Describe your changes, the overall approach, the underlying design. These notes will help understanding how your code works. Thanks! --> ### Contribution Checklist <!-- Before submitting this PR, please make sure: --> - [X] The code builds clean without any errors or warnings - [X] The PR follows the [SK Contribution Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md) and the [pre-submission formatting script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts) raises no violations - [X] All unit tests pass, and I have added new tests where possible - [X] I didn't break anyone 😄
188 lines
8.8 KiB
Python
188 lines
8.8 KiB
Python
# Copyright (c) Microsoft. All rights reserved.
|
|
|
|
import inspect
|
|
import logging
|
|
from collections.abc import Callable
|
|
from inspect import isasyncgen, isasyncgenfunction, isawaitable, iscoroutinefunction, isgenerator, isgeneratorfunction
|
|
from typing import Any
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from semantic_kernel.exceptions import FunctionExecutionException, FunctionInitializationError
|
|
from semantic_kernel.filters.functions.function_invocation_context import FunctionInvocationContext
|
|
from semantic_kernel.functions.function_result import FunctionResult
|
|
from semantic_kernel.functions.kernel_function import KernelFunction
|
|
from semantic_kernel.functions.kernel_function_metadata import KernelFunctionMetadata
|
|
from semantic_kernel.functions.kernel_parameter_metadata import KernelParameterMetadata
|
|
|
|
logger: logging.Logger = logging.getLogger(__name__)
|
|
|
|
|
|
class KernelFunctionFromMethod(KernelFunction):
|
|
"""Semantic Kernel Function from a method."""
|
|
|
|
method: Callable[..., Any]
|
|
stream_method: Callable[..., Any] | None = None
|
|
|
|
def __init__(
|
|
self,
|
|
method: Callable[..., Any],
|
|
plugin_name: str | None = None,
|
|
stream_method: Callable[..., Any] | None = None,
|
|
parameters: list[KernelParameterMetadata] | None = None,
|
|
return_parameter: KernelParameterMetadata | None = None,
|
|
additional_metadata: dict[str, Any] | None = None,
|
|
) -> None:
|
|
"""Initializes a new instance of the KernelFunctionFromMethod class.
|
|
|
|
Args:
|
|
method (Callable[..., Any]): The method to be called
|
|
plugin_name (str | None): The name of the plugin
|
|
stream_method (Callable[..., Any] | None): The stream method for the function
|
|
parameters (list[KernelParameterMetadata] | None): The parameters of the function
|
|
return_parameter (KernelParameterMetadata | None): The return parameter of the function
|
|
additional_metadata (dict[str, Any] | None): Additional metadata for the function
|
|
"""
|
|
if method is None:
|
|
raise FunctionInitializationError("Method cannot be `None`")
|
|
|
|
if not hasattr(method, "__kernel_function__") or method.__kernel_function__ is None:
|
|
raise FunctionInitializationError("Method is not a Kernel function")
|
|
|
|
# all these fields are created when the kernel function decorator is used,
|
|
# so no need to check before using, will raise an exception if not set
|
|
function_name = method.__kernel_function_name__ # type: ignore
|
|
description = method.__kernel_function_description__ # type: ignore
|
|
if parameters is None:
|
|
parameters = [KernelParameterMetadata(**param) for param in method.__kernel_function_parameters__] # type: ignore
|
|
if return_parameter is None:
|
|
return_parameter = KernelParameterMetadata(
|
|
name="return",
|
|
description=method.__kernel_function_return_description__, # type: ignore
|
|
default_value=None,
|
|
type_=method.__kernel_function_return_type__, # type: ignore
|
|
type_object=method.__kernel_function_return_type_object__, # type: ignore
|
|
is_required=method.__kernel_function_return_required__, # type: ignore
|
|
)
|
|
|
|
try:
|
|
metadata = KernelFunctionMetadata(
|
|
name=function_name,
|
|
description=description,
|
|
parameters=parameters,
|
|
return_parameter=return_parameter,
|
|
is_prompt=False,
|
|
is_asynchronous=isasyncgenfunction(method) or iscoroutinefunction(method),
|
|
plugin_name=plugin_name,
|
|
additional_properties=additional_metadata if additional_metadata is not None else {},
|
|
)
|
|
except ValidationError as exc:
|
|
# reraise the exception to clarify it comes from KernelFunction init
|
|
raise FunctionInitializationError("Failed to create KernelFunctionMetadata") from exc
|
|
|
|
args: dict[str, Any] = {
|
|
"metadata": metadata,
|
|
"method": method,
|
|
"stream_method": (
|
|
stream_method
|
|
if stream_method is not None
|
|
else method
|
|
if isasyncgenfunction(method) or isgeneratorfunction(method)
|
|
else None
|
|
),
|
|
}
|
|
|
|
super().__init__(**args)
|
|
|
|
async def _invoke_internal(
|
|
self,
|
|
context: FunctionInvocationContext,
|
|
) -> None:
|
|
"""Invoke the function with the given arguments."""
|
|
function_arguments = self.gather_function_parameters(context)
|
|
result = self.method(**function_arguments)
|
|
if isasyncgen(result):
|
|
result = [x async for x in result]
|
|
elif isawaitable(result):
|
|
result = await result
|
|
elif isgenerator(result):
|
|
result = list(result)
|
|
if not isinstance(result, FunctionResult):
|
|
result = FunctionResult(
|
|
function=self.metadata,
|
|
value=result,
|
|
metadata={"arguments": context.arguments, "used_arguments": function_arguments},
|
|
)
|
|
context.result = result
|
|
|
|
async def _invoke_internal_stream(self, context: FunctionInvocationContext) -> None:
|
|
if self.stream_method is None:
|
|
raise NotImplementedError("Stream method not implemented")
|
|
function_arguments = self.gather_function_parameters(context)
|
|
context.result = FunctionResult(function=self.metadata, value=self.stream_method(**function_arguments))
|
|
|
|
def _parse_parameter(self, value: Any, param_type: Any) -> Any:
|
|
"""Parses the value into the specified param_type, including handling lists of types."""
|
|
if isinstance(param_type, type) and hasattr(param_type, "model_validate"):
|
|
try:
|
|
return param_type.model_validate(value)
|
|
except Exception as exc:
|
|
raise FunctionExecutionException(
|
|
f"Parameter is expected to be parsed to {param_type} but is not."
|
|
) from exc
|
|
elif hasattr(param_type, "__origin__") and param_type.__origin__ is list:
|
|
if isinstance(value, list):
|
|
item_type = param_type.__args__[0]
|
|
return [self._parse_parameter(item, item_type) for item in value]
|
|
raise FunctionExecutionException(f"Expected a list for {param_type}, but got {type(value)}")
|
|
else:
|
|
try:
|
|
if isinstance(value, dict) and hasattr(param_type, "__init__"):
|
|
return param_type(**value)
|
|
return param_type(value)
|
|
except Exception as exc:
|
|
raise FunctionExecutionException(
|
|
f"Parameter is expected to be parsed to {param_type} but is not."
|
|
) from exc
|
|
|
|
def gather_function_parameters(self, context: FunctionInvocationContext) -> dict[str, Any]:
|
|
"""Gathers the function parameters from the arguments."""
|
|
function_arguments: dict[str, Any] = {}
|
|
for param in self.parameters:
|
|
if param.name is None:
|
|
raise FunctionExecutionException("Parameter name cannot be None")
|
|
if param.name == "kernel":
|
|
function_arguments[param.name] = context.kernel
|
|
continue
|
|
if param.name == "service":
|
|
function_arguments[param.name] = context.kernel.select_ai_service(self, context.arguments)[0]
|
|
continue
|
|
if param.name == "execution_settings":
|
|
function_arguments[param.name] = context.kernel.select_ai_service(self, context.arguments)[1]
|
|
continue
|
|
if param.name == "arguments":
|
|
function_arguments[param.name] = context.arguments
|
|
continue
|
|
if param.name in context.arguments:
|
|
value: Any = context.arguments[param.name]
|
|
if (
|
|
param.type_
|
|
and "," not in param.type_
|
|
and param.type_object
|
|
and param.type_object is not inspect._empty
|
|
):
|
|
try:
|
|
value = self._parse_parameter(value, param.type_object)
|
|
except Exception as exc:
|
|
raise FunctionExecutionException(
|
|
f"Parameter {param.name} is expected to be parsed to {param.type_object} but is not."
|
|
) from exc
|
|
function_arguments[param.name] = value
|
|
continue
|
|
if param.is_required:
|
|
raise FunctionExecutionException(
|
|
f"Parameter {param.name} is required but not provided in the arguments."
|
|
)
|
|
logger.debug(f"Parameter {param.name} is not provided, using default value {param.default_value}")
|
|
return function_arguments
|