diff --git a/python/samples/concepts/auto_function_calling/chat_gpt_api_function_calling.py b/python/samples/concepts/auto_function_calling/chat_gpt_api_function_calling.py index 11f872d7a9..005da7c522 100644 --- a/python/samples/concepts/auto_function_calling/chat_gpt_api_function_calling.py +++ b/python/samples/concepts/auto_function_calling/chat_gpt_api_function_calling.py @@ -11,7 +11,8 @@ from semantic_kernel.contents import ChatHistory from semantic_kernel.contents.chat_message_content import ChatMessageContent from semantic_kernel.contents.function_call_content import FunctionCallContent from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent -from semantic_kernel.core_plugins import MathPlugin, TimePlugin +from semantic_kernel.core_plugins.math_plugin import MathPlugin +from semantic_kernel.core_plugins.time_plugin import TimePlugin from semantic_kernel.functions import KernelArguments if TYPE_CHECKING: diff --git a/python/semantic_kernel/functions/kernel_function_decorator.py b/python/semantic_kernel/functions/kernel_function_decorator.py index 61b247f25a..1047d0bfe5 100644 --- a/python/semantic_kernel/functions/kernel_function_decorator.py +++ b/python/semantic_kernel/functions/kernel_function_decorator.py @@ -4,7 +4,7 @@ import logging import types from collections.abc import Callable from inspect import Parameter, Signature, isasyncgenfunction, isclass, isgeneratorfunction, signature -from typing import Any, ForwardRef, Union, get_args +from typing import Annotated, Any, ForwardRef, Union, get_args, get_origin NoneType = type(None) logger = logging.getLogger(__name__) @@ -103,7 +103,10 @@ def _process_signature(func_sig: Signature) -> list[dict[str, Any]]: annotation = arg.annotation default = arg.default if arg.default != arg.empty else None parsed_annotation = _parse_parameter(arg.name, annotation, default) - underlying_type = _get_underlying_type(annotation) + if get_origin(annotation) is Annotated or get_origin(annotation) in {Union, types.UnionType}: + underlying_type = _get_underlying_type(annotation) + else: + underlying_type = annotation parsed_annotation["type_object"] = underlying_type annotations.append(parsed_annotation) diff --git a/python/semantic_kernel/functions/kernel_function_from_method.py b/python/semantic_kernel/functions/kernel_function_from_method.py index e97c84205d..0daea5af0a 100644 --- a/python/semantic_kernel/functions/kernel_function_from_method.py +++ b/python/semantic_kernel/functions/kernel_function_from_method.py @@ -121,6 +121,30 @@ class KernelFunctionFromMethod(KernelFunction): 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] = {} @@ -147,24 +171,12 @@ class KernelFunctionFromMethod(KernelFunction): and param.type_object and param.type_object is not inspect._empty ): - if hasattr(param.type_object, "model_validate"): - try: - value = param.type_object.model_validate(value) - except Exception as exc: - raise FunctionExecutionException( - f"Parameter {param.name} is expected to be parsed to {param.type_} but is not." - ) from exc - else: - try: - if isinstance(value, dict) and hasattr(param.type_object, "__init__"): - value = param.type_object(**value) - else: - value = param.type_object(value) - except Exception as exc: - raise FunctionExecutionException( - f"Parameter {param.name} is expected to be parsed to " - f"{param.type_object} but is not." - ) from exc + 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: diff --git a/python/tests/conftest.py b/python/tests/conftest.py index f58dde8744..bff9b5eae2 100644 --- a/python/tests/conftest.py +++ b/python/tests/conftest.py @@ -3,9 +3,12 @@ import warnings from collections.abc import Callable from typing import TYPE_CHECKING +from unittest.mock import MagicMock import pytest +from semantic_kernel.contents.function_call_content import FunctionCallContent + if TYPE_CHECKING: from semantic_kernel.contents.chat_history import ChatHistory from semantic_kernel.filters.functions.function_invocation_context import FunctionInvocationContext @@ -134,6 +137,24 @@ def create_mock_function() -> Callable: return create_mock_function +@pytest.fixture(scope="function") +def get_tool_call_mock(): + tool_call_mock = MagicMock(spec=FunctionCallContent) + tool_call_mock.split_name_dict.return_value = {"arg_name": "arg_value"} + tool_call_mock.to_kernel_arguments.return_value = {"arg_name": "arg_value"} + tool_call_mock.name = "test-function" + tool_call_mock.function_name = "function" + tool_call_mock.plugin_name = "test" + tool_call_mock.arguments = {"arg_name": "arg_value"} + tool_call_mock.ai_model_id = None + tool_call_mock.metadata = {} + tool_call_mock.index = 0 + tool_call_mock.parse_arguments.return_value = {"arg_name": "arg_value"} + tool_call_mock.id = "test_id" + + return tool_call_mock + + @pytest.fixture(scope="function") def chat_history() -> "ChatHistory": from semantic_kernel.contents.chat_history import ChatHistory diff --git a/python/tests/unit/functions/test_kernel_function_decorators.py b/python/tests/unit/functions/test_kernel_function_decorators.py index b9cb0174b5..803e0d2ffc 100644 --- a/python/tests/unit/functions/test_kernel_function_decorators.py +++ b/python/tests/unit/functions/test_kernel_function_decorators.py @@ -63,7 +63,7 @@ class MiscClass: return input @kernel_function - def func_return_type_streaming(self, input: str) -> Annotated[AsyncGenerator[str, Any], "test return"]: + def func_return_type_streaming(self, input: str) -> Annotated[AsyncGenerator[str, Any], "test return"]: # type: ignore yield input @kernel_function diff --git a/python/tests/unit/functions/test_kernel_function_from_method.py b/python/tests/unit/functions/test_kernel_function_from_method.py index 9944d19d68..3f9566dc73 100644 --- a/python/tests/unit/functions/test_kernel_function_from_method.py +++ b/python/tests/unit/functions/test_kernel_function_from_method.py @@ -1,11 +1,14 @@ # Copyright (c) Microsoft. All rights reserved. from collections.abc import AsyncGenerator, Iterable from typing import Annotated, Any +from unittest.mock import Mock import pytest from semantic_kernel.connectors.ai.open_ai.services.open_ai_chat_completion import OpenAIChatCompletion from semantic_kernel.exceptions import FunctionExecutionException, FunctionInitializationError +from semantic_kernel.filters.functions.function_invocation_context import FunctionInvocationContext +from semantic_kernel.filters.kernel_filters_extension import _rebuild_function_invocation_context from semantic_kernel.functions.function_result import FunctionResult from semantic_kernel.functions.kernel_arguments import KernelArguments from semantic_kernel.functions.kernel_function import KernelFunction @@ -16,6 +19,38 @@ from semantic_kernel.kernel import Kernel from semantic_kernel.kernel_pydantic import KernelBaseModel +class CustomType(KernelBaseModel): + id: str + name: str + + +class CustomTypeNonPydantic: + id: str + name: str + + def __init__(self, id: str, name: str): + self.id = id + self.name = name + + +@pytest.fixture +def get_custom_type_function_pydantic(): + @kernel_function + def func_default(param: list[CustomType]): + return input + + return KernelFunction.from_method(func_default, "test") + + +@pytest.fixture +def get_custom_type_function_nonpydantic(): + @kernel_function + def func_default(param: list[CustomTypeNonPydantic]): + return input + + return KernelFunction.from_method(func_default, "test") + + def test_init_native_function_with_input_description(): @kernel_function(description="Mock description", name="mock_function") def mock_function(input: Annotated[str, "input"], arguments: "KernelArguments") -> None: @@ -449,3 +484,79 @@ async def test_default_handling_2(kernel: Kernel): res = await kernel.invoke(func, base="base") assert str(res) == "test" + + +def test_parse_list_of_objects(get_custom_type_function_pydantic): + func = get_custom_type_function_pydantic + + param_type = list[CustomType] + value = [{"id": "1", "name": "John"}, {"id": "2", "name": "Jane"}] + result = func._parse_parameter(value, param_type) + assert isinstance(result, list) + assert len(result) == 2 + assert all(isinstance(item, CustomType) for item in result) + + +def test_parse_individual_object(get_custom_type_function_pydantic): + value = {"id": "2", "name": "Jane"} + func = get_custom_type_function_pydantic + result = func._parse_parameter(value, CustomType) + assert isinstance(result, CustomType) + assert result.id == "2" + assert result.name == "Jane" + + +def test_parse_non_list_raises_exception(get_custom_type_function_pydantic): + func = get_custom_type_function_pydantic + param_type = list[CustomType] + value = {"id": "2", "name": "Jane"} + with pytest.raises(FunctionExecutionException, match=r"Expected a list for .*"): + func._parse_parameter(value, param_type) + + +def test_parse_invalid_dict_raises_exception(get_custom_type_function_pydantic): + func = get_custom_type_function_pydantic + value = {"id": "1"} + with pytest.raises(FunctionExecutionException, match=r"Parameter is expected to be parsed to .*"): + func._parse_parameter(value, CustomType) + + +def test_parse_invalid_value_raises_exception(get_custom_type_function_pydantic): + func = get_custom_type_function_pydantic + value = "invalid_value" + with pytest.raises(FunctionExecutionException, match=r"Parameter is expected to be parsed to .*"): + func._parse_parameter(value, CustomType) + + +def test_parse_invalid_list_raises_exception(get_custom_type_function_pydantic): + func = get_custom_type_function_pydantic + param_type = list[CustomType] + value = ["invalid_value"] + with pytest.raises(FunctionExecutionException, match=r"Parameter is expected to be parsed to .*"): + func._parse_parameter(value, param_type) + + +def test_parse_dict_with_init_non_pydantic(get_custom_type_function_nonpydantic): + func = get_custom_type_function_nonpydantic + value = {"id": "3", "name": "Alice"} + result = func._parse_parameter(value, CustomTypeNonPydantic) + assert isinstance(result, CustomTypeNonPydantic) + assert result.id == "3" + assert result.name == "Alice" + + +def test_parse_invalid_dict_raises_exception_new(get_custom_type_function_nonpydantic): + func = get_custom_type_function_nonpydantic + value = {"wrong_key": "3", "name": "Alice"} + with pytest.raises(FunctionExecutionException, match=r"Parameter is expected to be parsed to .*"): + func._parse_parameter(value, CustomTypeNonPydantic) + + +def test_gather_function_parameters_exception_handling(get_custom_type_function_pydantic): + kernel = Mock(spec=Kernel) # Mock kernel + func = get_custom_type_function_pydantic + _rebuild_function_invocation_context() + context = FunctionInvocationContext(kernel=kernel, function=func, arguments=KernelArguments(param="test")) + + with pytest.raises(FunctionExecutionException, match=r"Parameter param is expected to be parsed to .* but is not."): + func.gather_function_parameters(context) diff --git a/python/tests/unit/kernel/test_kernel.py b/python/tests/unit/kernel/test_kernel.py index f4e03aff39..3f504035dd 100644 --- a/python/tests/unit/kernel/test_kernel.py +++ b/python/tests/unit/kernel/test_kernel.py @@ -19,17 +19,23 @@ from semantic_kernel.contents import ChatMessageContent from semantic_kernel.contents.chat_history import ChatHistory from semantic_kernel.contents.function_call_content import FunctionCallContent from semantic_kernel.exceptions import ( - FunctionCallInvalidArgumentsException, KernelFunctionAlreadyExistsError, KernelServiceNotFoundError, ) -from semantic_kernel.exceptions.kernel_exceptions import KernelFunctionNotFoundError, KernelPluginNotFoundError +from semantic_kernel.exceptions.content_exceptions import FunctionCallInvalidArgumentsException +from semantic_kernel.exceptions.kernel_exceptions import ( + KernelFunctionNotFoundError, + KernelInvokeException, + KernelPluginNotFoundError, + OperationCancelledException, +) from semantic_kernel.exceptions.template_engine_exceptions import TemplateSyntaxError from semantic_kernel.functions.function_result import FunctionResult from semantic_kernel.functions.kernel_arguments import KernelArguments from semantic_kernel.functions.kernel_function import KernelFunction from semantic_kernel.functions.kernel_function_decorator import kernel_function from semantic_kernel.functions.kernel_function_metadata import KernelFunctionMetadata +from semantic_kernel.functions.kernel_parameter_metadata import KernelParameterMetadata from semantic_kernel.functions.kernel_plugin import KernelPlugin from semantic_kernel.prompt_template.kernel_prompt_template import KernelPromptTemplate from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig @@ -104,6 +110,21 @@ async def test_invoke_function(kernel: Kernel, create_mock_function): assert mock_function.call_count == 1 +@pytest.mark.asyncio +async def test_invoke_function_with_cancellation(kernel: Kernel, create_mock_function): + mock_function = create_mock_function(name="test_function") + + with ( + patch.object(Kernel, "invoke", AsyncMock(side_effect=OperationCancelledException("Operation cancelled"))), + pytest.raises(OperationCancelledException), + ): + result = await kernel.invoke(function=mock_function, arguments=KernelArguments()) + + assert result is None + + Kernel.invoke.assert_called_once_with(function=mock_function, arguments=KernelArguments()) + + @pytest.mark.asyncio async def test_invoke_functions_by_name(kernel: Kernel, create_mock_function): mock_function = kernel.add_function(plugin_name="test", function=create_mock_function(name="test_function")) @@ -117,6 +138,21 @@ async def test_invoke_functions_by_name(kernel: Kernel, create_mock_function): assert response[0].text == "test" +@pytest.mark.asyncio +async def test_invoke_functions_by_name_return_function_results(kernel: Kernel, create_mock_function): + mock_function = kernel.add_function(plugin_name="test", function=create_mock_function(name="test_function")) + + result = await kernel.invoke(function_name="test_function", plugin_name="test", arguments=KernelArguments()) + assert str(result) == "test" + + assert mock_function.call_count == 1 + + async for _ in kernel.invoke_stream( + function_name="test_function", plugin_name="test", return_function_results=True + ): + assert isinstance(result, FunctionResult) + + @pytest.mark.asyncio async def test_invoke_function_fail(kernel: Kernel, create_mock_function): mock_function = create_mock_function(name="test_function") @@ -182,19 +218,124 @@ async def test_invoke_prompt_no_prompt_error(kernel: Kernel): @pytest.mark.asyncio -async def test_invoke_function_call(kernel: Kernel): - tool_call_mock = MagicMock(spec=FunctionCallContent) - tool_call_mock.split_name_dict.return_value = {"arg_name": "arg_value"} - tool_call_mock.to_kernel_arguments.return_value = {"arg_name": "arg_value"} - tool_call_mock.name = "test-function" - tool_call_mock.function_name = "function" - tool_call_mock.plugin_name = "test" - tool_call_mock.arguments = {"arg_name": "arg_value"} - tool_call_mock.ai_model_id = None - tool_call_mock.metadata = {} - tool_call_mock.index = 0 - tool_call_mock.parse_arguments.return_value = {"arg_name": "arg_value"} - tool_call_mock.id = "test_id" +async def test_invoke_prompt_stream_no_prompt_throws(kernel: Kernel): + with pytest.raises(TemplateSyntaxError): + async for _ in kernel.invoke_prompt_stream(prompt=""): + pass + + +@pytest.mark.asyncio +async def test_invoke_prompt_stream(kernel: Kernel, create_mock_function): + mock_function = create_mock_function(name="test_function") + with ( + patch( + "semantic_kernel.kernel.Kernel.invoke_stream", + ) as mock_stream_invoke, + ): + mock_stream_invoke.return_value.__aiter__.return_value = [ + FunctionResult(function=mock_function.metadata, value="test") + ] + async for response in kernel.invoke_prompt_stream(prompt="test"): + assert response.value == "test" + + +@pytest.mark.asyncio +async def test_invoke_prompt_stream_returns_function_results(kernel: Kernel, create_mock_function): + mock_function = create_mock_function(name="test_function") + with ( + patch( + "semantic_kernel.kernel.Kernel.invoke_stream", + ) as mock_stream_invoke, + ): + mock_stream_invoke.return_value.__aiter__.return_value = [ + FunctionResult(function=mock_function.metadata, value="test") + ] + async for response in kernel.invoke_prompt_stream(prompt="test", return_function_results=True): + assert isinstance(response, FunctionResult) + + +@pytest.mark.asyncio +async def test_invoke_prompt_stream_raises_exception(kernel: Kernel, create_mock_function): + mock_function = create_mock_function(name="test_function") + with ( + patch( + "semantic_kernel.kernel.Kernel.invoke_stream", + ) as mock_stream_invoke, + pytest.raises(KernelInvokeException), + ): + mock_stream_invoke.return_value.__aiter__.return_value = [ + FunctionResult( + function=mock_function.metadata, value="", metadata={METADATA_EXCEPTION_KEY: KernelInvokeException()} + ) + ] + async for _ in kernel.invoke_prompt_stream(prompt="test"): + pass + + +@pytest.mark.asyncio +async def test_invoke_function_call(kernel: Kernel, get_tool_call_mock): + tool_call_mock = get_tool_call_mock + result_mock = MagicMock(spec=ChatMessageContent) + result_mock.items = [tool_call_mock] + chat_history_mock = MagicMock(spec=ChatHistory) + + func_mock = AsyncMock(spec=KernelFunction) + func_meta = KernelFunctionMetadata(name="function", is_prompt=False) + func_mock.metadata = func_meta + func_mock.name = "function" + func_result = FunctionResult(value="Function result", function=func_meta) + func_mock.invoke = MagicMock(return_value=func_result) + + arguments = KernelArguments() + + with ( + patch("semantic_kernel.kernel.logger", autospec=True), + patch("semantic_kernel.kernel.Kernel.get_list_of_function_metadata", return_value=[func_meta]), + ): + await kernel.invoke_function_call( + tool_call_mock, + chat_history_mock, + arguments, + 1, + 0, + FunctionChoiceBehavior.Auto(filters={"included_functions": ["function"]}), + ) + + +@pytest.mark.asyncio +async def test_invoke_function_call_throws_during_invoke(kernel: Kernel, get_tool_call_mock): + tool_call_mock = get_tool_call_mock + result_mock = MagicMock(spec=ChatMessageContent) + result_mock.items = [tool_call_mock] + chat_history_mock = MagicMock(spec=ChatHistory) + + func_mock = AsyncMock(spec=KernelFunction) + func_meta = KernelFunctionMetadata(name="function", is_prompt=False) + func_mock.metadata = func_meta + func_mock.name = "function" + func_result = FunctionResult(value="Function result", function=func_meta) + func_mock.invoke = MagicMock(return_value=func_result) + + arguments = KernelArguments() + + with ( + patch("semantic_kernel.kernel.logger", autospec=True), + patch("semantic_kernel.kernel.Kernel.get_list_of_function_metadata", return_value=[func_meta]), + patch("semantic_kernel.kernel.Kernel.get_function", return_value=func_mock), + ): + await kernel.invoke_function_call( + tool_call_mock, + chat_history_mock, + arguments, + 1, + 0, + FunctionChoiceBehavior.Auto(), + ) + + +@pytest.mark.asyncio +async def test_invoke_function_call_non_allowed_func_throws(kernel: Kernel, get_tool_call_mock): + tool_call_mock = get_tool_call_mock result_mock = MagicMock(spec=ChatMessageContent) result_mock.items = [tool_call_mock] chat_history_mock = MagicMock(spec=ChatHistory) @@ -209,6 +350,36 @@ async def test_invoke_function_call(kernel: Kernel): arguments = KernelArguments() with patch("semantic_kernel.kernel.logger", autospec=True): + await kernel.invoke_function_call( + tool_call_mock, + chat_history_mock, + arguments, + 1, + 0, + FunctionChoiceBehavior.Auto(filters={"included_functions": ["unknown"]}), + ) + + +@pytest.mark.asyncio +async def test_invoke_function_call_no_name_throws(kernel: Kernel, get_tool_call_mock): + tool_call_mock = get_tool_call_mock + tool_call_mock.name = None + result_mock = MagicMock(spec=ChatMessageContent) + result_mock.items = [tool_call_mock] + chat_history_mock = MagicMock(spec=ChatHistory) + + func_mock = AsyncMock(spec=KernelFunction) + func_meta = KernelFunctionMetadata(name="function", is_prompt=False) + func_mock.metadata = func_meta + func_mock.name = "function" + func_result = FunctionResult(value="Function result", function=func_meta) + func_mock.invoke = MagicMock(return_value=func_result) + + arguments = KernelArguments() + + with ( + patch("semantic_kernel.kernel.logger", autospec=True), + ): await kernel.invoke_function_call( tool_call_mock, chat_history_mock, @@ -220,7 +391,39 @@ async def test_invoke_function_call(kernel: Kernel): @pytest.mark.asyncio -async def test_invoke_function_call_with_continuation_on_malformed_arguments(kernel: Kernel): +async def test_invoke_function_call_not_enough_parsed_args(kernel: Kernel, get_tool_call_mock): + tool_call_mock = get_tool_call_mock + tool_call_mock.to_kernel_arguments.return_value = {} + result_mock = MagicMock(spec=ChatMessageContent) + result_mock.items = [tool_call_mock] + chat_history_mock = MagicMock(spec=ChatHistory) + + func_mock = AsyncMock(spec=KernelFunction) + func_meta = KernelFunctionMetadata(name="function", is_prompt=False) + func_mock.metadata = func_meta + func_mock.name = "function" + func_mock.parameters = [KernelParameterMetadata(name="param1", is_required=True)] + func_result = FunctionResult(value="Function result", function=func_meta) + func_mock.invoke = MagicMock(return_value=func_result) + + arguments = KernelArguments() + + with ( + patch("semantic_kernel.kernel.logger", autospec=True), + patch("semantic_kernel.kernel.Kernel.get_function", return_value=func_mock), + ): + await kernel.invoke_function_call( + tool_call_mock, + chat_history_mock, + arguments, + 1, + 0, + FunctionChoiceBehavior.Auto(), + ) + + +@pytest.mark.asyncio +async def test_invoke_function_call_with_continuation_on_malformed_arguments(kernel: Kernel, get_tool_call_mock): tool_call_mock = MagicMock(spec=FunctionCallContent) tool_call_mock.to_kernel_arguments.side_effect = FunctionCallInvalidArgumentsException("Malformed arguments") tool_call_mock.name = "test-function" diff --git a/python/tests/unit/planners/function_calling_stepwise_planner/test_function_calling_stepwise_planner.py b/python/tests/unit/planners/function_calling_stepwise_planner/test_function_calling_stepwise_planner.py index 0f89df2a62..6dd6508972 100644 --- a/python/tests/unit/planners/function_calling_stepwise_planner/test_function_calling_stepwise_planner.py +++ b/python/tests/unit/planners/function_calling_stepwise_planner/test_function_calling_stepwise_planner.py @@ -6,9 +6,15 @@ import pytest from semantic_kernel.connectors.ai.open_ai.services.open_ai_chat_completion import OpenAIChatCompletion from semantic_kernel.contents import AuthorRole +from semantic_kernel.contents.chat_history import ChatHistory +from semantic_kernel.contents.function_call_content import FunctionCallContent +from semantic_kernel.contents.function_result_content import FunctionResultContent +from semantic_kernel.contents.text_content import TextContent from semantic_kernel.exceptions.planner_exceptions import PlannerInvalidConfigurationError from semantic_kernel.functions.kernel_arguments import KernelArguments 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 from semantic_kernel.kernel import Kernel from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner import ( FunctionCallingStepwisePlanner, @@ -16,6 +22,50 @@ from semantic_kernel.planners.function_calling_stepwise_planner.function_calling from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner_options import ( FunctionCallingStepwisePlannerOptions, ) +from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner_result import ( + FunctionCallingStepwisePlannerResult, + UserInteraction, +) + + +@pytest.fixture +def get_function_call_content(): + function_call_content = AsyncMock(spec=FunctionCallContent) + function_call_content.id = "test" + function_call_content.function_name = "function" + function_call_content.plugin_name = "plugin" + function_call_content.name = "USER_INTERACTION_SEND_FINAL_ANSWER" + function_call_content.parse_arguments.return_value = {"answer": "Final answer"} + function_call_content.ai_model_id = "test_model" + function_call_content.metadata = KernelFunctionMetadata( + name="function", + plugin_name="plugin", + description="A sample function", + parameters=[ + KernelParameterMetadata(name="param1", description="Parameter 1", default_value=None), + KernelParameterMetadata(name="param2", description="Parameter 2", default_value="default"), + ], + is_prompt=False, + is_asynchronous=True, + ) + return function_call_content + + +@pytest.fixture +def get_kernel_function_metadata_list(): + [ + KernelFunctionMetadata( + name="function", + plugin_name="plugin", + description="A sample function", + parameters=[ + KernelParameterMetadata(name="param1", description="Parameter 1", default_value=None), + KernelParameterMetadata(name="param2", description="Parameter 2", default_value="default"), + ], + is_prompt=False, + is_asynchronous=True, + ) + ] def test_initialization(): @@ -92,3 +142,169 @@ async def test_invoke_with_no_configured_AI_service_raises_exception(kernel: Ker planner = FunctionCallingStepwisePlanner(service_id="test", options=None) with pytest.raises(PlannerInvalidConfigurationError): await planner.invoke(kernel, "test question") + + +@pytest.mark.asyncio +async def test_invoke_with_function_call_content_and_processing( + get_function_call_content, get_kernel_function_metadata_list +): + kernel_mock = AsyncMock(spec=Kernel) + question = "Test question" + arguments = KernelArguments() + + options = FunctionCallingStepwisePlannerOptions(get_available_functions=AsyncMock(), max_iterations=1) + options.get_available_functions.return_value = get_kernel_function_metadata_list + + planner = FunctionCallingStepwisePlanner(service_id="test_service", options=options) + + chat_completion_mock = AsyncMock(spec=OpenAIChatCompletion) + kernel_mock.get_service.return_value = chat_completion_mock + + planner._generate_plan = AsyncMock(return_value="generated plan") + planner._build_chat_history_for_step = AsyncMock(return_value=AsyncMock(spec=ChatHistory)) + chat_completion_mock.instantiate_prompt_execution_settings.return_value = AsyncMock() + + other_function_call_content = AsyncMock(spec=FunctionCallContent) + other_function_call_content.name = "SomeOtherFunction" + + chat_result = AsyncMock() + chat_result.items = [other_function_call_content] + chat_completion_mock.get_chat_message_contents.return_value = [chat_result] + + chat_completion_mock._process_function_call = AsyncMock(return_value=MagicMock(function_result="Function result")) + + with patch( + "semantic_kernel.contents.function_result_content.FunctionResultContent.from_function_call_content_and_result", + return_value=MagicMock(), + ) as frc_mock: + result = await planner.invoke(kernel_mock, question, arguments) + + assert isinstance(result, FunctionCallingStepwisePlannerResult) + assert result.final_answer == "" + assert result.iterations == options.max_iterations + + chat_completion_mock._process_function_call.assert_called_with( + function_call=other_function_call_content, + kernel=kernel_mock, + chat_history=planner._build_chat_history_for_step.return_value, + arguments=arguments, + function_call_count=1, + request_index=0, + function_call_behavior=chat_completion_mock.instantiate_prompt_execution_settings.return_value.function_choice_behavior, + ) + + frc_mock.assert_called_with(function_call_content=other_function_call_content, result="Function result") + + +@pytest.mark.asyncio +async def test_invoke_with_function_call_content_and_processing_error(get_kernel_function_metadata_list): + kernel_mock = AsyncMock(spec=Kernel) + question = "Test question" + arguments = KernelArguments() + + options = FunctionCallingStepwisePlannerOptions(get_available_functions=AsyncMock(), max_iterations=1) + options.get_available_functions.return_value = get_kernel_function_metadata_list + + planner = FunctionCallingStepwisePlanner(service_id="test_service", options=options) + + chat_completion_mock = AsyncMock(spec=OpenAIChatCompletion) + kernel_mock.get_service.return_value = chat_completion_mock + + planner._generate_plan = AsyncMock(return_value="generated plan") + planner._build_chat_history_for_step = AsyncMock(return_value=AsyncMock(spec=ChatHistory)) + chat_completion_mock.instantiate_prompt_execution_settings.return_value = AsyncMock() + + other_function_call_content = AsyncMock(spec=FunctionCallContent) + other_function_call_content.name = "SomeOtherFunction" + + chat_result = AsyncMock() + chat_result.items = [other_function_call_content] + chat_completion_mock.get_chat_message_contents.return_value = [chat_result] + + chat_completion_mock._process_function_call.side_effect = Exception("Function call error") + + with patch( + "semantic_kernel.contents.function_result_content.FunctionResultContent.from_function_call_content_and_result", + return_value=MagicMock(), + ) as frc_mock: + result = await planner.invoke(kernel_mock, question, arguments) + + assert isinstance(result, FunctionCallingStepwisePlannerResult) + assert result.final_answer == "" + assert result.iterations == options.max_iterations + + chat_completion_mock._process_function_call.assert_called_with( + function_call=other_function_call_content, + kernel=kernel_mock, + chat_history=planner._build_chat_history_for_step.return_value, + arguments=arguments, + function_call_count=1, + request_index=0, + function_call_behavior=chat_completion_mock.instantiate_prompt_execution_settings.return_value.function_choice_behavior, + ) + + frc_mock.assert_called_with( + function_call_content=other_function_call_content, + result=TextContent(text="An error occurred during planner invocation: Function call error"), + ) + + +@pytest.mark.asyncio +async def test_invoke_with_invalid_service_type(): + planner = FunctionCallingStepwisePlanner(service_id="test_service", options=None) + kernel_mock = AsyncMock(spec=Kernel) + kernel_mock.get_service.return_value = MagicMock() + with pytest.raises(PlannerInvalidConfigurationError): + await planner.invoke(kernel_mock, "test question") + + +@pytest.mark.asyncio +async def test_invoke_with_no_arguments(): + planner = FunctionCallingStepwisePlanner(service_id="test_service", options=None) + kernel_mock = AsyncMock(spec=Kernel) + question = "Test question" + + chat_completion_mock = AsyncMock(spec=OpenAIChatCompletion) + kernel_mock.get_service.return_value = chat_completion_mock + + planner._generate_plan = AsyncMock(return_value="generated plan") + planner._build_chat_history_for_step = AsyncMock(return_value=AsyncMock(spec=ChatHistory)) + chat_completion_mock.instantiate_prompt_execution_settings.return_value = AsyncMock() + + chat_completion_mock.get_chat_message_contents.return_value = [ + AsyncMock(items=[FunctionResultContent(text="Test result", id="test")]) + ] + + result = await planner.invoke(kernel_mock, question) + + assert isinstance(result, FunctionCallingStepwisePlannerResult) + assert result.final_answer == "" + assert result.iterations == planner.options.max_iterations + + +@patch("semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner.yaml.safe_load") +def test_create_config_from_yaml(mock_safe_load): + mock_safe_load.return_value = { + "template": "some template", + "execution_settings": {"default": {"setting1": "value1"}}, + } + + kernel_mock = MagicMock(spec=Kernel) + kernel_function_mock = MagicMock(spec=KernelFunction) + kernel_mock.add_function.return_value = kernel_function_mock + + planner = FunctionCallingStepwisePlanner(service_id="test_service") + + planner.generate_plan_yaml = "some_yaml_content" + + DEFAULT_SERVICE_NAME = "default" + + with patch("semantic_kernel.const.DEFAULT_SERVICE_NAME", DEFAULT_SERVICE_NAME): + result = planner._create_config_from_yaml(kernel_mock) + mock_safe_load.assert_called_once_with("some_yaml_content") + assert result == kernel_function_mock + + +def test_user_interaction(): + user_interaction = UserInteraction() + assert user_interaction.send_final_answer("answer") == "Thanks" diff --git a/python/tests/unit/planners/test_planner_extensions.py b/python/tests/unit/planners/test_planner_extensions.py new file mode 100644 index 0000000000..fc92addd37 --- /dev/null +++ b/python/tests/unit/planners/test_planner_extensions.py @@ -0,0 +1,154 @@ +# Copyright (c) Microsoft. All rights reserved. + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from semantic_kernel.functions.kernel_function_metadata import KernelFunctionMetadata +from semantic_kernel.functions.kernel_parameter_metadata import KernelParameterMetadata +from semantic_kernel.planners.planner_extensions import PlannerFunctionExtension, PlannerKernelExtension +from semantic_kernel.planners.planner_options import PlannerOptions + + +@pytest.fixture +def sample_function_metadata(): + return KernelFunctionMetadata( + name="function", + plugin_name="sample_plugin", + description="A sample function", + parameters=[ + KernelParameterMetadata(name="param1", description="Parameter 1", default_value=None), + KernelParameterMetadata(name="param2", description="Parameter 2", default_value="default"), + ], + is_prompt=False, + is_asynchronous=True, + ) + + +@pytest.mark.parametrize( + "function, expected_output", + [ + ( + KernelFunctionMetadata( + name="function", + plugin_name="sample_plugin", + description="A sample function", + parameters=[ + KernelParameterMetadata(name="param1", description="Parameter 1", default_value=None), + KernelParameterMetadata(name="param2", description="Parameter 2", default_value="default"), + ], + is_prompt=False, + is_asynchronous=True, + ), + "sample_plugin-function:\n description: A sample function\n inputs:\n - param1: Parameter 1\n - param2: Parameter 2 (default value: default)", # noqa: E501 + ) + ], +) +def test_to_manual_string(function, expected_output): + result = PlannerFunctionExtension.to_manual_string(function) + assert result == expected_output + + +@pytest.mark.parametrize( + "function, expected_output", + [ + ( + KernelFunctionMetadata( + name="function", + plugin_name="sample_plugin", + description="A sample function", + parameters=[ + KernelParameterMetadata(name="param1", description="Parameter 1", default_value=None), + KernelParameterMetadata(name="param2", description="Parameter 2", default_value="default"), + ], + is_prompt=False, + is_asynchronous=True, + ), + "function:\n description: A sample function\n inputs:\n - param1: Parameter 1\n - param2: Parameter 2", # noqa: E501 + ) + ], +) +def test_to_embedding_string(function, expected_output): + result = PlannerFunctionExtension.to_embedding_string(function) + assert result == expected_output + + +@pytest.mark.asyncio +async def test_get_functions_manual(): + kernel = MagicMock() + arguments = MagicMock() + options = PlannerOptions(get_available_functions=None) + + kernel.get_list_of_function_metadata = MagicMock( + return_value=[ + KernelFunctionMetadata( + name="function", + plugin_name="sample_plugin", + description="A sample function", + parameters=[ + KernelParameterMetadata(name="param1", description="Parameter 1", default_value=None), + KernelParameterMetadata(name="param2", description="Parameter 2", default_value="default"), + ], + is_prompt=False, + is_asynchronous=True, + ) + ] + ) + + result = await PlannerKernelExtension.get_functions_manual(kernel, arguments, options) + expected_output = "sample_plugin-function:\n description: A sample function\n inputs:\n - param1: Parameter 1\n - param2: Parameter 2 (default value: default)" # noqa: E501 + assert result == expected_output + + +@pytest.mark.asyncio +async def test_get_functions_manual_with_custom_get_available_functions(): + kernel = MagicMock() + arguments = MagicMock() + options = PlannerOptions(get_available_functions=AsyncMock()) + + # Mock get_available_functions method + options.get_available_functions.return_value = [ + KernelFunctionMetadata( + name="function", + plugin_name="sample_plugin", + description="A sample function", + parameters=[ + KernelParameterMetadata(name="param1", description="Parameter 1", default_value=None), + KernelParameterMetadata(name="param2", description="Parameter 2", default_value="default"), + ], + is_prompt=False, + is_asynchronous=True, + ) + ] + + result = await PlannerKernelExtension.get_functions_manual(kernel, arguments, options) + expected_output = "sample_plugin-function:\n description: A sample function\n inputs:\n - param1: Parameter 1\n - param2: Parameter 2 (default value: default)" # noqa: E501 + assert result == expected_output + + +@pytest.mark.asyncio +async def test_get_available_functions(): + kernel = MagicMock() + arguments = MagicMock() + options = PlannerOptions(excluded_plugins=[], excluded_functions=[]) + + # Mock get_list_of_function_metadata method + kernel.get_list_of_function_metadata = MagicMock( + return_value=[ + KernelFunctionMetadata( + name="function", + plugin_name="sample_plugin", + description="A sample function", + parameters=[ + KernelParameterMetadata(name="param1", description="Parameter 1", default_value=None), + KernelParameterMetadata(name="param2", description="Parameter 2", default_value="default"), + ], + is_prompt=False, + is_asynchronous=True, + ) + ] + ) + + result = await PlannerKernelExtension.get_available_functions(kernel, arguments, options) + assert len(result) == 1 + assert result[0].name == "function"