From 5e4e0edfaa94b96f008cbd3f8669056e7a145ffa Mon Sep 17 00:00:00 2001
From: Abby Harrison <54643756+awharrison-28@users.noreply.github.com>
Date: Fri, 14 Jul 2023 18:34:19 -0700
Subject: [PATCH] Python: Plan Object Model (#1907)

### Motivation and Context
This PR ports the plan object model found in the .NET SK to Python. This
fundamentally changes the concept of a planner from a `semantic skill`
to a unique type of `SKFunction`. With this model, we can begin
implementing sequential planner and offer much better plan
customization.

**Notes:**
- This PR does not implement plan serialization/deserialization to/from
JSON. There is a huge existing effort to make the Python SK completely
Pydantic enabled, which will bring serialization and deserialization
capability with it.
- This PR does not include ports of `Action Planner` and `Sequential
Planner` to the Plan Object Model (POM). Following PRs will address
this.
- I also added some missing copywrite headers. A future PR will add
headers to any still missing them.
- added a VSCode launch configuration to run the current file with
pytest
- closes #1486


### Description
- Added file called `plan.py` which is a python port of the Plan.cs file
found in the .NET SK. Note that it does not contain serialization logic
at this time.
- Added new kernel error `FunctionInvokeError` - mirrors the error list
of the .NET SK
- Added tests for Plan creation
- Added tests for Plan execution - both sync and async


### Contribution Checklist
<!-- Before submitting this PR, please make sure: -->
- [ ] The code builds clean without any errors or warnings
- [ ] The PR follows SK Contribution Guidelines
(https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
- [ ] The code follows the .NET coding conventions
(https://learn.microsoft.com/dotnet/csharp/fundamentals/coding-style/coding-conventions)
verified with `dotnet format`
- [ ] All unit tests pass, and I have added new tests where possible
- [ ] I didn't break anyone :smile:

---------

Co-authored-by: Shawn Callegari <36091529+shawncal@users.noreply.github.com>
---
 .vscode/launch.json                           |   8 +
 .../core_skills/file_io_skill.py              |   2 +
 .../semantic_kernel/core_skills/text_skill.py |   2 +
 .../semantic_kernel/core_skills/time_skill.py |   2 +
 .../semantic_kernel/core_skills/wait_skill.py |   2 +
 python/semantic_kernel/kernel_exception.py    |   4 +-
 python/semantic_kernel/planning/plan.py       | 439 +++++++++++++++++-
 .../tests/unit/planning/test_plan_creation.py | 192 ++++++++
 .../unit/planning/test_plan_execution.py      | 182 ++++++++
 9 files changed, 820 insertions(+), 13 deletions(-)
 create mode 100644 python/tests/unit/planning/test_plan_creation.py
 create mode 100644 python/tests/unit/planning/test_plan_execution.py

diff --git a/.vscode/launch.json b/.vscode/launch.json
index 186e934244..83c63b4f51 100644
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -34,6 +34,14 @@
       "name": ".NET Core Attach",
       "type": "coreclr",
       "request": "attach"
+    },
+    {
+      "cwd":"${workspaceFolder}/python",
+      "name": "Python: Test Module",
+      "type": "python",
+      "request": "launch",
+      "module": "pytest",
+      "args": ["${file}"],
     }
   ]
 }
diff --git a/python/semantic_kernel/core_skills/file_io_skill.py b/python/semantic_kernel/core_skills/file_io_skill.py
index ddd45c0c24..250af93078 100644
--- a/python/semantic_kernel/core_skills/file_io_skill.py
+++ b/python/semantic_kernel/core_skills/file_io_skill.py
@@ -1,3 +1,5 @@
+# Copyright (c) Microsoft. All rights reserved.
+
 import os
 
 import aiofiles
diff --git a/python/semantic_kernel/core_skills/text_skill.py b/python/semantic_kernel/core_skills/text_skill.py
index b80d13ee47..9e463caaca 100644
--- a/python/semantic_kernel/core_skills/text_skill.py
+++ b/python/semantic_kernel/core_skills/text_skill.py
@@ -1,3 +1,5 @@
+# Copyright (c) Microsoft. All rights reserved.
+
 from semantic_kernel.skill_definition import sk_function
 
 
diff --git a/python/semantic_kernel/core_skills/time_skill.py b/python/semantic_kernel/core_skills/time_skill.py
index b903652a12..227ff5286d 100644
--- a/python/semantic_kernel/core_skills/time_skill.py
+++ b/python/semantic_kernel/core_skills/time_skill.py
@@ -1,3 +1,5 @@
+# Copyright (c) Microsoft. All rights reserved.
+
 import datetime
 
 from semantic_kernel.skill_definition import sk_function
diff --git a/python/semantic_kernel/core_skills/wait_skill.py b/python/semantic_kernel/core_skills/wait_skill.py
index a59de9df38..8f4ca28fbd 100644
--- a/python/semantic_kernel/core_skills/wait_skill.py
+++ b/python/semantic_kernel/core_skills/wait_skill.py
@@ -1,3 +1,5 @@
+# Copyright (c) Microsoft. All rights reserved.
+
 import asyncio
 
 from semantic_kernel.skill_definition import sk_function
diff --git a/python/semantic_kernel/kernel_exception.py b/python/semantic_kernel/kernel_exception.py
index 47b4313bcd..910fce76ac 100644
--- a/python/semantic_kernel/kernel_exception.py
+++ b/python/semantic_kernel/kernel_exception.py
@@ -24,8 +24,10 @@ class KernelException(Exception):
         ServiceNotFound = 6
         # Skill collection not set.
         SkillCollectionNotSet = 7
+        # Represents an error that occurs when invoking a function.
+        FunctionInvokeError = 8
         # Ambiguous implementation.
-        AmbiguousImplementation = 8
+        AmbiguousImplementation = 9
 
     # The error code.
     _error_code: ErrorCodes
diff --git a/python/semantic_kernel/planning/plan.py b/python/semantic_kernel/planning/plan.py
index 90c0e2f851..d20217b480 100644
--- a/python/semantic_kernel/planning/plan.py
+++ b/python/semantic_kernel/planning/plan.py
@@ -1,17 +1,432 @@
-from typing import Any
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import re
+import threading
+from logging import Logger
+from typing import Any, Callable, List, Optional
+
+from semantic_kernel import Kernel
+from semantic_kernel.connectors.ai import CompleteRequestSettings
+from semantic_kernel.connectors.ai.text_completion_client_base import (
+    TextCompletionClientBase,
+)
+from semantic_kernel.kernel_exception import KernelException
+from semantic_kernel.memory.semantic_text_memory_base import SemanticTextMemoryBase
+from semantic_kernel.orchestration.context_variables import ContextVariables
+from semantic_kernel.orchestration.sk_context import SKContext
+from semantic_kernel.orchestration.sk_function_base import SKFunctionBase
+from semantic_kernel.skill_definition.function_view import FunctionView
+from semantic_kernel.skill_definition.read_only_skill_collection_base import (
+    ReadOnlySkillCollectionBase,
+)
 
 
-class Plan:
-    # The goal that wants to be achieved
-    goal: str
+class Plan(SKFunctionBase):
+    _state: ContextVariables
+    _steps: List["Plan"]
+    _function: SKFunctionBase
+    _parameters: ContextVariables
+    _outputs: List[str]
+    _has_next_step: bool
+    _next_step_index: int
+    _name: str
+    _skill_name: str
+    _description: str
+    _is_semantic: bool
+    _request_settings: CompleteRequestSettings
+    DEFAULT_RESULT_KEY = "PLAN.RESULT"
 
-    # The prompt to be used to generate the plan
-    prompt: str
+    @property
+    def name(self) -> str:
+        return self._name
 
-    # The generated plan that consists of a list of steps to complete the goal
-    generated_plan: Any
+    @property
+    def state(self) -> ContextVariables:
+        return self._state
 
-    def __init__(self, goal, prompt, plan):
-        self.goal = goal
-        self.prompt = prompt
-        self.generated_plan = plan
+    @property
+    def skill_name(self) -> str:
+        return self._skill_name
+
+    @property
+    def description(self) -> str:
+        return self._description
+
+    @property
+    def function(self) -> Callable[..., Any]:
+        return self._function
+
+    @property
+    def parameters(self) -> ContextVariables:
+        return self._parameters
+
+    @property
+    def is_semantic(self) -> bool:
+        return self._is_semantic
+
+    @property
+    def is_native(self) -> bool:
+        if self._is_semantic is None:
+            return None
+        else:
+            return not self._is_semantic
+
+    @property
+    def request_settings(self) -> CompleteRequestSettings:
+        return self._request_settings
+
+    @property
+    def has_next_step(self) -> bool:
+        return self._next_step_index < len(self._steps)
+
+    @property
+    def next_step_index(self) -> int:
+        return self._next_step_index
+
+    def __init__(
+        self,
+        name: Optional[str] = None,
+        skill_name: Optional[str] = None,
+        description: Optional[str] = None,
+        next_step_index: Optional[int] = None,
+        state: Optional[ContextVariables] = None,
+        parameters: Optional[ContextVariables] = None,
+        outputs: Optional[List[str]] = None,
+        steps: Optional[List["Plan"]] = None,
+        function: Optional[SKFunctionBase] = None,
+    ) -> None:
+        self._name = "" if name is None else name
+        self._skill_name = "" if skill_name is None else skill_name
+        self._description = "" if description is None else description
+        self._next_step_index = 0 if next_step_index is None else next_step_index
+        self._state = ContextVariables() if state is None else state
+        self._parameters = ContextVariables() if parameters is None else parameters
+        self._outputs = [] if outputs is None else outputs
+        self._steps = [] if steps is None else steps
+        self._has_next_step = len(self._steps) > 0
+        self._is_semantic = None
+        self._function = None if function is None else function
+        self._request_settings = None
+
+        if function is not None:
+            self.set_function(function)
+
+    async def invoke_async(
+        self,
+        input: Optional[str] = None,
+        context: Optional[SKContext] = None,
+        settings: Optional[CompleteRequestSettings] = None,
+        memory: Optional[SemanticTextMemoryBase] = None,
+        logger: Optional[Logger] = None,
+        # TODO: cancellation_token: CancellationToken,
+    ) -> SKContext:
+        if input is not None:
+            self._state.update(input)
+
+        if context is None:
+            context = SKContext(
+                variables=self._state,
+                skill_collection=None,
+                memory=memory,
+                logger=logger,
+            )
+
+        if self._function is not None:
+            result = await self._function.invoke_async(
+                context=context, settings=settings
+            )
+            if result.error_occurred:
+                result.log.error(
+                    msg="Something went wrong in plan step {0}.{1}:'{2}'".format(
+                        self._skill_name, self._name, context.last_error_description
+                    )
+                )
+                return result
+            context.variables.update(result.result)
+        else:
+            # loop through steps until completion
+            while self.has_next_step:
+                function_context = context
+                self.add_variables_to_context(self._state, function_context)
+                await self.invoke_next_step(function_context)
+                self.update_context_with_outputs(context)
+
+        return context
+
+    def invoke(
+        self,
+        input: Optional[str] = None,
+        context: Optional[SKContext] = None,
+        settings: Optional[CompleteRequestSettings] = None,
+        memory: Optional[SemanticTextMemoryBase] = None,
+        logger: Optional[Logger] = None,
+    ) -> SKContext:
+        if input is not None:
+            self._state.update(input)
+
+        if context is None:
+            context = SKContext(
+                variables=self._state,
+                skill_collection=None,
+                memory=memory,
+                logger=logger,
+            )
+
+        if self._function is not None:
+            result = self._function.invoke(context=context, settings=settings)
+            if result.error_occurred:
+                result.log.error(
+                    result.last_exception,
+                    "Something went wrong in plan step {0}.{1}:'{2}'".format(
+                        self.skill_name, self.name, context.last_error_description
+                    ),
+                )
+                return result
+            context.variables.update(result.result)
+        else:
+            # loop through steps until completion
+            while self.has_next_step:
+                # Check if there is an event loop
+                try:
+                    loop = asyncio.get_running_loop()
+                except RuntimeError:
+                    loop = None
+                function_context = context
+                self.add_variables_to_context(self._state, function_context)
+
+                # Handle "asyncio.run() cannot be called from a running event loop"
+                if loop and loop.is_running():
+                    self._runThread(self.invoke_next_step(function_context))
+                else:
+                    asyncio.run(self.invoke_next_step(function_context))
+                self.update_context_with_outputs(context)
+        return context
+
+    def set_ai_configuration(
+        self,
+        settings: CompleteRequestSettings,
+    ) -> SKFunctionBase:
+        if self._function is not None:
+            self._function.set_ai_configuration(settings)
+
+    def set_ai_service(
+        self, service: Callable[[], TextCompletionClientBase]
+    ) -> SKFunctionBase:
+        if self._function is not None:
+            self._function.set_ai_service(service)
+
+    def set_default_skill_collection(
+        self,
+        skills: ReadOnlySkillCollectionBase,
+    ) -> SKFunctionBase:
+        if self._function is not None:
+            self._function.set_default_skill_collection(skills)
+
+    def describe(self) -> FunctionView:
+        return self._function.describe()
+
+    def set_available_functions(self, plan: "Plan", context: SKContext) -> "Plan":
+        if len(plan.steps) == 0:
+            if context.skills is None:
+                raise KernelException(
+                    KernelException.ErrorCodes.SkillCollectionNotSet,
+                    "Skill collection not found in the context",
+                )
+            try:
+                skillFunction = context.skills.get_function(plan.skill_name, plan.name)
+                plan.set_function(skillFunction)
+            except Exception:
+                pass
+        else:
+            for step in plan.steps:
+                step = self.set_available_functions(step, context)
+
+        return plan
+
+    def add_steps(self, steps: Optional[List[SKFunctionBase]]) -> None:
+        for step in steps:
+            if type(step) is Plan:
+                self._steps.append(step)
+            else:
+                new_step = Plan(
+                    name=step.name,
+                    skill_name=step.skill_name,
+                    description=step.description,
+                    next_step_index=0,
+                    state=ContextVariables(),
+                    parameters=ContextVariables(),
+                    outputs=[],
+                    steps=[],
+                )
+                new_step.set_function(step)
+                self._steps.append(new_step)
+
+    def set_function(self, function: SKFunctionBase) -> None:
+        self._function = function
+        self._name = function.name
+        self._skill_name = function.skill_name
+        self._description = function.description
+        self._is_semantic = function.is_semantic
+        self._request_settings = function.request_settings
+
+    async def run_next_step_async(
+        self,
+        kernel: Kernel,
+        variables: ContextVariables,
+    ) -> "Plan":
+        context = kernel.create_new_context(variables)
+        return await self.invoke_next_step(context)
+
+    async def invoke_next_step(self, context: SKContext) -> "Plan":
+        if self.has_next_step:
+            step = self._steps[self._next_step_index]
+
+            # merge the state with the current context variables for step execution
+            variables = self.get_next_step_variables(context.variables, step)
+
+            # Invoke the step
+            func_context = SKContext(
+                variables=variables,
+                memory=context._memory,
+                skill_collection=context.skills,
+                logger=context.log,
+            )
+            result = await step.invoke_async(context=func_context)
+            result_value = result.result
+
+            if result.error_occurred:
+                raise KernelException(
+                    KernelException.ErrorCodes.FunctionInvokeError,
+                    "Error occurred while running plan step: "
+                    + result.last_error_description,
+                    result.last_exception,
+                )
+
+            # Update state with result
+            self.state.update(result_value)
+
+            # Update plan result in state with matching outputs (if any)
+            if set(self._outputs).intersection(set(step._outputs)):
+                current_plan_result = ""
+                if Plan.DEFAULT_RESULT_KEY in self._state._variables:
+                    current_plan_result = self._state[Plan.DEFAULT_RESULT_KEY]
+                self._state.set(
+                    Plan.DEFAULT_RESULT_KEY, current_plan_result.strip() + result_value
+                )
+
+            # Update state with outputs (if any)
+            for output in step._outputs:
+                if output in result.variables._variables:
+                    self._state.set(output, result.variables[output])
+                else:
+                    self._state.set(output, result_value)
+
+            # Increment the step
+            self._next_step_index += 1
+
+        return self
+
+    def add_variables_to_context(
+        self, variables: ContextVariables, context: SKContext
+    ) -> None:
+        for key in variables._variables:
+            if not context.variables.contains_key(key):
+                context.variables.set(key, variables[key])
+
+    def update_context_with_outputs(self, context: SKContext) -> None:
+        result_string = ""
+        if Plan.DEFAULT_RESULT_KEY in self._state._variables:
+            result_string = self._state[Plan.DEFAULT_RESULT_KEY]
+        else:
+            result_string = str(self._state)
+
+        context.variables.update(result_string)
+
+        for item in self._steps[self._next_step_index - 1]._outputs:
+            if item in self._state:
+                context.variables.set(item, self._state[item])
+            else:
+                context.variables.set(item, result_string)
+
+        return context
+
+    def get_next_step_variables(
+        self, variables: ContextVariables, step: "Plan"
+    ) -> ContextVariables:
+        # Priority for Input
+        # - Parameters (expand from variables if needed)
+        # - SKContext.Variables
+        # - Plan.State
+        # - Empty if sending to another plan
+        # - Plan.Description
+        input_string = ""
+        if step._parameters["input"] is not None:
+            input_string = self.expand_from_variables(
+                variables, step._parameters["input"]
+            )
+        elif variables["input"] is not None:
+            input_string = variables["input"]
+        elif self._state["input"] is not None:
+            input_string = self._state["input"]
+        elif len(step._steps) > 0:
+            input_string = ""
+        elif self._description is not None:
+            input_string = self._description
+
+        step_variables = ContextVariables(input_string)
+
+        # Priority for remaining stepVariables is:
+        # - Function Parameters (pull from variables or state by a key value)
+        # - Step Parameters (pull from variables or state by a key value)
+        function_params = step.describe()
+        for param in function_params._parameters:
+            if param.name.lower == "input":
+                continue
+            if step_variables.contains_key(param.name):
+                step_variables.set(param.name, variables[param.name])
+            elif (
+                self._state.contains_key(param.name)
+                and self._state[param.name] is not None
+            ):
+                step_variables.set(param.name, self._state[param.name])
+
+        for param_var in step.parameters._variables:
+            if step_variables.contains_key(param_var):
+                continue
+
+            expanded_value = self.expand_from_variables(variables, param_var)
+            if expanded_value.lower() == param_var.lower():
+                step_variables.set(param_var, expanded_value)
+            elif variables.contains_key(param_var):
+                step_variables.set(param_var, variables[param_var])
+            elif self._state.contains_key(param_var):
+                step_variables.set(param_var, self._state[param_var])
+            else:
+                step_variables.set(param_var, expanded_value)
+
+        return step_variables
+
+    def expand_from_variables(
+        self, variables: ContextVariables, input_string: str
+    ) -> str:
+        result = input_string
+        variables_regex = r"\$(?P<var>\w+)"
+        matches = re.findall(variables_regex, input_string)
+        ordered_matches = sorted(
+            matches, key=lambda m: len(m.group("var")), reverse=True
+        )
+
+        for match in ordered_matches:
+            var_name = match.group("var")
+            if variables.contains_key(var_name):
+                result = result.replace(f"${var_name}", variables[var_name])
+
+        return result
+
+    def _runThread(self, code: Callable):
+        result = []
+        thread = threading.Thread(target=self._runCode, args=(code, result))
+        thread.start()
+        thread.join()
+        return result[0]
diff --git a/python/tests/unit/planning/test_plan_creation.py b/python/tests/unit/planning/test_plan_creation.py
new file mode 100644
index 0000000000..1fdc740dcb
--- /dev/null
+++ b/python/tests/unit/planning/test_plan_creation.py
@@ -0,0 +1,192 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import semantic_kernel as sk
+from semantic_kernel.core_skills.math_skill import MathSkill
+from semantic_kernel.orchestration.context_variables import ContextVariables
+from semantic_kernel.planning import Plan
+
+
+def test_create_empty_plan():
+    plan = Plan()
+    assert plan is not None
+    assert plan.name == ""
+    assert type(plan.state) is ContextVariables
+    assert plan.skill_name == ""
+    assert plan.description == ""
+    assert plan.function is None
+    assert type(plan.parameters) is ContextVariables
+    assert plan.is_semantic is None
+    assert plan.is_native is None
+    assert plan.request_settings is None
+    assert plan.has_next_step is False
+    assert plan.next_step_index == 0
+    assert plan._steps == []
+
+
+def test_create_plan_with_name():
+    plan = Plan(name="test")
+    assert plan is not None
+    assert plan.name == "test"
+    assert type(plan.state) is ContextVariables
+    assert plan.skill_name == ""
+    assert plan.description == ""
+    assert plan.function is None
+    assert type(plan.parameters) is ContextVariables
+    assert plan.is_semantic is None
+    assert plan.is_native is None
+    assert plan.request_settings is None
+    assert plan.has_next_step is False
+    assert plan.next_step_index == 0
+    assert plan._steps == []
+
+
+def test_create_plan_with_name_and_description():
+    plan = Plan(name="test", description="test description")
+    assert plan is not None
+    assert plan.name == "test"
+    assert type(plan.state) is ContextVariables
+    assert plan.skill_name == ""
+    assert plan.description == "test description"
+    assert plan.function is None
+    assert type(plan.parameters) is ContextVariables
+    assert plan.is_semantic is None
+    assert plan.is_native is None
+    assert plan.request_settings is None
+    assert plan.has_next_step is False
+    assert plan.next_step_index == 0
+    assert plan._steps == []
+
+
+def test_create_plan_with_state_and_parameters():
+    plan = Plan(
+        name="test",
+        state=ContextVariables(),
+        parameters={"test_param": "test_param_val"},
+    )
+    assert plan is not None
+    assert plan.name == "test"
+    assert plan.state["input"] == ""
+    assert plan.skill_name == ""
+    assert plan.description == ""
+    assert plan.function is None
+    assert plan.parameters["test_param"] == "test_param_val"
+    assert plan.is_semantic is None
+    assert plan.is_native is None
+    assert plan.request_settings is None
+    assert plan.has_next_step is False
+    assert plan.next_step_index == 0
+    assert plan._steps == []
+
+
+def test_create_plan_with_name_and_function():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (math) skill
+    skill = MathSkill()
+    skill_config_dict = kernel.import_skill(skill, "math")
+
+    test_function = skill_config_dict["Add"]
+
+    plan = Plan(name="test", function=test_function)
+    assert plan is not None
+    assert plan.name == "Add"
+    assert type(plan.state) is ContextVariables
+    assert plan.skill_name == "math"
+    assert plan.description == test_function.description
+    assert plan.function is test_function
+    assert type(plan.parameters) is ContextVariables
+    assert plan.is_semantic is test_function.is_semantic
+    assert plan.is_native is test_function.is_native
+    assert plan.request_settings == test_function.request_settings
+    assert plan.has_next_step is False
+    assert plan.next_step_index == 0
+    assert plan._steps == []
+
+
+def test_create_multistep_plan_with_functions():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (math) skill
+    skill = MathSkill()
+    skill_config_dict = kernel.import_skill(skill, "math")
+
+    test_function1 = skill_config_dict["Add"]
+    test_function2 = skill_config_dict["Subtract"]
+
+    plan = Plan(name="multistep_test")
+    plan.add_steps([test_function1, test_function2])
+
+    assert plan is not None
+    assert plan.name == "multistep_test"
+    assert type(plan.state) is ContextVariables
+    assert plan.skill_name == ""
+    assert plan.description == ""
+    assert plan.function is None
+    assert type(plan.parameters) is ContextVariables
+    assert plan.is_semantic is None
+    assert plan.is_native is None
+    assert plan.request_settings is None
+    assert plan.has_next_step is True
+    assert plan.next_step_index == 0
+    assert len(plan._steps) == 2
+
+
+def test_create_multistep_plan_with_plans():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (math) skill
+    skill = MathSkill()
+    skill_config_dict = kernel.import_skill(skill, "math")
+
+    test_function1 = skill_config_dict["Add"]
+    test_function2 = skill_config_dict["Subtract"]
+
+    plan = Plan(name="multistep_test")
+    plan_step1 = Plan(name="step1", function=test_function1)
+    plan_step2 = Plan(name="step2", function=test_function2)
+    plan.add_steps([plan_step1, plan_step2])
+
+    assert plan is not None
+    assert plan.name == "multistep_test"
+    assert type(plan.state) is ContextVariables
+    assert plan.skill_name == ""
+    assert plan.description == ""
+    assert plan.function is None
+    assert type(plan.parameters) is ContextVariables
+    assert plan.is_semantic is None
+    assert plan.is_native is None
+    assert plan.request_settings is None
+    assert plan.has_next_step is True
+    assert plan.next_step_index == 0
+    assert len(plan._steps) == 2
+
+
+def test_add_step_to_plan():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (math) skill
+    skill = MathSkill()
+    skill_config_dict = kernel.import_skill(skill, "math")
+
+    test_function1 = skill_config_dict["Add"]
+    test_function2 = skill_config_dict["Subtract"]
+
+    plan = Plan(name="multistep_test", function=test_function1)
+    plan.add_steps([test_function2])
+    assert plan is not None
+    assert plan.name == "Add"
+    assert type(plan.state) is ContextVariables
+    assert plan.skill_name == "math"
+    assert plan.description == test_function1.description
+    assert plan.function is test_function1
+    assert type(plan.parameters) is ContextVariables
+    assert plan.is_semantic is test_function1.is_semantic
+    assert plan.is_native is test_function1.is_native
+    assert plan.request_settings == test_function1.request_settings
+    assert plan.has_next_step is True
+    assert plan.next_step_index == 0
+    assert len(plan._steps) == 1
diff --git a/python/tests/unit/planning/test_plan_execution.py b/python/tests/unit/planning/test_plan_execution.py
new file mode 100644
index 0000000000..3e993d6565
--- /dev/null
+++ b/python/tests/unit/planning/test_plan_execution.py
@@ -0,0 +1,182 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import pytest
+
+import semantic_kernel as sk
+from semantic_kernel.core_skills.text_skill import TextSkill
+from semantic_kernel.planning import Plan
+
+
+def test_invoke_empty_plan():
+    plan = Plan()
+    result = plan.invoke()
+    assert result.result == ""
+
+
+@pytest.mark.asyncio
+async def test_invoke_empty_plan_async():
+    plan = Plan()
+    result = await plan.invoke_async()
+    assert result.result == ""
+
+
+def test_invoke_plan_constructed_with_function():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (text) skill
+    skill = TextSkill()
+    skill_config_dict = kernel.import_skill(skill, "text")
+    test_function = skill_config_dict["uppercase"]
+
+    # setup context
+    context = kernel.create_new_context()
+    context["input"] = "hello world "
+
+    plan = Plan(name="test", function=test_function)
+    result = plan.invoke(context=context)
+    assert result.result == "HELLO WORLD "
+
+
+@pytest.mark.asyncio
+async def test_invoke_plan_constructed_with_function_async():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (text) skill
+    skill = TextSkill()
+    skill_config_dict = kernel.import_skill(skill, "text")
+    test_function = skill_config_dict["uppercase"]
+
+    # setup context
+    context = kernel.create_new_context()
+    context["input"] = "hello world "
+
+    plan = Plan(name="test", function=test_function)
+    result = await plan.invoke_async(context=context)
+    assert result.result == "HELLO WORLD "
+
+
+def test_invoke_empty_plan_with_added_function_step():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (text) skill
+    skill = TextSkill()
+    skill_config_dict = kernel.import_skill(skill, "text")
+    test_function = skill_config_dict["uppercase"]
+
+    # setup context
+    context = kernel.create_new_context()
+    context["input"] = "hello world "
+
+    plan = Plan(name="test")
+    plan.add_steps([test_function])
+    result = plan.invoke(context=context)
+    assert result.result == "HELLO WORLD "
+
+
+@pytest.mark.asyncio
+async def test_invoke_empty_plan_with_added_function_step_async():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (text) skill
+    skill = TextSkill()
+    skill_config_dict = kernel.import_skill(skill, "text")
+    test_function = skill_config_dict["uppercase"]
+
+    # setup context
+    context = kernel.create_new_context()
+    context["input"] = "hello world "
+
+    plan = Plan(name="test")
+    plan.add_steps([test_function])
+    result = await plan.invoke_async(context=context)
+    assert result.result == "HELLO WORLD "
+
+
+def test_invoke_empty_plan_with_added_plan_step():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (text) skill
+    skill = TextSkill()
+    skill_config_dict = kernel.import_skill(skill, "text")
+    test_function = skill_config_dict["uppercase"]
+
+    # setup context
+    context = kernel.create_new_context()
+    context["input"] = "hello world "
+
+    plan = Plan(name="test")
+    new_step = Plan(name="test", function=test_function)
+    plan.add_steps([new_step])
+    result = plan.invoke(context=context)
+    assert result.result == "HELLO WORLD "
+
+
+@pytest.mark.asyncio
+async def test_invoke_empty_plan_with_added_plan_step_async():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (text) skill
+    skill = TextSkill()
+    skill_config_dict = kernel.import_skill(skill, "text")
+    test_function = skill_config_dict["uppercase"]
+
+    # setup context
+    context = kernel.create_new_context()
+    context["input"] = "hello world "
+
+    plan = Plan(name="test")
+    new_step = Plan(name="test", function=test_function)
+    plan.add_steps([new_step])
+    result = await plan.invoke_async(context=context)
+    assert result.result == "HELLO WORLD "
+
+
+def test_invoke_multi_step_plan():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (text) skill
+    skill = TextSkill()
+    skill_config_dict = kernel.import_skill(skill, "text")
+    test_function = skill_config_dict["uppercase"]
+    test_function2 = skill_config_dict["trim_end"]
+
+    # setup context
+    context = kernel.create_new_context()
+    context["input"] = "hello world "
+
+    plan = Plan(name="test")
+    new_step = Plan(name="test", function=test_function)
+    new_step2 = Plan(name="test", function=test_function2)
+    plan.add_steps([new_step, new_step2])
+    result = plan.invoke(context=context)
+    assert result.result == "HELLO WORLD"
+
+
+@pytest.mark.asyncio
+async def test_invoke_multi_step_plan_async():
+    # create a kernel
+    kernel = sk.Kernel()
+
+    # import test (text) skill
+    skill = TextSkill()
+    skill_config_dict = kernel.import_skill(skill, "text")
+    test_function = skill_config_dict["uppercase"]
+    test_function2 = skill_config_dict["trim_end"]
+
+    # setup context
+    context = kernel.create_new_context()
+    context["input"] = "hello world "
+
+    plan = Plan(name="test")
+    new_step = Plan(name="test", function=test_function)
+    new_step2 = Plan(name="test", function=test_function2)
+    plan.add_steps([new_step, new_step2])
+    result = await plan.invoke_async(context=context)
+    assert result.result == "HELLO WORLD"