Python: Allow for factory callbacks in the process framework

[moonbox3]

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:

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

The bstep_factory looks like (along with its corresponding step)

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.

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 Python: handle complex dependencies with Dapr runtime #10409

Contribution Checklist

• The code builds clean without any errors or warnings
• The PR follows the SK Contribution Guidelines and the pre-submission formatting script raises no violations
• All unit tests pass, and I have added new tests where possible
• I didn't break anyone 😄

[markwallace-microsoft]

Python Test Coverage Report •

File	Stmts	Miss	Cover	Missing
semantic_kernel/processes
process_builder.py	77	44	43%	60–73, 77–79, 85–95, 99–108, 112–118, 122–126, 131, 135–140
semantic_kernel/processes/dapr_runtime
dapr_actor_registration.py	29	19	34%	25–40, 48–53, 61–66
semantic_kernel/processes/dapr_runtime/actors
process_actor.py	244	113	54%	72–76, 81, 84, 90, 93, 97, 103, 106, 122–125, 130, 143, 159–163, 168, 178, 183, 192–200, 204–206, 214–222, 229, 245–306, 311, 322, 336–344, 347–354, 358–386, 417–430
step_actor.py	259	129	50%	81–85, 90, 93, 97–98, 110–112, 156–159, 173–175, 183, 187–188, 214, 222–223, 239–246, 253–254, 262–358, 362, 366–369, 373–389, 393–412, 416–434, 438–441, 445, 449–452
semantic_kernel/processes/kernel_process
kernel_process.py	24	1	96%	58
semantic_kernel/processes/local_runtime
local_kernel_process_context.py	32	2	94%	67–68
local_process.py	138	53	62%	66, 103, 113, 131–142, 176–203, 207–212, 216, 220–226, 230–240, 244–245
local_step.py	172	106	38%	64, 74, 83–171, 175, 179, 183–184, 189–259, 263–266, 270–273, 277–286, 292–295, 299–301
TOTAL	18530	2169	88%

Python Unit Test Overview

Tests	Skipped	Failures	Errors	Time
3174	4 💤	0 ❌	0 🔥	1m 39s ⏱️