mirror of
https://github.com/microsoft/semantic-kernel.git
synced 2026-10-12 02:19:06 +00:00
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 😄 --------- Co-authored-by: Shawn Callegari <[email protected]>
This commit is contained in:
co-authored by
Shawn Callegari
parent
43b62e56cc
commit
5e4e0edfaa
@@ -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
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user