mirror of
https://github.com/microsoft/semantic-kernel.git
synced 2026-10-12 02:19:06 +00:00
Python: Introduce the function calling stepwise planner (#5350)
### Motivation and Context To further align with the SK dotnet SDK and its functionality, the Python SDK needs a function calling stepwise planner. Now that we have auto tool calling available, it makes sense to introduce this planner. <!-- 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: - Introduces the function calling stepwise planner, its planner options, and planner result. Closes #4642 - Introduces a kernel example `function_calling_stepwise_planner.py` to exercise the planner and show how it works. - Adds unit tests and an integration test. <!-- 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: --> - [ ] The code builds clean without any errors or warnings - [ ] 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 - [ ] All unit tests pass, and I have added new tests where possible - [ ] I didn't break anyone 😄
This commit is contained in:
Evan Mattson
@@ -146,7 +146,7 @@
|
||||
")\n",
|
||||
"\n",
|
||||
"summarize = kernel.create_function_from_prompt(\n",
|
||||
" function_name=\"summarizeFunc\", \n",
|
||||
" function_name=\"summarizeFunc\",\n",
|
||||
" plugin_name=\"summarizePlugin\",\n",
|
||||
" prompt_template_config=prompt_template_config,\n",
|
||||
")"
|
||||
|
||||
@@ -349,18 +349,18 @@
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"github_files = {}\n",
|
||||
"github_files[\"https://github.com/microsoft/semantic-kernel/blob/main/README.md\"] = (\n",
|
||||
" \"README: Installation, getting started, and how to contribute\"\n",
|
||||
")\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/blob/main/README.md\"\n",
|
||||
"] = \"README: Installation, getting started, and how to contribute\"\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/blob/main/dotnet/notebooks/02-running-prompts-from-file.ipynb\"\n",
|
||||
"] = \"Jupyter notebook describing how to pass prompts from a file to a semantic plugin or function\"\n",
|
||||
"github_files[\"https://github.com/microsoft/semantic-kernel/blob/main/dotnet/notebooks/00-getting-started.ipynb\"] = (\n",
|
||||
" \"Jupyter notebook describing how to get started with the Semantic Kernel\"\n",
|
||||
")\n",
|
||||
"github_files[\"https://github.com/microsoft/semantic-kernel/tree/main/samples/plugins/ChatPlugin/ChatGPT\"] = (\n",
|
||||
" \"Sample demonstrating how to create a chat plugin interfacing with ChatGPT\"\n",
|
||||
")\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/blob/main/dotnet/notebooks/00-getting-started.ipynb\"\n",
|
||||
"] = \"Jupyter notebook describing how to get started with the Semantic Kernel\"\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/tree/main/samples/plugins/ChatPlugin/ChatGPT\"\n",
|
||||
"] = \"Sample demonstrating how to create a chat plugin interfacing with ChatGPT\"\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/blob/main/dotnet/src/SemanticKernel/Memory/Volatile/VolatileMemoryStore.cs\"\n",
|
||||
"] = \"C# class that defines a volatile embedding store\""
|
||||
|
||||
@@ -422,18 +422,18 @@
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"github_files = {}\n",
|
||||
"github_files[\"https://github.com/microsoft/semantic-kernel/blob/main/README.md\"] = (\n",
|
||||
" \"README: Installation, getting started, and how to contribute\"\n",
|
||||
")\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/blob/main/README.md\"\n",
|
||||
"] = \"README: Installation, getting started, and how to contribute\"\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/blob/main/dotnet/notebooks/02-running-prompts-from-file.ipynb\"\n",
|
||||
"] = \"Jupyter notebook describing how to pass prompts from a file to a semantic plugin or function\"\n",
|
||||
"github_files[\"https://github.com/microsoft/semantic-kernel/blob/main/dotnet/notebooks/00-getting-started.ipynb\"] = (\n",
|
||||
" \"Jupyter notebook describing how to get started with the Semantic Kernel\"\n",
|
||||
")\n",
|
||||
"github_files[\"https://github.com/microsoft/semantic-kernel/tree/main/samples/plugins/ChatPlugin/ChatGPT\"] = (\n",
|
||||
" \"Sample demonstrating how to create a chat plugin interfacing with ChatGPT\"\n",
|
||||
")\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/blob/main/dotnet/notebooks/00-getting-started.ipynb\"\n",
|
||||
"] = \"Jupyter notebook describing how to get started with the Semantic Kernel\"\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/tree/main/samples/plugins/ChatPlugin/ChatGPT\"\n",
|
||||
"] = \"Sample demonstrating how to create a chat plugin interfacing with ChatGPT\"\n",
|
||||
"github_files[\n",
|
||||
" \"https://github.com/microsoft/semantic-kernel/blob/main/dotnet/src/SemanticKernel/Memory/Volatile/VolatileMemoryStore.cs\"\n",
|
||||
"] = \"C# class that defines a volatile embedding store\""
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import sys
|
||||
|
||||
if sys.version_info >= (3, 9):
|
||||
from typing import Annotated
|
||||
else:
|
||||
from typing_extensions import Annotated
|
||||
|
||||
import asyncio
|
||||
|
||||
import semantic_kernel as sk
|
||||
from semantic_kernel.connectors.ai.open_ai import (
|
||||
OpenAIChatCompletion,
|
||||
)
|
||||
from semantic_kernel.core_plugins.math_plugin import MathPlugin
|
||||
from semantic_kernel.core_plugins.time_plugin import TimePlugin
|
||||
from semantic_kernel.functions.kernel_function_decorator import kernel_function
|
||||
from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner import (
|
||||
FunctionCallingStepwisePlanner,
|
||||
)
|
||||
from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner_options import (
|
||||
FunctionCallingStepwisePlannerOptions,
|
||||
)
|
||||
|
||||
|
||||
# Define an example EmailPlugin
|
||||
class EmailPlugin:
|
||||
"""
|
||||
Description: EmailPlugin provides a set of functions to send emails.
|
||||
|
||||
Usage:
|
||||
kernel.import_plugin_from_object(EmailPlugin(), plugin_name="email")
|
||||
|
||||
Examples:
|
||||
{{email.SendEmail}} => Sends an email with the provided subject and body.
|
||||
"""
|
||||
|
||||
@kernel_function(name="SendEmail", description="Given an e-mail and message body, send an e-email")
|
||||
def send_email(
|
||||
self,
|
||||
subject: Annotated[str, "the subject of the email"],
|
||||
body: Annotated[str, "the body of the email"],
|
||||
) -> Annotated[str, "the output is a string"]:
|
||||
"""Sends an email with the provided subject and body."""
|
||||
return f"Email sent with subject: {subject} and body: {body}"
|
||||
|
||||
@kernel_function(name="GetEmailAddress", description="Given a name, find the email address")
|
||||
def get_email_address(
|
||||
self,
|
||||
input: Annotated[str, "the name of the person"],
|
||||
):
|
||||
email = ""
|
||||
if input == "Jane":
|
||||
email = "[email protected]"
|
||||
elif input == "Paul":
|
||||
email = "[email protected]"
|
||||
elif input == "Mary":
|
||||
email = "[email protected]"
|
||||
else:
|
||||
input = "[email protected]"
|
||||
return email
|
||||
|
||||
|
||||
async def main():
|
||||
kernel = sk.Kernel()
|
||||
|
||||
service_id = "planner"
|
||||
api_key, _ = sk.openai_settings_from_dot_env()
|
||||
kernel.add_service(
|
||||
OpenAIChatCompletion(
|
||||
service_id=service_id,
|
||||
ai_model_id="gpt-3.5-turbo-1106",
|
||||
api_key=api_key,
|
||||
),
|
||||
)
|
||||
|
||||
kernel.import_plugin_from_object(MathPlugin(), "MathPlugin")
|
||||
kernel.import_plugin_from_object(TimePlugin(), "TimePlugin")
|
||||
kernel.import_plugin_from_object(EmailPlugin(), "EmailPlugin")
|
||||
|
||||
questions = [
|
||||
"What is the current hour number, plus 5?",
|
||||
"What is 387 minus 22? Email the solution to John and Mary.",
|
||||
"Write a limerick, translate it to Spanish, and send it to Jane",
|
||||
]
|
||||
|
||||
options = FunctionCallingStepwisePlannerOptions(
|
||||
max_iterations=10,
|
||||
max_tokens=4000,
|
||||
)
|
||||
|
||||
planner = FunctionCallingStepwisePlanner(service_id=service_id, options=options)
|
||||
|
||||
for question in questions:
|
||||
result = await planner.invoke(kernel, question)
|
||||
print(f"Q: {question}\nA: {result.final_answer}\n")
|
||||
|
||||
# Uncomment the following line to view the planner's process for completing the request
|
||||
# print(f"Chat history: {result.chat_history}\n")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
+9
-3
@@ -336,9 +336,15 @@ class OpenAIChatCompletionBase(OpenAIHandler, ChatCompletionClientBase):
|
||||
if tool_call.function is None:
|
||||
continue
|
||||
logger.info(f"Calling {tool_call.function.name} function with args: {tool_call.function.arguments}")
|
||||
func_result = await kernel.invoke(
|
||||
**tool_call.function.split_name_dict(), arguments=tool_call.function.to_kernel_arguments()
|
||||
)
|
||||
try:
|
||||
func_result = await kernel.invoke(
|
||||
**tool_call.function.split_name_dict(), arguments=tool_call.function.to_kernel_arguments()
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.exception(f"Error occurred while invoking function {tool_call.function.name}")
|
||||
raise ServiceInvalidResponseError(
|
||||
f"Error occurred while invoking function {tool_call.function.name}"
|
||||
) from exc
|
||||
msg = OpenAIChatMessageContent(
|
||||
role=ChatRole.TOOL,
|
||||
content=str(func_result),
|
||||
|
||||
@@ -4,7 +4,6 @@ import logging
|
||||
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, TypeVar, Union
|
||||
|
||||
from semantic_kernel.exceptions import (
|
||||
FunctionInvalidNameError,
|
||||
PluginInitializationError,
|
||||
PluginInvalidNameError,
|
||||
)
|
||||
@@ -120,9 +119,7 @@ class KernelPluginCollection(KernelBaseModel):
|
||||
|
||||
for func in functions:
|
||||
if func.name in self.plugins[plugin_name].functions:
|
||||
raise FunctionInvalidNameError(
|
||||
f"Function with name '{func.name}' already exists in plugin '{plugin_name}'"
|
||||
)
|
||||
logger.warning(f'Overwriting function "{func.name}" in collection')
|
||||
self.plugins[plugin_name].functions[func.name] = func
|
||||
|
||||
def add_list_of_plugins(self, plugins: List["KernelPlugin"]) -> None:
|
||||
|
||||
@@ -2,7 +2,17 @@
|
||||
|
||||
from semantic_kernel.planners.action_planner.action_planner import ActionPlanner
|
||||
from semantic_kernel.planners.basic_planner import BasicPlanner
|
||||
from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner import (
|
||||
FunctionCallingStepwisePlanner,
|
||||
)
|
||||
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,
|
||||
)
|
||||
from semantic_kernel.planners.plan import Plan
|
||||
from semantic_kernel.planners.planner_options import PlannerOptions
|
||||
from semantic_kernel.planners.sequential_planner import SequentialPlanner
|
||||
from semantic_kernel.planners.stepwise_planner import StepwisePlanner
|
||||
|
||||
@@ -12,4 +22,8 @@ __all__ = [
|
||||
"SequentialPlanner",
|
||||
"StepwisePlanner",
|
||||
"ActionPlanner",
|
||||
"PlannerOptions",
|
||||
"FunctionCallingStepwisePlannerOptions",
|
||||
"FunctionCallingStepwisePlanner",
|
||||
"FunctionCallingStepwisePlannerResult",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner import (
|
||||
FunctionCallingStepwisePlanner,
|
||||
)
|
||||
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,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"FunctionCallingStepwisePlanner",
|
||||
"FunctionCallingStepwisePlannerOptions",
|
||||
"FunctionCallingStepwisePlannerResult",
|
||||
]
|
||||
+224
@@ -0,0 +1,224 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
from copy import copy
|
||||
from typing import Optional
|
||||
|
||||
import yaml
|
||||
|
||||
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import (
|
||||
OpenAIChatPromptExecutionSettings,
|
||||
)
|
||||
from semantic_kernel.connectors.ai.open_ai.services.open_ai_chat_completion import OpenAIChatCompletion
|
||||
from semantic_kernel.connectors.ai.open_ai.utils import (
|
||||
get_tool_call_object,
|
||||
)
|
||||
from semantic_kernel.contents.chat_history import ChatHistory
|
||||
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.kernel import Kernel
|
||||
from semantic_kernel.kernel_pydantic import KernelBaseModel
|
||||
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,
|
||||
)
|
||||
from semantic_kernel.planners.planner_extensions import PlannerKernelExtension
|
||||
from semantic_kernel.prompt_template.kernel_prompt_template import KernelPromptTemplate
|
||||
from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
|
||||
|
||||
CUR_DIR = os.path.dirname(os.path.realpath(__file__))
|
||||
PLAN_YAML_FILE_PATH = os.path.join(CUR_DIR, "generate_plan.yaml")
|
||||
STEP_PROMPT_FILE_PATH = os.path.join(CUR_DIR, "step_prompt.txt")
|
||||
|
||||
STEPWISE_PLANNER_PLUGIN_NAME = "StepwisePlanner_Excluded"
|
||||
|
||||
STEPWISE_USER_MESSAGE = (
|
||||
"Perform the next step of the plan if there is more work to do."
|
||||
"When you have reached a final answer, use the UserInteraction-SendFinalAnswer"
|
||||
"function to communicate this back to the user."
|
||||
)
|
||||
|
||||
USER_INTERACTION_SEND_FINAL_ANSWER = "UserInteraction-SendFinalAnswer"
|
||||
|
||||
logger: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FunctionCallingStepwisePlanner(KernelBaseModel):
|
||||
service_id: str
|
||||
options: FunctionCallingStepwisePlannerOptions
|
||||
generate_plan_yaml: str
|
||||
step_prompt: str
|
||||
|
||||
def __init__(self, service_id: str, options: Optional[FunctionCallingStepwisePlannerOptions] = None):
|
||||
"""Initialize a new instance of the FunctionCallingStepwisePlanner
|
||||
|
||||
The FunctionCallingStepwisePlanner is a planner based on top of an OpenAI Chat Completion service
|
||||
(whether it be AzureOpenAI or OpenAI), so that we can use tools.
|
||||
|
||||
If the options are configured to use callbacks to get the initial plan and the step prompt,
|
||||
the planner will use those provided callbacks to get that information. Otherwise it will
|
||||
read from the default yaml plan file and the step prompt file.
|
||||
|
||||
Args:
|
||||
service_id (str): The service id
|
||||
options (Optional[FunctionCallingStepwisePlannerOptions], optional): The options for
|
||||
the function calling stepwise planner. Defaults to None.
|
||||
"""
|
||||
options = options or FunctionCallingStepwisePlannerOptions()
|
||||
generate_plan_yaml = (
|
||||
options.get_initial_plan() if options.get_initial_plan else open(PLAN_YAML_FILE_PATH).read()
|
||||
)
|
||||
step_prompt = options.get_step_prompt() if options.get_step_prompt else open(STEP_PROMPT_FILE_PATH).read()
|
||||
options.excluded_plugins.add(STEPWISE_PLANNER_PLUGIN_NAME)
|
||||
|
||||
super().__init__(
|
||||
service_id=service_id,
|
||||
options=options,
|
||||
generate_plan_yaml=generate_plan_yaml,
|
||||
step_prompt=step_prompt,
|
||||
)
|
||||
|
||||
async def invoke(
|
||||
self,
|
||||
kernel: Kernel,
|
||||
question: str,
|
||||
) -> FunctionCallingStepwisePlannerResult:
|
||||
"""
|
||||
Execute the function calling stepwise planner
|
||||
|
||||
Args:
|
||||
kernel: The kernel instance
|
||||
question: The input question
|
||||
|
||||
Returns:
|
||||
FunctionCallingStepwisePlannerResult: The result of the function calling stepwise planner
|
||||
|
||||
Raises:
|
||||
PlannerInvalidConfigurationError: If the input question is empty
|
||||
"""
|
||||
if not question:
|
||||
raise PlannerInvalidConfigurationError("Input question cannot be empty")
|
||||
|
||||
try:
|
||||
chat_completion = kernel.get_service(service_id=self.service_id, type=OpenAIChatCompletion)
|
||||
except Exception as exc:
|
||||
raise PlannerInvalidConfigurationError(
|
||||
f"The OpenAI service `{self.service_id}` is not available. Please configure the AI service."
|
||||
) from exc
|
||||
|
||||
prompt_execution_settings: (
|
||||
OpenAIChatPromptExecutionSettings
|
||||
) = self.options.execution_settings or chat_completion.get_prompt_execution_settings_class()(
|
||||
service_id=self.service_id
|
||||
)
|
||||
|
||||
# Clone the kernel so that we can add planner-specific plugins without affecting the original kernel instance
|
||||
cloned_kernel = copy(kernel)
|
||||
cloned_kernel.import_plugin_from_object(UserInteraction(), "UserInteraction")
|
||||
|
||||
# Create and invoke a kernel function to generate the initial plan
|
||||
initial_plan = await self._generate_plan(question, cloned_kernel)
|
||||
|
||||
chat_history_for_steps = await self._build_chat_history_for_step(question, initial_plan, cloned_kernel)
|
||||
prompt_execution_settings.tool_choice = "auto"
|
||||
prompt_execution_settings.tools = get_tool_call_object(kernel, {"exclude_plugin": [self.service_id]})
|
||||
for i in range(self.options.max_iterations):
|
||||
# sleep for a bit to avoid rate limiting
|
||||
if i > 0:
|
||||
await asyncio.sleep(self.options.min_iteration_time_ms / 1000.0) # convert ms to sec
|
||||
# For each step, request another completion to select a function for that step
|
||||
chat_history_for_steps.add_user_message(STEPWISE_USER_MESSAGE)
|
||||
chat_result = await chat_completion.complete_chat(
|
||||
chat_history=chat_history_for_steps,
|
||||
settings=prompt_execution_settings,
|
||||
kernel=cloned_kernel,
|
||||
)
|
||||
chat_result = chat_result[0]
|
||||
chat_history_for_steps.add_message(chat_result)
|
||||
|
||||
if not chat_result.tool_calls:
|
||||
chat_history_for_steps.add_user_message("That function call is invalid. Try something else!")
|
||||
continue
|
||||
|
||||
# Try to get the final answer out
|
||||
if chat_result.tool_calls[0].function.name == USER_INTERACTION_SEND_FINAL_ANSWER:
|
||||
args = chat_result.tool_calls[0].function.parse_arguments()
|
||||
answer = args["answer"]
|
||||
return FunctionCallingStepwisePlannerResult(
|
||||
final_answer=answer,
|
||||
chat_history=chat_history_for_steps,
|
||||
iterations=i + 1,
|
||||
)
|
||||
|
||||
try:
|
||||
await chat_completion._process_tool_calls(chat_result, cloned_kernel, chat_history_for_steps)
|
||||
except Exception as exc:
|
||||
chat_history_for_steps.add_user_message(f"An error occurred during planner invocation: {exc}")
|
||||
continue
|
||||
|
||||
# We're done, but the model hasn't returned a final answer.
|
||||
return FunctionCallingStepwisePlannerResult(
|
||||
final_answer="",
|
||||
chat_history=chat_history_for_steps,
|
||||
iterations=i + 1,
|
||||
)
|
||||
|
||||
async def _build_chat_history_for_step(
|
||||
self,
|
||||
goal: str,
|
||||
initial_plan: str,
|
||||
kernel: Kernel,
|
||||
) -> ChatHistory:
|
||||
"""Build the chat history for the stepwise planner"""
|
||||
chat_history = ChatHistory()
|
||||
arguments = KernelArguments(
|
||||
goal=goal,
|
||||
initial_plan=initial_plan,
|
||||
)
|
||||
kernel_prompt_template = KernelPromptTemplate(
|
||||
prompt_template_config=PromptTemplateConfig(
|
||||
template=self.step_prompt,
|
||||
)
|
||||
)
|
||||
system_message = await kernel_prompt_template.render(kernel, arguments)
|
||||
chat_history.add_system_message(system_message)
|
||||
return chat_history
|
||||
|
||||
def _create_config_from_yaml(self, kernel: Kernel) -> "KernelFunction":
|
||||
"""A temporary method to create a function from the yaml file.
|
||||
The yaml.safe_load will be replaced with the proper kernel
|
||||
method later."""
|
||||
data = yaml.safe_load(self.generate_plan_yaml)
|
||||
prompt_template_config = PromptTemplateConfig(**data)
|
||||
if "default" in prompt_template_config.execution_settings:
|
||||
settings = prompt_template_config.execution_settings.pop("default")
|
||||
prompt_template_config.execution_settings[self.service_id] = settings
|
||||
return kernel.create_function_from_prompt(
|
||||
function_name="create_plan",
|
||||
plugin_name="sequential_planner",
|
||||
description="Create a plan for the given goal",
|
||||
prompt_template_config=prompt_template_config,
|
||||
)
|
||||
|
||||
async def _generate_plan(
|
||||
self,
|
||||
question: str,
|
||||
kernel: Kernel,
|
||||
arguments: KernelArguments = None,
|
||||
) -> str:
|
||||
"""Generate the plan for the given question using the kernel"""
|
||||
generate_plan_function = self._create_config_from_yaml(kernel)
|
||||
functions_manual = await PlannerKernelExtension.get_functions_manual(kernel, arguments)
|
||||
generated_plan_args = KernelArguments(
|
||||
name_delimiter="-",
|
||||
available_functions=functions_manual,
|
||||
goal=question,
|
||||
)
|
||||
generate_plan_result = await kernel.invoke(generate_plan_function, generated_plan_args)
|
||||
return str(generate_plan_result)
|
||||
+42
@@ -0,0 +1,42 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
from pydantic import model_validator
|
||||
|
||||
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import (
|
||||
OpenAIPromptExecutionSettings,
|
||||
)
|
||||
from semantic_kernel.planners.planner_options import PlannerOptions
|
||||
|
||||
|
||||
class FunctionCallingStepwisePlannerOptions(PlannerOptions):
|
||||
"""The Function Calling Stepwise Planner Options."""
|
||||
|
||||
max_tokens: Optional[int] = None
|
||||
max_tokens_ratio: Optional[float] = 0.1
|
||||
max_completion_tokens: Optional[int] = None
|
||||
max_prompt_tokens: Optional[int] = None
|
||||
get_initial_plan: Optional[Callable[[], str]] = None
|
||||
get_step_prompt: Optional[Callable[[], str]] = None
|
||||
max_iterations: Optional[int] = 15
|
||||
min_iteration_time_ms: Optional[int] = 100
|
||||
execution_settings: Optional[OpenAIPromptExecutionSettings] = None
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def calculate_token_limits(cls, data: Any) -> Any:
|
||||
if isinstance(data, dict):
|
||||
max_tokens = data.get("max_tokens")
|
||||
# Ensure max_tokens_ratio has a default value if not provided
|
||||
max_tokens_ratio = data.get("max_tokens_ratio", 0.1)
|
||||
|
||||
if max_tokens is not None:
|
||||
data["max_completion_tokens"] = int(max_tokens * max_tokens_ratio)
|
||||
data["max_prompt_tokens"] = int(max_tokens * (1 - max_tokens_ratio))
|
||||
else:
|
||||
# Explicitly setting them to None if max_tokens is None, for clarity
|
||||
data["max_completion_tokens"] = None
|
||||
data["max_prompt_tokens"] = None
|
||||
|
||||
return data
|
||||
+28
@@ -0,0 +1,28 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import sys
|
||||
|
||||
if sys.version_info >= (3, 9):
|
||||
from typing import Annotated
|
||||
else:
|
||||
from typing_extensions import Annotated
|
||||
|
||||
from semantic_kernel.contents.chat_history import ChatHistory
|
||||
from semantic_kernel.functions.kernel_function_decorator import kernel_function
|
||||
from semantic_kernel.kernel_pydantic import KernelBaseModel
|
||||
|
||||
|
||||
class FunctionCallingStepwisePlannerResult(KernelBaseModel):
|
||||
"""The result of the function calling stepwise planner"""
|
||||
|
||||
final_answer: str = ""
|
||||
chat_history: ChatHistory = None
|
||||
iterations: int = 0
|
||||
|
||||
|
||||
class UserInteraction:
|
||||
"""The Kernel Function used to interact with the user"""
|
||||
|
||||
@kernel_function(description="The final answer to return to the user", name="SendFinalAnswer")
|
||||
def send_final_answer(self, answer: Annotated[str, "The final answer"]) -> str:
|
||||
return "Thanks"
|
||||
@@ -0,0 +1,32 @@
|
||||
template_format: semantic-kernel
|
||||
template: |
|
||||
<message role="system">
|
||||
You are an expert at generating plans from a given GOAL. Think step by step and determine a plan to satisfy the specified GOAL using only the FUNCTIONS provided to you. You can also make use of your own knowledge while forming an answer but you must not use functions that are not provided. Once you have come to a final answer, use the UserInteraction{{$name_delimiter}}SendFinalAnswer function to communicate this back to the user.
|
||||
|
||||
[FUNCTIONS]
|
||||
|
||||
{{$available_functions}}
|
||||
|
||||
[END FUNCTIONS]
|
||||
|
||||
To create the plan, follow these steps:
|
||||
0. Each step should be something that is capable of being done by the list of available functions.
|
||||
1. Steps can use output from one or more previous steps as input, if appropriate.
|
||||
2. The plan should be as short as possible.
|
||||
</message>
|
||||
<message role="user">{{$goal}}</message>
|
||||
description: Generate a step-by-step plan to satisfy a given goal
|
||||
name: GeneratePlan
|
||||
input_variables:
|
||||
- name: available_functions
|
||||
description: A list of functions that can be used to generate the plan
|
||||
- name: goal
|
||||
description: The goal to satisfy
|
||||
execution_settings:
|
||||
default:
|
||||
temperature: 0.0
|
||||
top_p: 0.0
|
||||
presence_penalty: 0.0
|
||||
frequency_penalty: 0.0
|
||||
max_tokens: 256
|
||||
stop_sequences: []
|
||||
@@ -0,0 +1,6 @@
|
||||
Original request: {{$goal}}
|
||||
|
||||
You are in the process of helping the user fulfill this request using the following plan:
|
||||
{{$initial_plan}}
|
||||
|
||||
The user will ask you for help with each step.
|
||||
@@ -0,0 +1,77 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import logging
|
||||
|
||||
from semantic_kernel.functions.kernel_arguments import KernelArguments
|
||||
from semantic_kernel.functions.kernel_function_metadata import KernelFunctionMetadata
|
||||
from semantic_kernel.kernel import Kernel
|
||||
from semantic_kernel.planners.planner_options import PlannerOptions
|
||||
|
||||
logger: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
# NOTE these extensions will be used going forward by the FunctionCallingStepWisePlanner and others.
|
||||
# The other sequential planner extensions will be left in right now for as long as we support that planner
|
||||
# and will be removed once we deprecate it.
|
||||
|
||||
|
||||
class PlannerFunctionExtension:
|
||||
@staticmethod
|
||||
def to_manual_string(function: KernelFunctionMetadata):
|
||||
inputs = [
|
||||
f" - {parameter.name}: {parameter.description}"
|
||||
+ (f" (default value: {parameter.default_value})" if parameter.default_value else "")
|
||||
for parameter in function.parameters
|
||||
]
|
||||
|
||||
inputs = "\n".join(inputs)
|
||||
qualified_name = PlannerFunctionExtension.to_fully_qualified_name(function)
|
||||
|
||||
return f"{qualified_name}:\n description: {function.description}\n inputs:\n " f" {inputs}"
|
||||
|
||||
@staticmethod
|
||||
def to_fully_qualified_name(function: KernelFunctionMetadata):
|
||||
return f"{function.plugin_name}-{function.name}"
|
||||
|
||||
@staticmethod
|
||||
def to_embedding_string(function: KernelFunctionMetadata):
|
||||
inputs = "\n".join([f" - {parameter.name}: {parameter.description}" for parameter in function.parameters])
|
||||
return f"{function.name}:\n description: {function.description}\n " f" inputs:\n{inputs}"
|
||||
|
||||
|
||||
class PlannerKernelExtension:
|
||||
PLANNER_MEMORY_COLLECTION_NAME = " Planning.KernelFunctionManual"
|
||||
PLAN_KERNEL_FUNCTIONS_ARE_REMEMBERED = "Planning.KernelFunctionsAreRemembered"
|
||||
|
||||
@staticmethod
|
||||
async def get_functions_manual(
|
||||
kernel: "Kernel",
|
||||
arguments: KernelArguments,
|
||||
options: PlannerOptions = None,
|
||||
) -> str:
|
||||
options = options or PlannerOptions()
|
||||
|
||||
if options.get_available_functions is None:
|
||||
functions = await PlannerKernelExtension.get_available_functions(kernel, arguments, options)
|
||||
else:
|
||||
functions = await options.get_available_functions(options)
|
||||
|
||||
return "\n\n".join([PlannerFunctionExtension.to_manual_string(func) for func in functions])
|
||||
|
||||
@staticmethod
|
||||
async def get_available_functions(
|
||||
kernel: Kernel,
|
||||
arguments: KernelArguments,
|
||||
options: PlannerOptions,
|
||||
):
|
||||
excluded_plugins = options.excluded_plugins or []
|
||||
excluded_functions = options.excluded_functions or []
|
||||
|
||||
available_functions = [
|
||||
func
|
||||
for func in kernel.plugins.get_list_of_function_metadata()
|
||||
if (func.plugin_name not in excluded_plugins and func.name not in excluded_functions)
|
||||
]
|
||||
|
||||
# TODO support for semantic memory query
|
||||
|
||||
return available_functions
|
||||
@@ -0,0 +1,17 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Callable, List, Optional, Set
|
||||
|
||||
from semantic_kernel.functions.kernel_function_metadata import KernelFunctionMetadata
|
||||
from semantic_kernel.kernel_pydantic import KernelBaseModel
|
||||
|
||||
|
||||
class PlannerOptions(KernelBaseModel):
|
||||
"""The default planner options that planners inherit from"""
|
||||
|
||||
excluded_plugins: Set[str] = set()
|
||||
excluded_functions: Set[str] = set()
|
||||
get_available_functions: Optional[Callable[["PlannerOptions", Optional[str]], List[KernelFunctionMetadata]]] = None
|
||||
# TODO semantic_memory_config
|
||||
@@ -14,6 +14,7 @@ from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.azure_chat_
|
||||
ExtraBody,
|
||||
)
|
||||
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
|
||||
from semantic_kernel.contents.chat_history import ChatHistory
|
||||
from semantic_kernel.functions.kernel_arguments import KernelArguments
|
||||
from semantic_kernel.kernel import Kernel
|
||||
from semantic_kernel.memory.memory_record import MemoryRecord
|
||||
@@ -119,7 +120,7 @@ async def create_with_data_chat_function(get_aoai_config, kernel: Kernel, create
|
||||
)
|
||||
kernel.add_service(chat_service)
|
||||
|
||||
prompt = "{{$input}}"
|
||||
prompt = "{{$chat_history}}{{$input}}"
|
||||
|
||||
exec_settings = PromptExecutionSettings(
|
||||
service_id="chat-gpt-extensions",
|
||||
@@ -155,7 +156,9 @@ async def test_azure_e2e_chat_completion_with_extensions(
|
||||
memory_store,
|
||||
) = await create_with_data_chat_function
|
||||
|
||||
arguments = KernelArguments(input="who are Emily and David?")
|
||||
chat_history = ChatHistory()
|
||||
chat_history.add_user_message("A story about Emily and David...")
|
||||
arguments = KernelArguments(input="who are Emily and David?", chat_history=chat_history)
|
||||
|
||||
# TODO: get streaming working for this test
|
||||
use_streaming = False
|
||||
|
||||
+51
@@ -0,0 +1,51 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import pytest
|
||||
|
||||
from semantic_kernel.connectors.ai.open_ai import (
|
||||
OpenAIChatCompletion,
|
||||
)
|
||||
from semantic_kernel.core_plugins.math_plugin import MathPlugin
|
||||
from semantic_kernel.kernel import Kernel
|
||||
from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner import (
|
||||
FunctionCallingStepwisePlanner,
|
||||
)
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_can_execute_function_calling_stepwise_plan(kernel: Kernel, get_oai_config):
|
||||
api_key, _ = get_oai_config
|
||||
|
||||
service_id = "planner"
|
||||
kernel.add_service(
|
||||
OpenAIChatCompletion(
|
||||
service_id=service_id,
|
||||
ai_model_id="gpt-3.5-turbo-1106",
|
||||
api_key=api_key,
|
||||
),
|
||||
)
|
||||
|
||||
kernel.import_plugin_from_object(MathPlugin(), "MathPlugin")
|
||||
|
||||
questions = [
|
||||
"What is the current hour number, plus 5?",
|
||||
]
|
||||
|
||||
options = FunctionCallingStepwisePlannerOptions(
|
||||
max_iterations=5,
|
||||
max_tokens=1000,
|
||||
)
|
||||
|
||||
planner = FunctionCallingStepwisePlanner(service_id=service_id, options=options)
|
||||
|
||||
for question in questions:
|
||||
result = await planner.invoke(kernel, question)
|
||||
print(f"Q: {question}\nA: {result.final_answer}\n")
|
||||
assert isinstance(result, FunctionCallingStepwisePlannerResult)
|
||||
assert 0 < len(result.final_answer) < 100
|
||||
@@ -20,7 +20,6 @@ from semantic_kernel.exceptions import (
|
||||
ServiceInvalidTypeError,
|
||||
)
|
||||
from semantic_kernel.exceptions.function_exceptions import (
|
||||
FunctionInvalidNameError,
|
||||
FunctionNameNotUniqueError,
|
||||
PluginInitializationError,
|
||||
PluginInvalidNameError,
|
||||
@@ -496,13 +495,6 @@ def test_register_with_none_plugin_name(kernel: Kernel, decorated_native_functio
|
||||
kernel.register_function_from_method(method=decorated_native_function, plugin_name=None)
|
||||
|
||||
|
||||
def test_register_overloaded_native_function(kernel: Kernel, decorated_native_function):
|
||||
kernel.register_function_from_method("TestPlugin", decorated_native_function)
|
||||
|
||||
with pytest.raises(FunctionInvalidNameError):
|
||||
kernel.register_function_from_method("TestPlugin", decorated_native_function)
|
||||
|
||||
|
||||
# endregion
|
||||
# region Services
|
||||
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
|
||||
from typing import Callable
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from semantic_kernel import Kernel
|
||||
from semantic_kernel.exceptions import FunctionInitializationError
|
||||
from semantic_kernel.functions.kernel_arguments import KernelArguments
|
||||
from semantic_kernel.functions.kernel_function import KernelFunction
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_valid_native_function(kernel: Kernel, decorated_native_function: Callable):
|
||||
registered_func = kernel.register_function_from_method("TestPlugin", decorated_native_function)
|
||||
|
||||
assert isinstance(registered_func, KernelFunction)
|
||||
assert kernel.plugins["TestPlugin"]["getLightStatus"] == registered_func
|
||||
func_result = await registered_func.invoke(kernel, KernelArguments(arg1="testtest"))
|
||||
assert str(func_result) == "test"
|
||||
|
||||
|
||||
def test_register_undecorated_native_function(kernel: Kernel, not_decorated_native_function: Callable):
|
||||
with pytest.raises(FunctionInitializationError):
|
||||
kernel.register_function_from_method("TestPlugin", not_decorated_native_function)
|
||||
|
||||
|
||||
def test_register_with_none_plugin_name(kernel: Kernel, decorated_native_function: Callable):
|
||||
with pytest.raises(ValidationError):
|
||||
kernel.register_function_from_method(method=decorated_native_function, plugin_name=None)
|
||||
+48
@@ -0,0 +1,48 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import pytest
|
||||
|
||||
from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner_options import (
|
||||
FunctionCallingStepwisePlannerOptions,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"max_tokens, max_tokens_ratio, expected_max_completion, expected_max_prompt",
|
||||
[
|
||||
(100, 0.1, 10, 90),
|
||||
(200, 0.2, 40, 160),
|
||||
(None, 0.1, None, None),
|
||||
],
|
||||
)
|
||||
def test_calculate_token_limits(max_tokens, max_tokens_ratio, expected_max_completion, expected_max_prompt):
|
||||
options_data = {"max_tokens": max_tokens, "max_tokens_ratio": max_tokens_ratio}
|
||||
options = FunctionCallingStepwisePlannerOptions(**options_data)
|
||||
|
||||
assert (
|
||||
options.max_completion_tokens == expected_max_completion
|
||||
), "max_completion_tokens did not match expected value"
|
||||
assert options.max_prompt_tokens == expected_max_prompt, "max_prompt_tokens did not match expected value"
|
||||
|
||||
|
||||
def test_defaults():
|
||||
"""Test that defaults are applied correctly when no values are provided."""
|
||||
options = FunctionCallingStepwisePlannerOptions()
|
||||
assert options.max_tokens_ratio == 0.1, "Default max_tokens_ratio not applied"
|
||||
assert options.max_iterations == 15, "Default max_iterations not applied"
|
||||
assert options.min_iteration_time_ms == 100, "Default min_iteration_time_ms not applied"
|
||||
assert options.max_completion_tokens is None, "max_completion_tokens should be None when max_tokens is not set"
|
||||
assert options.max_prompt_tokens is None, "max_prompt_tokens should be None when max_tokens is not set"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"field",
|
||||
[
|
||||
("get_initial_plan"),
|
||||
("get_step_prompt"),
|
||||
("execution_settings"),
|
||||
],
|
||||
)
|
||||
def test_optional_fields_none_by_default(field):
|
||||
options = FunctionCallingStepwisePlannerOptions()
|
||||
assert getattr(options, field) is None, f"{field} should be None by default"
|
||||
+86
@@ -0,0 +1,86 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from semantic_kernel.exceptions.planner_exceptions import PlannerInvalidConfigurationError
|
||||
from semantic_kernel.functions.kernel_function import KernelFunction
|
||||
from semantic_kernel.kernel import Kernel
|
||||
from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner import (
|
||||
FunctionCallingStepwisePlanner,
|
||||
)
|
||||
from semantic_kernel.planners.function_calling_stepwise_planner.function_calling_stepwise_planner_options import (
|
||||
FunctionCallingStepwisePlannerOptions,
|
||||
)
|
||||
|
||||
|
||||
def test_initialization():
|
||||
planner = FunctionCallingStepwisePlanner(service_id="test_service", options=None)
|
||||
assert planner.service_id == "test_service"
|
||||
assert planner.options is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invoke_with_empty_question_raises_error():
|
||||
planner = FunctionCallingStepwisePlanner(service_id="test-service", options=None)
|
||||
kernel_mock = AsyncMock(Kernel)
|
||||
with pytest.raises(PlannerInvalidConfigurationError):
|
||||
await planner.invoke(kernel_mock, "")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_initial_plan_callback_usage():
|
||||
fake_get_initial_plan = MagicMock(return_value="custom initial plan")
|
||||
options = FunctionCallingStepwisePlannerOptions(get_initial_plan=fake_get_initial_plan)
|
||||
planner = FunctionCallingStepwisePlanner(service_id="test_service", options=options)
|
||||
assert planner.generate_plan_yaml == "custom initial plan"
|
||||
fake_get_initial_plan.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_step_prompt_callback_usage():
|
||||
fake_get_step_prompt = MagicMock(return_value="custom step prompt")
|
||||
options = FunctionCallingStepwisePlannerOptions(get_step_prompt=fake_get_step_prompt)
|
||||
planner = FunctionCallingStepwisePlanner(service_id="test_service", options=options)
|
||||
assert planner.step_prompt == "custom step prompt"
|
||||
fake_get_step_prompt.assert_called_once_with()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_chat_history_for_step():
|
||||
planner = FunctionCallingStepwisePlanner(service_id="test_service", options=None)
|
||||
kernel_mock = AsyncMock(Kernel)
|
||||
kernel_mock.get_service.return_value = AsyncMock()
|
||||
chat_history = await planner._build_chat_history_for_step("goal", "initial_plan", kernel_mock)
|
||||
assert chat_history is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_plan():
|
||||
planner = FunctionCallingStepwisePlanner(service_id="test_service")
|
||||
|
||||
kernel_mock = AsyncMock(Kernel)
|
||||
kernel_mock.get_service.return_value = AsyncMock()
|
||||
|
||||
with patch(
|
||||
"semantic_kernel.planners.function_calling_stepwise_planner.FunctionCallingStepwisePlanner._create_config_from_yaml",
|
||||
return_value=AsyncMock(spec=KernelFunction),
|
||||
) as mock_create_yaml_config, patch(
|
||||
"semantic_kernel.planners.planner_extensions.PlannerKernelExtension.get_functions_manual",
|
||||
new=AsyncMock(),
|
||||
) as mock_get_functions_manual:
|
||||
question = "Why is the sky blue?"
|
||||
result = await planner._generate_plan(question, kernel_mock)
|
||||
|
||||
mock_create_yaml_config.assert_called_once_with(kernel_mock)
|
||||
mock_get_functions_manual.assert_awaited_once()
|
||||
assert result is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invoke_with_no_configured_AI_service_raises_exception(kernel: Kernel):
|
||||
planner = FunctionCallingStepwisePlanner(service_id="test", options=None)
|
||||
with pytest.raises(PlannerInvalidConfigurationError):
|
||||
await planner.invoke(kernel, "test question")
|
||||
Reference in New Issue
Block a user