Python: Support callers passing in max supersteps for processes (#11798)

### Motivation and Context

Today in Python processes the `max_supersteps` config is hidden from the
caller. The internal execute method defaults the value to 100, and this
doesn't allow for developers to leverage custom values depending on
their use cases.

<!-- 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

Provide a way for callers to pass in `max_supersteps` via the process
runtime `start` method.
- Update README
- Update unit tests accordingly. 

<!-- 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: -->

- [X] The code builds clean without any errors or warnings
- [X] 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
- [X] All unit tests pass, and I have added new tests where possible
- [X] I didn't break anyone 😄
This commit is contained in:
Evan Mattson
2025-04-30 09:25:47 +00:00
committed by GitHub
parent ee8050dc64
commit d3404ccc28
10 changed files with 123 additions and 16 deletions
@@ -221,6 +221,25 @@ Similar to the Semantic Kernel Python concept samples, it is necessary to config
and keys used by the kernel. See the follow "Configuring the Kernel" [guide](../concepts/README.md#configuring-the-kernel) for
more information.
## Configuring Max Supersteps
Process execution is run with a configuration of `max_supersteps`. By default, the `max_supersteps` is configured at `100`.
To adjust the value, pass the `max_supersteps` keyword argument to the `start` method for the given runtime:
```python
from semantic_kernel.processes.local_runtime.local_kernel_process import start
async with await start(
process=kernel_process,
kernel=kernel,
initial_event=KernelProcessEvent(id=CommonEvents.StartProcess),
max_supersteps=50, # Configure the max number of supersteps for process run
) as process_context:
process_state = await process_context.get_state()
# Handle process state...
```
## Running Concept Samples
Concept samples can be run in an IDE or via the command line. After setting up the required api key
@@ -47,6 +47,8 @@ logger: logging.Logger = logging.getLogger(__name__)
class ProcessActor(StepActor, ProcessInterface):
"""A local process that contains a collection of steps."""
max_supersteps: int = 100
def __init__(self, ctx: ActorRuntimeContext, actor_id: ActorId, kernel: Kernel, factories: dict[str, Callable]):
"""Initializes a new instance of ProcessActor.
@@ -85,10 +87,14 @@ class ProcessActor(StepActor, ProcessInterface):
process_info_data = input.get("process_info")
parent_process_id = input.get("parent_process_id")
max_supersteps = input.get("max_supersteps", None)
if process_info_data is None:
raise ValueError("The process info is not defined.")
if max_supersteps is not None:
self.max_supersteps = max_supersteps
if isinstance(process_info_data, str):
process_info_dict = json.loads(process_info_data)
elif isinstance(process_info_data, dict):
@@ -130,7 +136,9 @@ class ProcessActor(StepActor, ProcessInterface):
# Only create the task if it doesn't already exist or is not running
if not self.process_task or self.process_task.done():
self.process_task = asyncio.create_task(self.internal_execute(keep_alive=keep_alive))
self.process_task = asyncio.create_task(
self.internal_execute(max_supersteps=self.max_supersteps, keep_alive=keep_alive)
)
async def run_once(self, process_event: KernelProcessEvent | str | None) -> None:
"""Starts the process with an initial event and waits for it to finish.
@@ -325,6 +333,8 @@ class ProcessActor(StepActor, ProcessInterface):
async def internal_execute(self, max_supersteps: int = 100, keep_alive: bool = True):
"""Internal execution logic for the process."""
logger.debug(f"Running process for {max_supersteps} supersteps.")
try:
for _ in range(max_supersteps):
if await self._is_end_message_sent():
@@ -17,9 +17,19 @@ async def start(
process: "KernelProcess",
initial_event: KernelProcessEvent | str | Enum,
process_id: str | None = None,
max_supersteps: int | None = None,
**kwargs,
) -> DaprKernelProcessContext:
"""Start the kernel process."""
"""Start the kernel process.
Args:
process: The kernel process to start.
initial_event: The initial event to start the process with.
process_id: The ID of the process. If None, a new ID will be generated.
max_supersteps: The maximum number of supersteps. This is the total number of times process steps will run.
Defaults to None, and thus the process will run its steps 100 times.
**kwargs: Additional keyword arguments.
"""
if process is None:
raise ProcessInvalidConfigurationException("process cannot be None")
if process.state is None:
@@ -37,6 +47,6 @@ async def start(
if process_id is not None:
process.state.id = process_id
process_context = DaprKernelProcessContext(process=process)
process_context = DaprKernelProcessContext(process=process, max_supersteps=max_supersteps)
await process_context.start_with_event(initial_event_str)
return process_context
@@ -18,14 +18,24 @@ class DaprKernelProcessContext:
dapr_process: ProcessInterface
process: KernelProcess
max_supersteps: int = 100
def __init__(self, process: KernelProcess):
"""Initialize a new instance of DaprKernelProcessContext."""
def __init__(self, process: KernelProcess, max_supersteps: int | None = None) -> None:
"""Initialize a new instance of DaprKernelProcessContext.
Args:
process: The kernel process to start.
max_supersteps: The maximum number of supersteps. This is the total number of times process steps will run.
Defaults to None, and thus the process will run its steps 100 times.
"""
if process.state.name is None:
raise ValueError("Process state name must not be None")
if process.state.id is None or process.state.id == "":
process.state.id = str(uuid.uuid4().hex)
if max_supersteps is not None:
self.max_supersteps = max_supersteps
self.process = process
process_id = ActorId(process.state.id)
self.dapr_process = ActorProxy.create( # type: ignore
@@ -42,6 +52,7 @@ class DaprKernelProcessContext:
payload = {
"process_info": dapr_process_dict,
"parent_process_id": None,
"max_supersteps": self.max_supersteps,
}
await self.dapr_process.initialize_process(payload)
@@ -15,9 +15,22 @@ if TYPE_CHECKING:
@experimental
async def start(
process: "KernelProcess", kernel: "Kernel", initial_event: KernelProcessEvent | str | Enum, **kwargs
process: "KernelProcess",
kernel: "Kernel",
initial_event: KernelProcessEvent | str | Enum,
max_supersteps: int | None = None,
**kwargs,
) -> LocalKernelProcessContext:
"""Start the kernel process."""
"""Start the kernel process.
Args:
process: The kernel process to start.
kernel: The kernel instance.
initial_event: The initial event to start the process with.
max_supersteps: The maximum number of supersteps. This is the total number of times process steps will run.
Defaults to None, and thus the process will run its steps 100 times.
**kwargs: Additional keyword arguments.
"""
if process is None:
raise ProcessInvalidConfigurationException("process cannot be None")
if process.state is None or not process.state.name:
@@ -34,6 +47,6 @@ async def start(
if isinstance(initial_event_str, str):
initial_event_str = KernelProcessEvent(id=initial_event_str, data=kwargs.get("data"))
process_context = LocalKernelProcessContext(process, kernel)
process_context = LocalKernelProcessContext(process, kernel, max_supersteps)
await process_context.start_with_event(initial_event_str)
return process_context
@@ -18,8 +18,15 @@ class LocalKernelProcessContext(KernelBaseModel):
local_process: LocalProcess
def __init__(self, process: "KernelProcess", kernel: "Kernel"):
"""Initializes the local kernel process context."""
def __init__(self, process: "KernelProcess", kernel: "Kernel", max_supersteps: int | None = None) -> None:
"""Initializes the local kernel process context.
Args:
process: The kernel process to start.
kernel: The kernel instance.
max_supersteps: The maximum number of supersteps. This is the total number of times process steps will run.
Defaults to None.
"""
from semantic_kernel.processes.kernel_process.kernel_process import KernelProcess # noqa: F401
LocalProcess.model_rebuild()
@@ -34,6 +41,7 @@ class LocalKernelProcessContext(KernelBaseModel):
kernel=kernel,
parent_process_id=None,
factories=process.factories,
max_supersteps=max_supersteps,
)
super().__init__(local_process=local_process) # type: ignore
@@ -44,6 +44,9 @@ class LocalProcess(LocalStep):
external_event_queue: Queue = Field(default_factory=Queue)
process_task: asyncio.Task | None = None
factories: dict[str, Callable] = Field(default_factory=dict)
max_supersteps: int = Field(
default=100, ge=1, description="Maximum number of supersteps to execute before stopping the process."
)
def __init__(
self,
@@ -51,6 +54,7 @@ class LocalProcess(LocalStep):
kernel: Kernel,
factories: dict[str, Callable] | None = None,
parent_process_id: str | None = None,
max_supersteps: int | None = None,
):
"""Initializes the local process."""
args: dict[str, Any] = {
@@ -62,6 +66,9 @@ class LocalProcess(LocalStep):
"initialize_task": False,
}
if max_supersteps is not None:
args["max_supersteps"] = max_supersteps
if factories:
args["factories"] = factories
@@ -73,13 +80,23 @@ class LocalProcess(LocalStep):
self.initialize_process()
self.initialize_task = True
async def start(self, keep_alive: bool = True):
"""Starts the process with an initial event."""
async def start(self, keep_alive: bool = True) -> None:
"""Starts the process with an initial event.
Args:
keep_alive: Indicates if the process should wait for external events after it's finished processing.
"""
self.ensure_initialized()
self.process_task = asyncio.create_task(self.internal_execute(keep_alive=keep_alive))
self.process_task = asyncio.create_task(
self.internal_execute(max_supersteps=self.max_supersteps, keep_alive=keep_alive)
)
async def run_once(self, process_event: KernelProcessEvent):
"""Starts the process with an initial event and waits for it to finish."""
"""Starts the process with an initial event and waits for it to finish.
Args:
process_event: The KernelProcessEvent to start the process with.
"""
if process_event is None:
raise ProcessEventUndefinedException("The process event must be specified.")
self.external_event_queue.put(process_event)
@@ -175,6 +192,8 @@ class LocalProcess(LocalStep):
"""Internal execution logic for the process."""
message_channel: Queue[LocalMessage] = Queue()
logger.debug(f"Running process for {max_supersteps} supersteps.")
try:
for _ in range(max_supersteps):
self.enqueue_external_messages(message_channel)
@@ -14,9 +14,10 @@ from semantic_kernel.processes.kernel_process.kernel_process_step_info import Ke
class FakeDaprKernelProcessContext:
def __init__(self, process):
def __init__(self, process, max_supersteps: int | None = None):
self.process = process
self.start_with_event = AsyncMock()
self.max_supersteps = max_supersteps
async def test_start_with_valid_parameters():
@@ -34,10 +35,11 @@ async def test_start_with_valid_parameters():
"semantic_kernel.processes.dapr_runtime.dapr_kernel_process.DaprKernelProcessContext",
new=FakeDaprKernelProcessContext,
):
result = await start(process=process, kernel=kernel, initial_event=initial_event)
result = await start(process=process, kernel=kernel, initial_event=initial_event, max_supersteps=10)
assert isinstance(result, FakeDaprKernelProcessContext)
assert result.process == process
assert result.max_supersteps == 10
result.start_with_event.assert_called_once_with(initial_event)
@@ -54,6 +54,7 @@ async def test_start_with_event(process_context):
expected_payload = {
"process_info": dapr_process_info.model_dump_json(),
"parent_process_id": None,
"max_supersteps": context.max_supersteps,
}
mock_dapr_process.initialize_process.assert_awaited_once_with(expected_payload)
@@ -80,6 +80,20 @@ def test_initialization(mock_process, mock_kernel, build_model):
assert local_process.external_event_queue.empty()
def test_initialization_max_supersteps(mock_process, mock_kernel, build_model):
# Act
local_process = LocalProcess(process=mock_process, kernel=mock_kernel, max_supersteps=10)
# Assert
assert local_process.process == mock_process
assert local_process.kernel == mock_kernel
assert not local_process.initialize_task
assert len(local_process.step_infos) == len(mock_process.steps)
assert local_process.step_infos[0] == mock_process.steps[0]
assert local_process.external_event_queue.empty()
assert local_process.max_supersteps == 10
def test_ensure_initialized(mock_process, mock_kernel, build_model):
# Arrange
local_process = LocalProcess(process=mock_process, kernel=mock_kernel)