Python: add stepwise planner (#2223)

### Motivation and Context  implement stepwise planner #2221 ### Description  ### Contribution Checklist  - [ ] 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 😄 --------- Co-authored-by: Abby Harrison <[email protected]> Co-authored-by: Lee Miller <[email protected]> Co-authored-by: Abby Harrison <[email protected]>
2026-10-12 02:19:06 +00:00 · 2023-08-23 19:12:52 +00:00
parent 0a477b198c
commit 05afd29ef6
14 changed files with 893 additions and 22 deletions
@@ -16,6 +16,7 @@ from semantic_kernel.semantic_functions.semantic_function_config import (
 from semantic_kernel.utils.null_logger import NullLogger
 from semantic_kernel.utils.settings import (
    azure_openai_settings_from_dot_env,
+    bing_search_settings_from_dot_env,
    google_palm_settings_from_dot_env,
    openai_settings_from_dot_env,
    pinecone_settings_from_dot_env,
@@ -29,6 +30,7 @@ __all__ = [
    "azure_openai_settings_from_dot_env",
    "postgres_settings_from_dot_env",
    "pinecone_settings_from_dot_env",
+    "bing_search_settings_from_dot_env",
    "google_palm_settings_from_dot_env",
    "PromptTemplateConfig",
    "PromptTemplate",
@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.

 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Dict
+from typing import TYPE_CHECKING, Dict, List

 if TYPE_CHECKING:
    from semantic_kernel.semantic_functions.prompt_template_config import (
@@ -18,16 +18,19 @@ class ChatRequestSettings:
    number_of_responses: int = 1
    max_tokens: int = 256
    token_selection_biases: Dict[int, int] = field(default_factory=dict)
+    stop_sequences: List[str] = field(default_factory=list)

    def update_from_completion_config(
        self, completion_config: "PromptTemplateConfig.CompletionConfig"
    ):
        self.temperature = completion_config.temperature
        self.top_p = completion_config.top_p
+        self.number_of_responses = completion_config.number_of_responses
+        self.stop_sequences = completion_config.stop_sequences
+        self.max_tokens = completion_config.max_tokens
        self.presence_penalty = completion_config.presence_penalty
        self.frequency_penalty = completion_config.frequency_penalty
-        self.number_of_responses = completion_config.number_of_responses
-        self.max_tokens = completion_config.max_tokens
+        self.token_selection_biases = completion_config.token_selection_biases

    @staticmethod
    def from_completion_config(
@@ -101,15 +101,8 @@ class OpenAIChatCompletion(ChatCompletionClientBase, TextCompletionClientBase):
            str -- The completed text.
        """
        prompt_to_message = [("user", prompt)]
-        chat_settings = ChatRequestSettings(
-            temperature=request_settings.temperature,
-            top_p=request_settings.top_p,
-            presence_penalty=request_settings.presence_penalty,
-            frequency_penalty=request_settings.frequency_penalty,
-            max_tokens=request_settings.max_tokens,
-            number_of_responses=request_settings.number_of_responses,
-            token_selection_biases=request_settings.token_selection_biases,
-        )
+        chat_settings = ChatRequestSettings.from_completion_config(request_settings)
+
        response = await self._send_chat_request(
            prompt_to_message, chat_settings, False
        )
@@ -131,6 +124,7 @@ class OpenAIChatCompletion(ChatCompletionClientBase, TextCompletionClientBase):
            max_tokens=request_settings.max_tokens,
            number_of_responses=request_settings.number_of_responses,
            token_selection_biases=request_settings.token_selection_biases,
+            stop_sequences=request_settings.stop_sequences,
        )
        response = await self._send_chat_request(prompt_to_message, chat_settings, True)

@@ -205,11 +199,17 @@ class OpenAIChatCompletion(ChatCompletionClientBase, TextCompletionClientBase):
                messages=formatted_messages,
                temperature=request_settings.temperature,
                top_p=request_settings.top_p,
-                presence_penalty=request_settings.presence_penalty,
-                frequency_penalty=request_settings.frequency_penalty,
-                max_tokens=request_settings.max_tokens,
                n=request_settings.number_of_responses,
                stream=stream,
+                stop=(
+                    request_settings.stop_sequences
+                    if request_settings.stop_sequences is not None
+                    and len(request_settings.stop_sequences) > 0
+                    else None
+                ),
+                max_tokens=request_settings.max_tokens,
+                presence_penalty=request_settings.presence_penalty,
+                frequency_penalty=request_settings.frequency_penalty,
                logit_bias=(
                    request_settings.token_selection_biases
                    if request_settings.token_selection_biases is not None
@@ -4,10 +4,12 @@ from semantic_kernel.planning.action_planner.action_planner import ActionPlanner
 from semantic_kernel.planning.basic_planner import BasicPlanner
 from semantic_kernel.planning.plan import Plan
 from semantic_kernel.planning.sequential_planner import SequentialPlanner
+from semantic_kernel.planning.stepwise_planner import StepwisePlanner

 __all__ = [
-    "SequentialPlanner",
    "BasicPlanner",
    "Plan",
+    "SequentialPlanner",
+    "StepwisePlanner",
    "ActionPlanner",
 ]
@@ -0,0 +1,32 @@
+{
+    "schema": 1,
+    "description": "Given a request or command or goal generate multi-step plan to reach the goal. After each step LLM is called to perform the reasoning for the next step.",
+    "type": "completion",
+    "completion": {
+      "max_tokens": 1024,
+      "temperature": 0,
+      "top_p": 0,
+      "presence_penalty": 0,
+      "frequency_penalty": 0,
+      "stop_sequences": ["[OBSERVATION]", "\n[THOUGHT]"]
+    },
+    "input": {
+      "parameters": [
+        {
+          "name": "question",
+          "description": "The question to answer",
+          "defaultValue": ""
+        },
+        {
+          "name": "agentScratchPad",
+          "description": "The agent's scratch pad",
+          "defaultValue": ""
+        },
+        {
+          "name": "functionDescriptions",
+          "description": "The manual of the agent's functions",
+          "defaultValue": ""
+        }
+      ]
+    }
+  }
@@ -0,0 +1,53 @@
+[INSTRUCTION]
+Answer the following questions as accurately as possible using the provided functions.
+
+[AVAILABLE FUNCTIONS]
+The function definitions below are in the following format:
+<functionName>: <description>
+  inputs:
+    - <parameterName>: <parameterDescription>
+    - ...
+
+{{$function_descriptions}}
+[END AVAILABLE FUNCTIONS]
+
+[USAGE INSTRUCTIONS]
+To use the functions, specify a JSON blob representing an action. The JSON blob should contain an "action" key with the name of the function to use, and an "action_variables" key with a JSON object of string values to use when calling the function.
+Do not call functions directly; they must be invoked through an action.
+The "action_variables" value should always include an "input" key, even if the input value is empty. Additional keys in the "action_variables" value should match the defined [PARAMETERS] of the named "action" in [AVAILABLE FUNCTIONS].
+Dictionary values in "action_variables" must be strings and represent the actual values to be passed to the function.
+Ensure that the $JSON_BLOB contains only a SINGLE action; do NOT return multiple actions.
+IMPORTANT: Use only the available functions listed in the [AVAILABLE FUNCTIONS] section. Do not attempt to use any other functions that are not specified.
+
+Here is an example of a valid $JSON_BLOB:
+{
+  "action": "functionName",
+  "action_variables": {"parameterName": "some value", ...}
+}
+[END USAGE INSTRUCTIONS]
+[END INSTRUCTION]
+
+[THOUGHT PROCESS]
+[QUESTION]
+the input question I must answer
+[THOUGHT]
+To solve this problem, I should carefully analyze the given question and identify the necessary steps. Any facts I discover earlier in my thought process should be repeated here to keep them readily available.
+[ACTION]
+{
+  "action": "functionName",
+  "action_variables": {"parameterName": "some value", ...}
+}
+[OBSERVATION]
+The result of the action will be provided here.
+... (These Thought/Action/Observation can repeat until the final answer is reached.)
+[FINAL ANSWER]
+Once I have gathered all the necessary observations and performed any required actions, I can provide the final answer in a clear and human-readable format.
+[END THOUGHT PROCESS]
+
+Let's break down the problem step by step and think about the best approach. Questions and observations should be followed by a single thought and an optional single action to take.
+
+Begin!
+
+[QUESTION]
+{{$question}}
+{{$agent_scratch_pad}}
@@ -0,0 +1,5 @@
+from semantic_kernel.planning.stepwise_planner.stepwise_planner import StepwisePlanner
+
+__all__ = [
+    "StepwisePlanner",
+]
@@ -0,0 +1,461 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import itertools
+import json
+import os
+import re
+from typing import TYPE_CHECKING, Dict, List
+
+from semantic_kernel.kernel import Kernel
+from semantic_kernel.orchestration.sk_context import SKContext
+from semantic_kernel.planning.plan import Plan
+from semantic_kernel.planning.planning_exception import PlanningException
+from semantic_kernel.planning.stepwise_planner.stepwise_planner_config import (
+    StepwisePlannerConfig,
+)
+from semantic_kernel.planning.stepwise_planner.system_step import SystemStep
+from semantic_kernel.semantic_functions.prompt_template import PromptTemplate
+from semantic_kernel.semantic_functions.prompt_template_config import (
+    PromptTemplateConfig,
+)
+from semantic_kernel.semantic_functions.semantic_function_config import (
+    SemanticFunctionConfig,
+)
+from semantic_kernel.skill_definition.function_view import FunctionView
+from semantic_kernel.skill_definition.sk_function_context_parameter_decorator import (
+    sk_function_context_parameter,
+)
+from semantic_kernel.skill_definition.sk_function_decorator import sk_function
+
+if TYPE_CHECKING:
+    from semantic_kernel.orchestration.sk_function_base import SKFunctionBase
+
+
+CUR_DIR = os.path.dirname(os.path.realpath(__file__))
+PROMPT_CONFIG_FILE_PATH = os.path.join(CUR_DIR, "Skills/StepwiseStep/config.json")
+PROMPT_TEMPLATE_FILE_PATH = os.path.join(CUR_DIR, "Skills/StepwiseStep/skprompt.txt")
+
+
+def read_file(file_path: str) -> str:
+    with open(file_path, "r") as file:
+        return file.read()
+
+
+# TODO: Original C# uses "StepwisePlanner_Excluded" for RESTRICTED_SKILL_NAME
+RESTRICTED_SKILL_NAME = "StepwisePlanner"
+S_FINAL_ANSWER_REGEX = re.compile(
+    r"\[FINAL[_\s\-]ANSWER\](?P<final_answer>.+)", re.DOTALL
+)
+S_THOUGHT_REGEX = re.compile(
+    r"(\[THOUGHT\])?(?P<thought>.+?)(?=\[ACTION\]|$)", re.DOTALL
+)
+S_ACTION_REGEX = re.compile(r"\[ACTION\][^{}]*({(?:[^{}]*{[^{}]*})*[^{}]*})", re.DOTALL)
+
+ACTION = "[ACTION]"
+THOUGHT = "[THOUGHT]"
+OBSERVATION = "[OBSERVATION]"
+SCRATCH_PAD_PREFIX = (
+    "This was my previous work (but they haven't seen any of it!"
+    " They only see what I return as final answer):"
+)
+
+
+def is_null_or_empty(value: str) -> bool:
+    return value is None or value == ""
+
+
+class StepwisePlanner:
+    config: StepwisePlannerConfig
+    _context: "SKContext"
+    _function_flow_function: "SKFunctionBase"
+
+    def __init__(
+        self,
+        kernel: Kernel,
+        config: StepwisePlannerConfig = None,
+        prompt: str = None,
+        prompt_user_config: PromptTemplateConfig = None,
+    ):
+        assert isinstance(kernel, Kernel)
+        self._kernel = kernel
+
+        self.config = config or StepwisePlannerConfig()
+        self.config.excluded_skills.append(RESTRICTED_SKILL_NAME)
+
+        prompt_config = prompt_user_config or PromptTemplateConfig()
+        prompt_template = prompt or read_file(PROMPT_TEMPLATE_FILE_PATH)
+
+        if prompt_user_config is None:
+            prompt_config = PromptTemplateConfig.from_json(
+                read_file(PROMPT_CONFIG_FILE_PATH)
+            )
+
+        prompt_config.completion.max_tokens = self.config.max_tokens
+
+        self._system_step_function = self.import_semantic_function(
+            kernel, "StepwiseStep", prompt_template, prompt_config
+        )
+        self._native_functions = self._kernel.import_skill(self, RESTRICTED_SKILL_NAME)
+
+        self._context = kernel.create_new_context()
+        self._logger = self._kernel.logger
+
+    def create_plan(self, goal: str) -> Plan:
+        if is_null_or_empty(goal):
+            raise PlanningException(
+                PlanningException.ErrorCodes.InvalidGoal, "The goal specified is empty"
+            )
+
+        function_descriptions = self.get_function_descriptions()
+
+        plan_step: Plan = Plan.from_function(self._native_functions["ExecutePlan"])
+        plan_step.parameters.set("function_descriptions", function_descriptions)
+        plan_step.parameters.set("question", goal)
+
+        plan_step._outputs.append("agent_scratch_pad")
+        plan_step._outputs.append("step_count")
+        plan_step._outputs.append("skill_count")
+        plan_step._outputs.append("steps_taken")
+
+        plan = Plan(goal)
+
+        plan.add_steps([plan_step])
+
+        return plan
+
+    # TODO: sync C# with https://github.com/microsoft/semantic-kernel/pull/1195
+    @sk_function(name="ExecutePlan", description="Execute a plan")
+    @sk_function_context_parameter(
+        name="question", description="The question to answer"
+    )
+    @sk_function_context_parameter(
+        name="function_descriptions", description="List of tool descriptions"
+    )
+    async def execute_plan_async(self, context: SKContext):
+        question = context["question"]
+
+        steps_taken: List[SystemStep] = []
+        if not is_null_or_empty(question):
+            for i in range(self.config.max_iterations):
+                scratch_pad = self.create_scratch_pad(question, steps_taken)
+
+                context.variables.set("agent_scratch_pad", scratch_pad)
+
+                llm_response = await self._system_step_function.invoke_async(
+                    context=context
+                )
+
+                if llm_response.error_occurred:
+                    raise PlanningException(
+                        PlanningException.ErrorCodes.UnknownError,
+                        f"Error occurred while executing stepwise plan: {llm_response.last_exception}",
+                        llm_response.last_exception,
+                    )
+
+                action_text = llm_response.result.strip()
+                self._logger.debug(f"Response: {action_text}")
+
+                next_step = self.parse_result(action_text)
+                steps_taken.append(next_step)
+
+                if not is_null_or_empty(next_step.final_answer):
+                    self._logger.debug(f"Final Answer: {next_step.final_answer}")
+
+                    context.variables.update(next_step.final_answer)
+                    updated_scratch_pad = self.create_scratch_pad(question, steps_taken)
+                    context.variables.set("agent_scratch_pad", updated_scratch_pad)
+
+                    # Add additional results to the context
+                    self.add_execution_stats_to_context(steps_taken, context)
+
+                    return context
+
+                self._logger.debug("Thoughts: {next_step.thought}")
+
+                if not is_null_or_empty(next_step.action):
+                    self._logger.info(f"Action: {next_step.action}. Iteration: {i+1}.")
+                    self._logger.debug(
+                        f"Action: {next_step.action}({next_step.action_variables}). Iteration: {i+1}.",
+                    )
+
+                    try:
+                        await asyncio.sleep(self.config.min_iteration_time_ms / 1000)
+                        result = await self.invoke_action_async(
+                            next_step.action, next_step.action_variables
+                        )
+
+                        if is_null_or_empty(result):
+                            next_step.observation = "Got no result from action"
+                        else:
+                            next_step.observation = result
+
+                    except Exception as e:
+                        next_step.observation = (
+                            f"Error invoking action {next_step.action}: {str(e)}"
+                        )
+                        self._logger.warning(
+                            f"Error invoking action {next_step.action}"
+                        )
+
+                    self._logger.debug(f"Observation: {next_step.observation}")
+                else:
+                    self._logger.info("Action: No action to take")
+
+                # sleep 3 seconds
+                await asyncio.sleep(self.config.min_iteration_time_ms / 1000)
+
+            steps_taken_str = json.dumps([s.__dict__ for s in steps_taken], indent=4)
+            context.variables.update(
+                f"Result not found, review _steps_taken to see what happened.\n{steps_taken_str}"
+            )
+        else:
+            context.variables.update("Question not found.")
+
+        return context
+
+    def parse_result(self, input: str):
+        result = SystemStep(original_response=input)
+
+        # Extract final answer
+        final_answer_match = re.search(S_FINAL_ANSWER_REGEX, input)
+
+        if final_answer_match:
+            result.final_answer = final_answer_match.group(1).strip()
+            return result
+
+        # Extract thought
+        thought_match = re.search(S_THOUGHT_REGEX, input)
+
+        if thought_match:
+            result.thought = thought_match.group(0).strip()
+        elif not input.contains(ACTION):
+            result.thought = input
+        else:
+            raise ValueError("Unexpected input format")
+
+        result.thought = result.thought.replace(THOUGHT, "").strip()
+
+        # Extract action
+        action_match = re.search(S_ACTION_REGEX, input)
+
+        if action_match:
+            action_json = action_match.group(1).strip()
+
+            try:
+                system_step_results = json.loads(action_json)
+
+                if system_step_results is None or len(system_step_results) == 0:
+                    result.observation = (
+                        f"System step parsing error, empty JSON: {action_json}"
+                    )
+                else:
+                    result.action = system_step_results["action"]
+                    result.action_variables = system_step_results["action_variables"]
+            except Exception:
+                result.observation = (
+                    f"System step parsing error, invalid JSON: {action_json}"
+                )
+
+        if is_null_or_empty(result.thought) and is_null_or_empty(result.action):
+            result.observation = (
+                "System step error, no thought or action found.",
+                "Please give a valid thought and/or action.",
+            )
+
+        return result
+
+    def add_execution_stats_to_context(
+        self, steps_taken: List[SystemStep], context: SKContext
+    ):
+        context.variables.set("step_count", str(len(steps_taken)))
+        context.variables.set(
+            "steps_taken", json.dumps([s.__dict__ for s in steps_taken], indent=4)
+        )
+
+        action_counts: Dict[str, int] = {}
+        for step in steps_taken:
+            if is_null_or_empty(step.action):
+                continue
+
+            current_count = action_counts.get(step.action, 0)
+            action_counts[step.action] = current_count + 1
+
+        skill_call_list_with_counts = [
+            f"{skill}({action_counts[skill]})" for skill in action_counts
+        ]
+        skill_call_list_with_counts = ", ".join(skill_call_list_with_counts)
+        skill_call_count_str = str(sum(action_counts.values()))
+
+        context.variables.set(
+            "skill_count", f"{skill_call_count_str} ({skill_call_list_with_counts})"
+        )
+
+    def create_scratch_pad(self, question: str, steps_taken: List[SystemStep]) -> str:
+        if len(steps_taken) == 0:
+            return ""
+
+        scratch_pad_lines: List[str] = []
+
+        # Add the original first thought
+        scratch_pad_lines.append(SCRATCH_PAD_PREFIX)
+        scratch_pad_lines.append(f"{THOUGHT}\n{steps_taken[0].thought}")
+
+        # Keep track of where to insert the next step
+        insert_point = len(scratch_pad_lines)
+
+        for i in reversed(range(len(steps_taken))):
+            if len(scratch_pad_lines) / 4.0 > (self.config.max_tokens * 0.75):
+                self._logger.debug(
+                    f"Scratchpad is too long, truncating. Skipping {i + 1} steps."
+                )
+                break
+
+            s = steps_taken[i]
+
+            if not is_null_or_empty(s.observation):
+                scratch_pad_lines.insert(
+                    insert_point, f"{OBSERVATION}\n{s.observation}"
+                )
+
+            if not is_null_or_empty(s.action):
+                scratch_pad_lines.insert(
+                    insert_point,
+                    f'{ACTION}\n{{"action": "{s.action}", "action_variables": {json.dumps(s.action_variables)}}}',
+                )
+
+            if i != 0:
+                scratch_pad_lines.insert(insert_point, f"{THOUGHT}\n{s.thought}")
+
+        scratch_pad = "\n".join(scratch_pad_lines).strip()
+
+        if not (is_null_or_empty(scratch_pad.strip())):
+            self._logger.debug(f"Scratchpad: {scratch_pad}")
+
+        return scratch_pad
+
+    async def invoke_action_async(
+        self, action_name: str, action_variables: Dict[str, str]
+    ) -> str:
+        available_functions = self.get_available_functions()
+        target_function = next(
+            (
+                f
+                for f in available_functions
+                if self.to_fully_qualified_name(f) == action_name
+            ),
+            None,
+        )
+
+        if target_function is None:
+            raise PlanningException(
+                PlanningException.ErrorCodes.UnknownError,
+                f"The function '{action_name}' was not found.",
+            )
+
+        try:
+            function = self._kernel.func(
+                target_function.skill_name, target_function.name
+            )
+            action_context = self.create_action_context(action_variables)
+
+            result = await function.invoke_async(context=action_context)
+
+            if result.error_occurred:
+                self._logger.error(f"Error occurred: {result.last_exception}")
+                return f"Error occurred: {result.last_exception}"
+
+            self._logger.debug(
+                f"Invoked {target_function.name}. Result: {result.result}"
+            )
+
+            return result.result
+
+        except Exception as e:
+            self._logger.error(
+                e,
+                f"Something went wrong in system step: {target_function.skill_name}.{target_function.name}. Error: {e}",
+            )
+            return (
+                "Something went wrong in system step: ",
+                f"{target_function.skill_name}.{target_function.name}. Error: {e}",
+            )
+
+    def create_action_context(self, action_variables: Dict[str, str]) -> SKContext:
+        action_context = self._kernel.create_new_context()
+        if action_variables is not None:
+            for k, v in action_variables.items():
+                action_context.variables.set(k, v)
+
+        return action_context
+
+    def get_available_functions(self) -> List[FunctionView]:
+        functions_view = self._context.skills.get_functions_view()
+
+        excluded_skills = self.config.excluded_skills or []
+        excluded_functions = self.config.excluded_functions or []
+
+        available_functions: List[FunctionView] = [
+            *functions_view.semantic_functions.values(),
+            *functions_view.native_functions.values(),
+        ]
+        available_functions = itertools.chain.from_iterable(available_functions)
+        available_functions = [
+            func
+            for func in available_functions
+            if (
+                func.skill_name not in excluded_skills
+                and func.name not in excluded_functions
+            )
+        ]
+        available_functions = sorted(
+            available_functions, key=lambda x: (x.skill_name, x.name)
+        )
+
+        return available_functions
+
+    def get_function_descriptions(self) -> str:
+        available_functions = self.get_available_functions()
+
+        function_descriptions = "\n".join(
+            [self.to_manual_string(f) for f in available_functions]
+        )
+        return function_descriptions
+
+    def import_semantic_function(
+        self,
+        kernel: Kernel,
+        function_name: str,
+        prompt_template: str,
+        config: PromptTemplateConfig = None,
+    ) -> "SKFunctionBase":
+        template = PromptTemplate(
+            prompt_template, kernel.prompt_template_engine, config
+        )
+        function_config = SemanticFunctionConfig(config, template)
+
+        return kernel.register_semantic_function(
+            RESTRICTED_SKILL_NAME, function_name, function_config
+        )
+
+    def to_manual_string(self, function: FunctionView) -> str:
+        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)
+
+        function_description = function.description.strip()
+
+        if is_null_or_empty(inputs):
+            return f"{self.to_fully_qualified_name(function)}: {function_description}\n  inputs: None\n"
+
+        return f"{self.to_fully_qualified_name(function)}: {function_description}\n  inputs:\n{inputs}\n"
+
+    def to_fully_qualified_name(self, function: FunctionView):
+        return f"{function.skill_name}.{function.name}"
@@ -0,0 +1,25 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from typing import List, Optional
+
+
+class StepwisePlannerConfig:
+    def __init__(
+        self,
+        relevancy_threshold: Optional[float] = None,
+        max_relevant_functions: int = 100,
+        excluded_skills: List[str] = None,
+        excluded_functions: List[str] = None,
+        included_functions: List[str] = None,
+        max_tokens: int = 1024,
+        max_iterations: int = 100,
+        min_iteration_time_ms: int = 0,
+    ):
+        self.relevancy_threshold: float = relevancy_threshold
+        self.max_relevant_functions: int = max_relevant_functions
+        self.excluded_skills: List[str] = excluded_skills or []
+        self.excluded_functions: List[str] = excluded_functions or []
+        self.included_functions: List[str] = included_functions or []
+        self.max_tokens: int = max_tokens
+        self.max_iterations: int = max_iterations
+        self.min_iteration_time_ms: int = min_iteration_time_ms
@@ -0,0 +1,12 @@
+from dataclasses import dataclass, field
+from typing import Dict, Optional
+
+
+@dataclass
+class SystemStep:
+    thought: Optional[str] = None
+    action: Optional[str] = None
+    action_variables: Optional[Dict[str, str]] = field(default_factory=dict)
+    observation: Optional[str] = None
+    final_answer: Optional[str] = None
+    original_response: Optional[str] = None
@@ -81,6 +81,22 @@ def pinecone_settings_from_dot_env() -> Tuple[str, str]:
    return api_key, environment


+def bing_search_settings_from_dot_env() -> str:
+    """Reads the Bing Search API key from the .env file.
+
+    Returns:
+        Tuple[str, str]: The Bing Search API key, the Bing Search endpoint
+    """
+
+    api_key = None
+    config = dotenv_values(".env")
+    api_key = config.get("BING_API_KEY", None)
+
+    assert api_key is not None, "Bing Search API key not found in .env file"
+
+    return api_key
+
+
 def google_palm_settings_from_dot_env() -> str:
    """
    Reads the Google PaLM API key from the .env file.
@@ -0,0 +1,174 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import json
+import os
+
+import pytest
+
+import semantic_kernel as sk
+import semantic_kernel.connectors.ai.open_ai as sk_oai
+from semantic_kernel.connectors.search_engine import BingConnector
+from semantic_kernel.core_skills.math_skill import MathSkill
+from semantic_kernel.core_skills.time_skill import TimeSkill
+from semantic_kernel.kernel import Kernel
+from semantic_kernel.orchestration.sk_context import SKContext
+from semantic_kernel.planning import StepwisePlanner
+from semantic_kernel.planning.stepwise_planner.stepwise_planner_config import (
+    StepwisePlannerConfig,
+)
+from semantic_kernel.skill_definition import sk_function, sk_function_context_parameter
+
+
+class TempWebSearchEngineSkill:
+    """
+    TODO: replace this class with semantic_kernel.core_skills.web_search_engine_skill.WebSearchEngineSkill
+
+    SKFunction.describe() does not contains info for arguments.
+
+    so that `query: str` is not shown in the function description,
+    BUT this argument must be passed to planner to work appropriately.
+
+    This function temporarily add `query` as parameter by using @sk_function_context_parameter.
+    original file is here: semantic-kernel/python/semantic_kernel/core_skills/web_search_engine_skill.py
+    """
+
+    def __init__(self, connector) -> None:
+        self._connector = connector
+
+    @sk_function(
+        description="Performs a web search for a given query", name="searchAsync"
+    )
+    @sk_function_context_parameter(
+        name="query",
+        description="The search query",
+    )
+    async def search_async(self, query: str, context: SKContext) -> str:
+        query = query or context.variables.get("query")[1]
+        result = await self._connector.search_async(query, num_results=5, offset=0)
+        return str(result)
+
+
+@pytest.fixture(scope="session")
+def get_bing_config():
+    if "Python_Integration_Tests" in os.environ:
+        api_key = os.environ["Bing__ApiKey"]
+    else:
+        # Load credentials from .env file
+        api_key = sk.bing_search_settings_from_dot_env()
+
+    return api_key
+
+
+def initialize_kernel(get_aoai_config, use_embeddings=False, use_chat_model=False):
+    _, api_key, endpoint = get_aoai_config
+
+    kernel = Kernel()
+    if use_chat_model:
+        kernel.add_chat_service(
+            "chat_completion",
+            sk_oai.AzureChatCompletion("gpt-35-turbo", endpoint, api_key),
+        )
+    else:
+        kernel.add_text_completion_service(
+            "text_completion",
+            sk_oai.AzureChatCompletion("gpt-35-turbo", endpoint, api_key),
+        )
+
+    if use_embeddings:
+        kernel.add_text_embedding_generation_service(
+            "text_embedding",
+            sk_oai.AzureTextEmbedding("text-embedding-ada-002", endpoint, api_key),
+        )
+    return kernel
+
+
+@pytest.mark.parametrize(
+    "use_chat_model, prompt, expected_function, expected_skill",
+    [
+        (
+            False,
+            "What is the tallest mountain on Earth? How tall is it divided by 2?",
+            "ExecutePlan",
+            "StepwisePlanner",
+        ),
+        (
+            True,
+            "What is the tallest mountain on Earth? How tall is it divided by 2?",
+            "ExecutePlan",
+            "StepwisePlanner",
+        ),
+    ],
+)
+@pytest.mark.asyncio
+async def test_can_create_stepwise_plan(
+    get_aoai_config,
+    get_bing_config,
+    use_chat_model,
+    prompt,
+    expected_function,
+    expected_skill,
+):
+    # Arrange
+    use_embeddings = False
+    kernel = initialize_kernel(get_aoai_config, use_embeddings, use_chat_model)
+    bing_connector = BingConnector(api_key=get_bing_config)
+    web_search_engine_skill = TempWebSearchEngineSkill(bing_connector)
+    kernel.import_skill(web_search_engine_skill, "WebSearch")
+    kernel.import_skill(TimeSkill(), "time")
+
+    planner = StepwisePlanner(
+        kernel, StepwisePlannerConfig(max_iterations=10, min_iteration_time_ms=1000)
+    )
+
+    # Act
+    plan = planner.create_plan(prompt)
+
+    # Assert
+    assert any(
+        step.name == expected_function and step.skill_name == expected_skill
+        for step in plan._steps
+    )
+
+
+@pytest.mark.parametrize(
+    "use_chat_model, prompt",
+    [
+        (
+            False,
+            "What is the tallest mountain on Earth? How tall is it divided by 2?",
+        )
+    ],
+)
+@pytest.mark.asyncio
+async def test_can_execute_stepwise_plan(
+    get_aoai_config,
+    get_bing_config,
+    use_chat_model,
+    prompt,
+):
+    # Arrange
+    use_embeddings = False
+    kernel = initialize_kernel(get_aoai_config, use_embeddings, use_chat_model)
+    bing_connector = BingConnector(api_key=get_bing_config)
+    web_search_engine_skill = TempWebSearchEngineSkill(bing_connector)
+    kernel.import_skill(web_search_engine_skill, "WebSearch")
+    kernel.import_skill(TimeSkill(), "time")
+    kernel.import_skill(MathSkill(), "math")
+
+    planner = StepwisePlanner(
+        kernel, StepwisePlannerConfig(max_iterations=10, min_iteration_time_ms=1000)
+    )
+
+    # Act
+    plan = planner.create_plan(prompt)
+    result = await plan.invoke_async()
+
+    steps_taken_string = result.variables["steps_taken"]
+    assert steps_taken_string is not None
+
+    steps_taken = json.loads(steps_taken_string)
+    assert steps_taken is not None and len(steps_taken) > 0
+
+    assert (
+        3 <= len(steps_taken) <= 10
+    ), f"Actual: {len(steps_taken)}. Expected at least 3 steps and at most 10 steps to be taken."
@@ -148,12 +148,13 @@ async def test_azure_chat_completion_call_with_parameters() -> None:
            organization=None,
            messages=messages,
            temperature=complete_request_settings.temperature,
-            max_tokens=complete_request_settings.max_tokens,
            top_p=complete_request_settings.top_p,
-            presence_penalty=complete_request_settings.presence_penalty,
-            frequency_penalty=complete_request_settings.frequency_penalty,
            n=complete_request_settings.number_of_responses,
            stream=False,
+            stop=None,
+            max_tokens=complete_request_settings.max_tokens,
+            presence_penalty=complete_request_settings.presence_penalty,
+            frequency_penalty=complete_request_settings.frequency_penalty,
            logit_bias={},
        )

@@ -197,11 +198,62 @@ async def test_azure_chat_completion_call_with_parameters_and_Logit_Bias_Defined
            organization=None,
            messages=messages,
            temperature=complete_request_settings.temperature,
-            max_tokens=complete_request_settings.max_tokens,
            top_p=complete_request_settings.top_p,
-            presence_penalty=complete_request_settings.presence_penalty,
-            frequency_penalty=complete_request_settings.frequency_penalty,
            n=complete_request_settings.number_of_responses,
            stream=False,
+            stop=None,
+            max_tokens=complete_request_settings.max_tokens,
+            presence_penalty=complete_request_settings.presence_penalty,
+            frequency_penalty=complete_request_settings.frequency_penalty,
            logit_bias=token_bias,
        )
+
+
+@pytest.mark.asyncio
+async def test_azure_chat_completion_call_with_parameters_and_Stop_Defined() -> None:
+    mock_openai = AsyncMock()
+    with patch(
+        "semantic_kernel.connectors.ai.open_ai.services.open_ai_chat_completion.openai",
+        new=mock_openai,
+    ):
+        deployment_name = "test_deployment"
+        endpoint = "https://test-endpoint.com"
+        api_key = "test_api_key"
+        api_type = "azure"
+        api_version = "2023-03-15-preview"
+        logger = Logger("test_logger")
+        prompt = "hello world"
+        messages = [{"role": "user", "content": prompt}]
+        complete_request_settings = CompleteRequestSettings()
+
+        stop = ["!"]
+        complete_request_settings.stop_sequences = stop
+
+        azure_chat_completion = AzureChatCompletion(
+            deployment_name=deployment_name,
+            endpoint=endpoint,
+            api_key=api_key,
+            api_version=api_version,
+            logger=logger,
+        )
+
+        await azure_chat_completion.complete_async(prompt, complete_request_settings)
+
+        mock_openai.ChatCompletion.acreate.assert_called_once_with(
+            engine=deployment_name,
+            api_key=api_key,
+            api_type=api_type,
+            api_base=endpoint,
+            api_version=api_version,
+            organization=None,
+            messages=messages,
+            temperature=complete_request_settings.temperature,
+            top_p=complete_request_settings.top_p,
+            n=complete_request_settings.number_of_responses,
+            stream=False,
+            stop=complete_request_settings.stop_sequences,
+            max_tokens=complete_request_settings.max_tokens,
+            presence_penalty=complete_request_settings.presence_penalty,
+            frequency_penalty=complete_request_settings.frequency_penalty,
+            logit_bias={},
+        )
@@ -0,0 +1,34 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from unittest.mock import Mock
+
+import pytest
+
+from semantic_kernel.kernel import Kernel
+from semantic_kernel.planning.stepwise_planner.stepwise_planner import StepwisePlanner
+
+
+@pytest.mark.parametrize(
+    "input, expected",
+    [
+        ("[FINAL ANSWER] 42", "42"),
+        ("[FINAL ANSWER]42", "42"),
+        ("I think I have everything I need.\n[FINAL ANSWER] 42", "42"),
+        ("I think I have everything I need.\n[FINAL ANSWER] 42\n", "42"),
+        ("I think I have everything I need.\n[FINAL ANSWER] 42\n\n", "42"),
+        ("I think I have everything I need.\n[FINAL ANSWER]42\n\n\n", "42"),
+        ("I think I have everything I need.\n[FINAL ANSWER]\n 42\n\n\n", "42"),
+    ],
+)
+def test_when_input_is_final_answer_returns_final_answer(input: str, expected: str):
+    kernel = Mock(spec=Kernel)
+
+    planner = StepwisePlanner(kernel)
+
+    result = planner.parse_result(input)
+
+    assert result.final_answer == expected
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])