Python: Allow for factory callbacks in the process framework (microsoft#10451)

### Motivation and Context

In the current Python process framework, in runtimes like Dapr, there is
no easy way to pass complex (unserializable) dependencies to a step --
this includes things like a ChatCompletion service or an agent of type
ChatCompletionAgent.

Similar to how the kernel dependency was propagated to the step_actor or
process_actor, we're introducing the ability to specify a factory
callback that will be called as the step is instantiated. The factory is
created, if specified via the optional kwarg when adding a step to the
process builder like:

```python
myBStep = process.add_step(step_type=BStep, factory_function=bstep_factory)
```

The `bstep_factory` looks like (along with its corresponding step)

```python
async def bstep_factory():
    """Creates a BStep instance with ephemeral references like ChatCompletionAgent."""
    kernel = Kernel()
    kernel.add_service(AzureChatCompletion())

    agent = ChatCompletionAgent(kernel=kernel, name="echo", instructions="repeat the input back")
    step_instance = BStep()
    step_instance.agent = agent

    return step_instance

class BStep(KernelProcessStep):
    """A sample BStep that optionally holds a ChatCompletionAgent.

    By design, the agent is ephemeral (not stored in state).
    """

    # Ephemeral references won't be persisted to Dapr
    # because we do not place them in a step state model.
    # We'll set this in the factory function:
    agent: ChatCompletionAgent | None = None

    @kernel_function(name="do_it")
    async def do_it(self, context: KernelProcessStepContext):
        print("##### BStep ran (do_it).")
        await asyncio.sleep(2)

        if self.agent:
            history = ChatHistory()
            history.add_user_message("Hello from BStep!")
            async for msg in self.agent.invoke(history):
                print(f"BStep got agent response: {msg.content}")

        await context.emit_event(process_event="BStepDone", data="I did B")
```

Although this isn't explicitly necessary with the local runtime, the
factory callback will also work, if desired.

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

Adds the ability to specify a factory callback for a step in the process
framework.
- Adjusts the Dapr FastAPI demo sample to show how one can include a
dependency like a `ChatCompletionAgent` and use the factory callback for
`BStep`. Although the output from the agent isn't needed, it
demonstrates the capability to handle these types of dependencies while
running Dapr.
- Adds unit tests
- Closes microsoft#10409

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

Author: moonbox3

python/samples/demos/process_with_dapr/fastapi_app.py

@@ -11,10 +11,7 @@
 from samples.demos.process_with_dapr.process.process import get_process
 from samples.demos.process_with_dapr.process.steps import CommonEvents
 from semantic_kernel import Kernel
-from semantic_kernel.processes.dapr_runtime import (
-    register_fastapi_dapr_actors,
-    start,
-)
+from semantic_kernel.processes.dapr_runtime import register_fastapi_dapr_actors, start
 
 logging.basicConfig(level=logging.ERROR)
 
@@ -34,12 +31,16 @@
 # and returns the actor instance with the kernel injected.              #
 #########################################################################
 
+# Get the process which means we have the `KernelProcess` object
+# along with any defined step factories
+process = get_process()
+
 
 # Define a lifespan method that registers the actors with the Dapr runtime
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     print("## actor startup ##")
-    await register_fastapi_dapr_actors(actor, kernel)
+    await register_fastapi_dapr_actors(actor, kernel, process.factories)
     yield
 
 
@@ -56,8 +57,6 @@ async def healthcheck():
 @app.get("/processes/{process_id}")
 async def start_process(process_id: str):
     try:
-        process = get_process()
-
         _ = await start(
             process=process,
             kernel=kernel,

python/samples/demos/process_with_dapr/process/process.py

@@ -2,7 +2,15 @@
 
 from typing import TYPE_CHECKING
 
-from samples.demos.process_with_dapr.process.steps import AStep, BStep, CommonEvents, CStep, CStepState, KickOffStep
+from samples.demos.process_with_dapr.process.steps import (
+    AStep,
+    BStep,
+    CommonEvents,
+    CStep,
+    CStepState,
+    KickOffStep,
+    bstep_factory,
+)
 from semantic_kernel.processes import ProcessBuilder
 
 if TYPE_CHECKING:
@@ -16,7 +24,7 @@ def get_process() -> "KernelProcess":
     # Add the step types to the builder
     kickoff_step = process.add_step(step_type=KickOffStep)
     myAStep = process.add_step(step_type=AStep)
-    myBStep = process.add_step(step_type=BStep)
+    myBStep = process.add_step(step_type=BStep, factory_function=bstep_factory)
 
     # Initialize the CStep with an initial state and the state's current cycle set to 1
     myCStep = process.add_step(step_type=CStep, initial_state=CStepState(current_cycle=1))

python/samples/demos/process_with_dapr/process/steps.py

@@ -6,7 +6,11 @@
 
 from pydantic import Field
 
+from semantic_kernel.agents.chat_completion.chat_completion_agent import ChatCompletionAgent
+from semantic_kernel.connectors.ai.open_ai.services.azure_chat_completion import AzureChatCompletion
+from semantic_kernel.contents.chat_history import ChatHistory
 from semantic_kernel.functions import kernel_function
+from semantic_kernel.kernel import Kernel
 from semantic_kernel.kernel_pydantic import KernelBaseModel
 from semantic_kernel.processes.kernel_process import (
     KernelProcessStep,
@@ -52,14 +56,43 @@ async def do_it(self, context: KernelProcessStepContext):
         await context.emit_event(process_event=CommonEvents.AStepDone, data="I did A")
 
 
-# Define a sample `BStep` step that will emit an event after 2 seconds.
-# The event will be sent to the `CStep` step with the data `I did B`.
+# Define a simple factory for the BStep that can create the dependency that the BStep requires
+# As an example, this factory creates a kernel and adds an `AzureChatCompletion` service to it.
+async def bstep_factory():
+    """Creates a BStep instance with ephemeral references like ChatCompletionAgent."""
+    kernel = Kernel()
+    kernel.add_service(AzureChatCompletion())
+
+    agent = ChatCompletionAgent(kernel=kernel, name="echo", instructions="repeat the input back")
+    step_instance = BStep()
+    step_instance.agent = agent
+
+    return step_instance
+
+
 class BStep(KernelProcessStep):
-    @kernel_function()
+    """A sample BStep that optionally holds a ChatCompletionAgent.
+
+    By design, the agent is ephemeral (not stored in state).
+    """
+
+    # Ephemeral references won't be persisted to Dapr
+    # because we do not place them in a step state model.
+    # We'll set this in the factory function:
+    agent: ChatCompletionAgent | None = None
+
+    @kernel_function(name="do_it")
     async def do_it(self, context: KernelProcessStepContext):
-        print("##### BStep ran.")
+        print("##### BStep ran (do_it).")
         await asyncio.sleep(2)
-        await context.emit_event(process_event=CommonEvents.BStepDone, data="I did B")
+
+        if self.agent:
+            history = ChatHistory()
+            history.add_user_message("Hello from BStep!")
+            async for msg in self.agent.invoke(history):
+                print(f"BStep got agent response: {msg.content}")
+
+        await context.emit_event(process_event="BStepDone", data="I did B")
 
 
 # Define a sample `CStepState` that will keep track of the current cycle.

python/semantic_kernel/processes/dapr_runtime/actors/process_actor.py

@@ -5,6 +5,7 @@
 import json
 import logging
 import uuid
+from collections.abc import Callable
 from queue import Queue
 from typing import Any
 
@@ -46,16 +47,18 @@
 class ProcessActor(StepActor, ProcessInterface):
     """A local process that contains a collection of steps."""
 
-    def __init__(self, ctx: ActorRuntimeContext, actor_id: ActorId, kernel: Kernel):
+    def __init__(self, ctx: ActorRuntimeContext, actor_id: ActorId, kernel: Kernel, factories: dict[str, Callable]):
         """Initializes a new instance of ProcessActor.
 
         Args:
             ctx: The actor runtime context.
             actor_id: The unique ID for the actor.
             kernel: The Kernel dependency to be injected.
+            factories: The factory dictionary that contains step types to factory methods.
         """
-        super().__init__(ctx, actor_id, kernel)
+        super().__init__(ctx, actor_id, kernel, factories)
         self.kernel = kernel
+        self.factories = factories
         self.steps: list[StepInterface] = []
         self.step_infos: list[DaprStepInfo] = []
         self.initialize_task: bool | None = False

python/semantic_kernel/processes/dapr_runtime/actors/step_actor.py

@@ -4,6 +4,8 @@
 import importlib
 import json
 import logging
+from collections.abc import Callable
+from inspect import isawaitable
 from queue import Queue
 from typing import Any
 
@@ -46,16 +48,18 @@
 class StepActor(Actor, StepInterface, KernelProcessMessageChannel):
     """Represents a step actor that follows the Step abstract class."""
 
-    def __init__(self, ctx: ActorRuntimeContext, actor_id: ActorId, kernel: Kernel):
+    def __init__(self, ctx: ActorRuntimeContext, actor_id: ActorId, kernel: Kernel, factories: dict[str, Callable]):
         """Initializes a new instance of StepActor.
 
         Args:
             ctx: The actor runtime context.
             actor_id: The unique ID for the actor.
             kernel: The Kernel dependency to be injected.
+            factories: The factory dictionary to use for creating the step.
         """
         super().__init__(ctx, actor_id)
         self.kernel = kernel
+        self.factories: dict[str, Callable] = factories
         self.parent_process_id: str | None = None
         self.step_info: DaprStepInfo | None = None
         self.initialize_task: bool | None = False
@@ -172,31 +176,38 @@ def _get_class_from_string(self, full_class_name: str):
 
     async def activate_step(self):
         """Initializes the step."""
-        # Instantiate an instance of the inner step object
-        step_cls = self._get_class_from_string(self.inner_step_type)
-
-        step_instance: KernelProcessStep = step_cls()  # type: ignore
+        # Instantiate an instance of the inner step object and retrieve its class reference.
+        if self.factories and self.inner_step_type in self.factories:
+            step_object = self.factories[self.inner_step_type]()
+            if isawaitable(step_object):
+                step_object = await step_object
+            step_cls = step_object.__class__
+            step_instance: KernelProcessStep = step_object  # type: ignore
+        else:
+            step_cls = self._get_class_from_string(self.inner_step_type)
+            step_instance: KernelProcessStep = step_cls()  # type: ignore
 
         kernel_plugin = self.kernel.add_plugin(
-            step_instance, self.step_info.state.name if self.step_info.state else "default_name"
+            step_instance,
+            self.step_info.state.name if self.step_info.state else "default_name",
         )
 
-        # Load the kernel functions
+        # Load the kernel functions.
         for name, f in kernel_plugin.functions.items():
             self.functions[name] = f
 
-        # Initialize the input channels
+        # Initialize the input channels.
         self.initial_inputs = find_input_channels(channel=self, functions=self.functions)
         self.inputs = {k: {kk: vv for kk, vv in v.items()} if v else {} for k, v in self.initial_inputs.items()}
 
-        # Use the existing state or create a new one if not provided
+        # Use the existing state or create a new one if not provided.
         state_object = self.step_info.state
 
-        # Extract TState from inner_step_type
+        # Extract TState from inner_step_type using the class reference.
         t_state = get_generic_state_type(step_cls)
 
         if t_state is not None:
-            # Create state_type as KernelProcessStepState[TState]
+            # Create state_type as KernelProcessStepState[TState].
             state_type = KernelProcessStepState[t_state]
 
             if state_object is None:
@@ -206,7 +217,7 @@ async def activate_step(self):
                     state=None,
                 )
             else:
-                # Make sure state_object is an instance of state_type
+                # Ensure that state_object is an instance of the expected type.
                 if not isinstance(state_object, KernelProcessStepState):
                     error_message = "State object is not of the expected type."
                     raise KernelException(error_message)
@@ -215,25 +226,22 @@ async def activate_step(self):
                     ActorStateKeys.StepStateType.value,
                     get_fully_qualified_name(t_state),
                 )
-
                 await self._state_manager.try_add_state(
                     ActorStateKeys.StepStateJson.value,
                     json.dumps(state_object.model_dump()),
                 )
-
                 await self._state_manager.save_state()
 
-            # Make sure that state_object.state is not None
+            # Initialize state_object.state if it is not already set.
             if state_object.state is None:
                 try:
                     state_object.state = t_state()
                 except Exception as e:
                     error_message = f"Cannot instantiate state of type {t_state}: {e}"
                     raise KernelException(error_message)
         else:
-            # The step has no user-defined state; use the base KernelProcessStepState
+            # The step has no user-defined state; use the base KernelProcessStepState.
             state_type = KernelProcessStepState
-
             if state_object is None:
                 state_object = state_type(
                     name=step_cls.__name__,
@@ -245,7 +253,7 @@ async def activate_step(self):
             error_message = "The state object for the KernelProcessStep could not be created."
             raise KernelException(error_message)
 
-        # Set the step state and activate the step with the state object
+        # Set the step state and activate the step with the state object.
         self.step_state = state_object
         await step_instance.activate(state_object)
 

python/semantic_kernel/processes/dapr_runtime/dapr_actor_registration.py

@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+from collections.abc import Callable
 from typing import TYPE_CHECKING
 
 from dapr.actor import ActorId
@@ -19,22 +20,32 @@
     from semantic_kernel.kernel import Kernel
 
 
-def create_actor_factories(kernel: "Kernel") -> tuple:
+def create_actor_factories(kernel: "Kernel", factories: dict[str, Callable] | None = None) -> tuple:
     """Creates actor factories for ProcessActor and StepActor."""
+    if factories is None:
+        factories = {}
 
-    def process_actor_factory(ctx: ActorRuntimeContext, actor_id: ActorId) -> ProcessActor:
-        return ProcessActor(ctx, actor_id, kernel)
+    def process_actor_factory(
+        ctx: ActorRuntimeContext,
+        actor_id: ActorId,
+    ) -> ProcessActor:
+        return ProcessActor(ctx, actor_id, kernel=kernel, factories=factories)
 
-    def step_actor_factory(ctx: ActorRuntimeContext, actor_id: ActorId) -> StepActor:
-        return StepActor(ctx, actor_id, kernel=kernel)
+    def step_actor_factory(
+        ctx: ActorRuntimeContext,
+        actor_id: ActorId,
+    ) -> StepActor:
+        return StepActor(ctx, actor_id, kernel=kernel, factories=factories)
 
     return process_actor_factory, step_actor_factory
 
 
 # Asynchronous registration for FastAPI
-async def register_fastapi_dapr_actors(actor: FastAPIDaprActor, kernel: "Kernel") -> None:
+async def register_fastapi_dapr_actors(
+    actor: FastAPIDaprActor, kernel: "Kernel", factories: dict[str, Callable] | None = None
+) -> None:
     """Registers the actors with the Dapr runtime for use with a FastAPI app."""
-    process_actor_factory, step_actor_factory = create_actor_factories(kernel)
+    process_actor_factory, step_actor_factory = create_actor_factories(kernel, factories)
     await actor.register_actor(ProcessActor, actor_factory=process_actor_factory)
     await actor.register_actor(StepActor, actor_factory=step_actor_factory)
     await actor.register_actor(EventBufferActor)
@@ -43,9 +54,11 @@ async def register_fastapi_dapr_actors(actor: FastAPIDaprActor, kernel: "Kernel"
 
 
 # Synchronous registration for Flask
-def register_flask_dapr_actors(actor: FlaskDaprActor, kernel: "Kernel") -> None:
+def register_flask_dapr_actors(
+    actor: FlaskDaprActor, kernel: "Kernel", factory: dict[str, Callable] | None = None
+) -> None:
     """Registers the actors with the Dapr runtime for use with a Flask app."""
-    process_actor_factory, step_actor_factory = create_actor_factories(kernel)
+    process_actor_factory, step_actor_factory = create_actor_factories(kernel, factory)
     actor.register_actor(ProcessActor, actor_factory=process_actor_factory)
     actor.register_actor(StepActor, actor_factory=step_actor_factory)
     actor.register_actor(EventBufferActor)

python/semantic_kernel/processes/kernel_process/kernel_process.py

@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+from collections.abc import Callable
 from typing import TYPE_CHECKING, Any
 
 from pydantic import Field
@@ -18,14 +19,24 @@ class KernelProcess(KernelProcessStepInfo):
     """A kernel process."""
 
     steps: list[KernelProcessStepInfo] = Field(default_factory=list)
+    factories: dict[str, Callable] = Field(default_factory=dict)
 
     def __init__(
         self,
         state: KernelProcessState,
         steps: list[KernelProcessStepInfo],
         edges: dict[str, list["KernelProcessEdge"]] | None = None,
+        factories: dict[str, Callable] | None = None,
     ):
-        """Initialize the kernel process."""
+        """Initialize the kernel process.
+
+        Args:
+            state: The state of the process.
+            steps: The steps of the process.
+            edges: The edges of the process. Defaults to None.
+            factories: The factories of the process. This allows for the creation of
+                steps that require complex dependencies that cannot be JSON serialized or deserialized.
+        """
         if not state:
             raise ValueError("state cannot be None")
         if not steps:
@@ -43,4 +54,7 @@ def __init__(
             "output_edges": edges or {},
         }
 
+        if factories:
+            args["factories"] = factories
+
         super().__init__(**args)

python/semantic_kernel/processes/local_runtime/local_kernel_process_context.py

@@ -33,6 +33,7 @@ def __init__(self, process: "KernelProcess", kernel: "Kernel"):
             process=process,
             kernel=kernel,
             parent_process_id=None,
+            factories=process.factories,
         )
 
         super().__init__(local_process=local_process)

python/semantic_kernel/processes/local_runtime/local_process.py

@@ -4,6 +4,7 @@
 import contextlib
 import logging
 import uuid
+from collections.abc import Callable
 from queue import Queue
 from typing import TYPE_CHECKING, Any
 
@@ -42,8 +43,15 @@ class LocalProcess(LocalStep):
     initialize_task: bool | None = False
     external_event_queue: Queue = Field(default_factory=Queue)
     process_task: asyncio.Task | None = None
-
-    def __init__(self, process: "KernelProcess", kernel: Kernel, parent_process_id: str | None = None):
+    factories: dict[str, Callable] = Field(default_factory=dict)
+
+    def __init__(
+        self,
+        process: "KernelProcess",
+        kernel: Kernel,
+        factories: dict[str, Callable] | None = None,
+        parent_process_id: str | None = None,
+    ):
         """Initializes the local process."""
         args: dict[str, Any] = {
             "step_info": process,
@@ -54,6 +62,9 @@ def __init__(self, process: "KernelProcess", kernel: Kernel, parent_process_id:
             "initialize_task": False,
         }
 
+        if factories:
+            args["factories"] = factories
+
         super().__init__(**args)
 
     def ensure_initialized(self):
@@ -124,6 +135,7 @@ def initialize_process(self):
                 process = LocalProcess(
                     process=step,
                     kernel=self.kernel,
+                    factorie=self.factories,
                     parent_process_id=self.id,
                 )
 
@@ -136,6 +148,7 @@ def initialize_process(self):
                 local_step = LocalStep(
                     step_info=step,
                     kernel=self.kernel,
+                    factories=self.factories,
                     parent_process_id=self.id,
                 )
 

python/semantic_kernel/processes/local_runtime/local_step.py

@@ -3,6 +3,8 @@
 import asyncio
 import logging
 import uuid
+from collections.abc import Callable
+from inspect import isawaitable
 from queue import Queue
 from typing import Any
 
@@ -19,13 +21,12 @@
 from semantic_kernel.processes.kernel_process.kernel_process_edge import KernelProcessEdge
 from semantic_kernel.processes.kernel_process.kernel_process_event import KernelProcessEvent
 from semantic_kernel.processes.kernel_process.kernel_process_message_channel import KernelProcessMessageChannel
-from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep
 from semantic_kernel.processes.kernel_process.kernel_process_step_info import KernelProcessStepInfo
 from semantic_kernel.processes.kernel_process.kernel_process_step_state import KernelProcessStepState
 from semantic_kernel.processes.local_runtime.local_event import LocalEvent
 from semantic_kernel.processes.local_runtime.local_message import LocalMessage
 from semantic_kernel.processes.process_types import get_generic_state_type
-from semantic_kernel.processes.step_utils import find_input_channels
+from semantic_kernel.processes.step_utils import find_input_channels, get_fully_qualified_name
 from semantic_kernel.utils.experimental_decorator import experimental_class
 
 logger: logging.Logger = logging.getLogger(__name__)
@@ -47,6 +48,7 @@ class LocalStep(KernelProcessMessageChannel, KernelBaseModel):
     output_edges: dict[str, list[KernelProcessEdge]] = Field(default_factory=dict)
     parent_process_id: str | None = None
     init_lock: asyncio.Lock = Field(default_factory=asyncio.Lock, exclude=True)
+    factories: dict[str, Callable]
 
     @model_validator(mode="before")
     @classmethod
@@ -185,8 +187,16 @@ async def initialize_step(self):
         """Initializes the step."""
         # Instantiate an instance of the inner step object
         step_cls = self.step_info.inner_step_type
-
-        step_instance: KernelProcessStep = step_cls()  # type: ignore
+        factory = (
+            self.factories.get(get_fully_qualified_name(self.step_info.inner_step_type)) if self.factories else None
+        )
+        if factory:
+            step_instance = factory()
+            if isawaitable(step_instance):
+                step_instance = await step_instance
+            step_cls = type(step_instance)
+        else:
+            step_instance = step_cls()  # type: ignore
 
         kernel_plugin = self.kernel.add_plugin(
             step_instance, self.step_info.state.name if self.step_info.state else "default_name"

python/semantic_kernel/processes/process_builder.py

@@ -2,6 +2,7 @@
 
 import contextlib
 import inspect
+from collections.abc import Callable
 from copy import copy
 from enum import Enum
 from typing import TYPE_CHECKING
@@ -17,6 +18,7 @@
 from semantic_kernel.processes.process_step_builder import ProcessStepBuilder
 from semantic_kernel.processes.process_step_edge_builder import ProcessStepEdgeBuilder
 from semantic_kernel.processes.process_types import TState, TStep
+from semantic_kernel.processes.step_utils import get_fully_qualified_name
 from semantic_kernel.utils.experimental_decorator import experimental_class
 
 if TYPE_CHECKING:
@@ -32,20 +34,38 @@ class ProcessBuilder(ProcessStepBuilder):
     has_parent_process: bool = False
 
     steps: list["ProcessStepBuilder"] = Field(default_factory=list)
+    factories: dict[str, Callable] = Field(default_factory=dict)
 
     def add_step(
         self,
         step_type: type[TStep],
         name: str | None = None,
         initial_state: TState | None = None,
+        factory_function: Callable | None = None,
         **kwargs,
     ) -> ProcessStepBuilder[TState, TStep]:
-        """Register a step type with optional constructor arguments."""
+        """Register a step type with optional constructor arguments.
+
+        Args:
+            step_type: The step type.
+            name: The name of the step. Defaults to None.
+            initial_state: The initial state of the step. Defaults to None.
+            factory_function: The factory function. Allows for a callable that is used to create the step instance
+                that may have complex dependencies that cannot be JSON serialized or deserialized. Defaults to None.
+            kwargs: Additional keyword arguments.
+
+        Returns:
+            The process step builder.
+        """
         if not inspect.isclass(step_type):
             raise ProcessInvalidConfigurationException(
                 f"Expected a class type, but got an instance of {type(step_type).__name__}"
             )
 
+        if factory_function:
+            fq_name = get_fully_qualified_name(step_type)
+            self.factories[fq_name] = factory_function
+
         name = name or step_type.__name__
         process_step_builder = ProcessStepBuilder(type=step_type, name=name, initial_state=initial_state, **kwargs)
         self.steps.append(process_step_builder)
@@ -117,4 +137,4 @@ def build(self) -> "KernelProcess":
         built_edges = {key: [edge.build() for edge in edges] for key, edges in self.edges.items()}
         built_steps = [step.build_step() for step in self.steps]
         process_state = KernelProcessState(name=self.name, id=self.id if self.has_parent_process else None)
-        return KernelProcess(state=process_state, steps=built_steps, edges=built_edges)
+        return KernelProcess(state=process_state, steps=built_steps, edges=built_edges, factories=self.factories)

python/tests/unit/processes/dapr_runtime/test_process_actor.py

@@ -27,7 +27,7 @@ def actor_context():
         actor_client=MagicMock(),
     )
     kernel_mock = MagicMock()
-    actor = ProcessActor(runtime_context, actor_id, kernel=kernel_mock)
+    actor = ProcessActor(runtime_context, actor_id, kernel=kernel_mock, factories={})
 
     actor._state_manager = AsyncMock()
     actor._state_manager.try_add_state = AsyncMock(return_value=True)

python/tests/unit/processes/dapr_runtime/test_step_actor.py

@@ -6,18 +6,28 @@
 import pytest
 from dapr.actor import ActorId
 
+from semantic_kernel.processes.dapr_runtime.actors.actor_state_key import ActorStateKeys
 from semantic_kernel.processes.dapr_runtime.actors.step_actor import StepActor
 from semantic_kernel.processes.dapr_runtime.dapr_step_info import DaprStepInfo
 from semantic_kernel.processes.kernel_process.kernel_process_step_state import KernelProcessStepState
 from semantic_kernel.processes.process_message import ProcessMessage
 
 
+class FakeStep:
+    async def activate(self, state):
+        self.activated_state = state
+
+
+class FakeState:
+    pass
+
+
 @pytest.fixture
 def actor_context():
     ctx = MagicMock()
     actor_id = ActorId("test_actor")
     kernel = MagicMock()
-    return StepActor(ctx, actor_id, kernel)
+    return StepActor(ctx, actor_id, kernel, factories={})
 
 
 async def test_initialize_step(actor_context):
@@ -97,3 +107,98 @@ async def test_process_incoming_messages(actor_context):
         expected_messages = []
         expected_messages = [json.dumps(msg.model_dump()) for msg in list(actor_context.incoming_messages.queue)]
         mock_try_add_state.assert_any_call("incomingMessagesState", expected_messages)
+
+
+async def test_activate_step_with_factory_creates_state(actor_context):
+    fake_step_instance = FakeStep()
+    fake_step_instance.activate = AsyncMock(side_effect=fake_step_instance.activate)
+
+    fake_plugin = MagicMock()
+    fake_plugin.functions = {"test_function": lambda x: x}
+
+    with (
+        patch(
+            "semantic_kernel.processes.dapr_runtime.actors.step_actor.get_generic_state_type",
+            return_value=FakeState,
+        ),
+        patch(
+            "semantic_kernel.processes.dapr_runtime.actors.step_actor.get_fully_qualified_name",
+            return_value="FakeStateFullyQualified",
+        ),
+        patch(
+            "semantic_kernel.processes.dapr_runtime.actors.step_actor.find_input_channels",
+            return_value={"channel": {"input": "value"}},
+        ),
+    ):
+        actor_context.factories = {"FakeStep": lambda: fake_step_instance}
+        actor_context.inner_step_type = "FakeStep"
+        actor_context.step_info = DaprStepInfo(
+            state=KernelProcessStepState(name="default_name", id="step_123"),
+            inner_step_python_type="FakeStep",
+            edges={},
+        )
+        actor_context.kernel.add_plugin = MagicMock(return_value=fake_plugin)
+        actor_context._state_manager.try_add_state = AsyncMock()
+        actor_context._state_manager.save_state = AsyncMock()
+
+        await actor_context.activate_step()
+
+        actor_context.kernel.add_plugin.assert_called_once_with(fake_step_instance, "default_name")
+        assert actor_context.functions == fake_plugin.functions
+        assert actor_context.initial_inputs == {"channel": {"input": "value"}}
+        assert actor_context.inputs == {"channel": {"input": "value"}}
+        assert actor_context.step_state is not None
+        assert isinstance(actor_context.step_state.state, FakeState)
+        fake_step_instance.activate.assert_awaited_once_with(actor_context.step_state)
+
+
+async def test_activate_step_with_factory_uses_existing_state(actor_context):
+    fake_step_instance = FakeStep()
+    fake_step_instance.activate = AsyncMock(side_effect=fake_step_instance.activate)
+
+    fake_plugin = MagicMock()
+    fake_plugin.functions = {"test_function": lambda x: x}
+
+    pre_existing_state = KernelProcessStepState(name="ExistingState", id="ExistingState", state=None)
+
+    with (
+        patch.object(
+            KernelProcessStepState,
+            "model_dump",
+            return_value={"name": "ExistingState", "id": "ExistingState", "state": None},
+        ),
+        patch(
+            "semantic_kernel.processes.dapr_runtime.actors.step_actor.get_generic_state_type",
+            return_value=FakeState,
+        ),
+        patch(
+            "semantic_kernel.processes.dapr_runtime.actors.step_actor.get_fully_qualified_name",
+            return_value="FakeStateFullyQualified",
+        ),
+        patch(
+            "semantic_kernel.processes.dapr_runtime.actors.step_actor.find_input_channels",
+            return_value={"channel": {"input": "value"}},
+        ),
+    ):
+        actor_context.factories = {"FakeStep": lambda: fake_step_instance}
+        actor_context.inner_step_type = "FakeStep"
+        actor_context.step_info = DaprStepInfo(state=pre_existing_state, inner_step_python_type="FakeStep", edges={})
+        actor_context.kernel.add_plugin = MagicMock(return_value=fake_plugin)
+        actor_context._state_manager.try_add_state = AsyncMock()
+        actor_context._state_manager.save_state = AsyncMock()
+
+        await actor_context.activate_step()
+
+        actor_context.kernel.add_plugin.assert_called_once_with(fake_step_instance, pre_existing_state.name)
+        assert actor_context.functions == fake_plugin.functions
+        assert actor_context.initial_inputs == {"channel": {"input": "value"}}
+        assert actor_context.inputs == {"channel": {"input": "value"}}
+        actor_context._state_manager.try_add_state.assert_any_await(
+            ActorStateKeys.StepStateType.value, "FakeStateFullyQualified"
+        )
+        actor_context._state_manager.try_add_state.assert_any_await(
+            ActorStateKeys.StepStateJson.value, json.dumps(pre_existing_state.model_dump())
+        )
+        actor_context._state_manager.save_state.assert_awaited_once()
+        assert isinstance(actor_context.step_state.state, FakeState)
+        fake_step_instance.activate.assert_awaited_once_with(actor_context.step_state)

python/tests/unit/processes/local_runtime/test_local_kernel_process_context.py

@@ -26,6 +26,7 @@ def mock_process():
     process = MagicMock(spec=KernelProcess)
     process.state = state
     process.steps = [step_info]
+    process.factories = {}
     return process
 
 

python/tests/unit/processes/local_runtime/test_local_process.py

@@ -271,6 +271,7 @@ def test_initialize_process(mock_process, mock_kernel, build_model):
                 mock_local_step_init.assert_called_with(
                     step_info=step_info,
                     kernel=mock_kernel,
+                    factories={},
                     parent_process_id=local_process.id,
                 )