mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: semantic-kernel to agent-framework migration code samples (#1045)
* wip migrations * Wip: workflow migrations * Add migration samples for sk to af * Fix typo * Fixes
This commit is contained in:
Evan Mattson
committed by
GitHub
Unverified
GitHub
parent
498fc06fd6
commit
fb51d917fd
@@ -0,0 +1,252 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Side-by-side sample comparing Semantic Kernel Process Framework and Agent Framework workflows."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import TYPE_CHECKING, ClassVar, cast
|
||||
|
||||
######################################################################
|
||||
# region Agent Framework imports
|
||||
######################################################################
|
||||
from agent_framework import Executor, WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, handler
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
######################################################################
|
||||
# region Semantic Kernel imports
|
||||
######################################################################
|
||||
from semantic_kernel import Kernel
|
||||
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
|
||||
from semantic_kernel.functions import kernel_function
|
||||
from semantic_kernel.processes.kernel_process.kernel_process_event import KernelProcessEvent
|
||||
from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep
|
||||
from semantic_kernel.processes.kernel_process.kernel_process_step_context import KernelProcessStepContext
|
||||
from semantic_kernel.processes.kernel_process.kernel_process_step_state import KernelProcessStepState
|
||||
from semantic_kernel.processes.process_builder import ProcessBuilder
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from semantic_kernel.processes.kernel_process import KernelProcess
|
||||
from semantic_kernel.processes.local_runtime.local_kernel_process import LocalKernelProcessContext
|
||||
|
||||
|
||||
async def _start_local_kernel_process(
|
||||
*,
|
||||
process: "KernelProcess",
|
||||
kernel: Kernel,
|
||||
initial_event: KernelProcessEvent | str | Enum,
|
||||
**kwargs: object,
|
||||
) -> "LocalKernelProcessContext":
|
||||
from semantic_kernel.processes.local_runtime.local_kernel_process import start as start_local_kernel_process
|
||||
|
||||
return await start_local_kernel_process(
|
||||
process=process,
|
||||
kernel=kernel,
|
||||
initial_event=initial_event,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
|
||||
|
||||
class CommonEvents(Enum):
|
||||
"""Common events for both samples."""
|
||||
|
||||
USER_INPUT_RECEIVED = "UserInputReceived"
|
||||
COMPLETION_RESPONSE_GENERATED = "CompletionResponseGenerated"
|
||||
WELCOME_DONE = "WelcomeDone"
|
||||
A_STEP_DONE = "AStepDone"
|
||||
B_STEP_DONE = "BStepDone"
|
||||
C_STEP_DONE = "CStepDone"
|
||||
START_A_REQUESTED = "StartARequested"
|
||||
START_B_REQUESTED = "StartBRequested"
|
||||
EXIT_REQUESTED = "ExitRequested"
|
||||
START_PROCESS = "StartProcess"
|
||||
|
||||
######################################################################
|
||||
# region Semantic Kernel Process Framework path
|
||||
######################################################################
|
||||
|
||||
|
||||
class KickOffStep(KernelProcessStep[None]):
|
||||
KICK_OFF_FUNCTION: ClassVar[str] = "kick_off"
|
||||
|
||||
@kernel_function(name=KICK_OFF_FUNCTION)
|
||||
async def print_welcome_message(self, context: KernelProcessStepContext):
|
||||
await context.emit_event(process_event=CommonEvents.START_A_REQUESTED, data="Get Going A")
|
||||
await context.emit_event(process_event=CommonEvents.START_B_REQUESTED, data="Get Going B")
|
||||
|
||||
|
||||
class AStep(KernelProcessStep[None]):
|
||||
@kernel_function()
|
||||
async def do_it(self, context: KernelProcessStepContext):
|
||||
await asyncio.sleep(1)
|
||||
await context.emit_event(process_event=CommonEvents.A_STEP_DONE.value, data="I did A")
|
||||
|
||||
|
||||
class BStep(KernelProcessStep[None]):
|
||||
@kernel_function()
|
||||
async def do_it(self, context: KernelProcessStepContext):
|
||||
await asyncio.sleep(2)
|
||||
await context.emit_event(process_event=CommonEvents.B_STEP_DONE.value, data="I did B")
|
||||
|
||||
|
||||
class CStepState(BaseModel):
|
||||
current_cycle: int = 0
|
||||
|
||||
|
||||
class CStep(KernelProcessStep[CStepState]):
|
||||
state: CStepState = Field(default_factory=CStepState)
|
||||
|
||||
async def activate(self, state: KernelProcessStepState[CStepState]):
|
||||
self.state = state.state
|
||||
|
||||
@kernel_function()
|
||||
async def do_it(self, context: KernelProcessStepContext, astepdata: str, bstepdata: str):
|
||||
self.state.current_cycle += 1
|
||||
print(f"CStep Current Cycle: {self.state.current_cycle}")
|
||||
if self.state.current_cycle == 3:
|
||||
print("CStep Exit Requested")
|
||||
await context.emit_event(process_event=CommonEvents.EXIT_REQUESTED.value)
|
||||
return
|
||||
await context.emit_event(process_event=CommonEvents.C_STEP_DONE.value)
|
||||
|
||||
|
||||
kernel = Kernel()
|
||||
|
||||
|
||||
async def run_semantic_kernel_process_example() -> None:
|
||||
kernel.add_service(OpenAIChatCompletion(service_id="default"))
|
||||
|
||||
process = ProcessBuilder(name="Process Framework Sample")
|
||||
|
||||
kickoff_step = process.add_step(step_type=KickOffStep)
|
||||
step_a = process.add_step(step_type=AStep)
|
||||
step_b = process.add_step(step_type=BStep)
|
||||
step_c = process.add_step(step_type=CStep)
|
||||
|
||||
process.on_input_event(event_id=CommonEvents.START_PROCESS.value).send_event_to(target=kickoff_step)
|
||||
|
||||
kickoff_step.on_event(event_id=CommonEvents.START_A_REQUESTED.value).send_event_to(target=step_a)
|
||||
kickoff_step.on_event(event_id=CommonEvents.START_B_REQUESTED.value).send_event_to(target=step_b)
|
||||
step_a.on_event(event_id=CommonEvents.A_STEP_DONE.value).send_event_to(target=step_c, parameter_name="astepdata")
|
||||
step_b.on_event(event_id=CommonEvents.B_STEP_DONE.value).send_event_to(target=step_c, parameter_name="bstepdata")
|
||||
step_c.on_event(event_id=CommonEvents.C_STEP_DONE.value).send_event_to(target=kickoff_step)
|
||||
step_c.on_event(event_id=CommonEvents.EXIT_REQUESTED.value).stop_process()
|
||||
|
||||
kernel_process: "KernelProcess" = process.build()
|
||||
|
||||
async with await _start_local_kernel_process(
|
||||
process=kernel_process,
|
||||
kernel=kernel,
|
||||
initial_event=KernelProcessEvent(id=CommonEvents.START_PROCESS.value, data="Initial"),
|
||||
) as process_context:
|
||||
process_state = await process_context.get_state()
|
||||
c_step_state: KernelProcessStepState[CStepState] | None = next(
|
||||
(s.state for s in process_state.steps if s.state.name == "CStep"),
|
||||
None,
|
||||
)
|
||||
if c_step_state is None or c_step_state.state is None:
|
||||
raise RuntimeError("CStep state unavailable")
|
||||
assert c_step_state.state.current_cycle == 3 # nosec
|
||||
print(f"Final State Check: CStepState current cycle: {c_step_state.state.current_cycle}")
|
||||
|
||||
######################################################################
|
||||
# region Agent Framework workflow path
|
||||
######################################################################
|
||||
|
||||
|
||||
@dataclass
|
||||
class StepResult:
|
||||
origin: str
|
||||
cycle: int
|
||||
data: str
|
||||
|
||||
|
||||
class KickOffExecutor(Executor):
|
||||
def __init__(self, *, id: str = "kickoff") -> None:
|
||||
super().__init__(id=id)
|
||||
self._next_cycle = 0
|
||||
|
||||
@handler
|
||||
async def handle(self, event: CommonEvents, ctx: WorkflowContext[int]) -> None:
|
||||
if event not in {CommonEvents.START_PROCESS, CommonEvents.C_STEP_DONE}:
|
||||
return
|
||||
self._next_cycle += 1
|
||||
await ctx.send_message(self._next_cycle)
|
||||
|
||||
|
||||
class DelayedStepExecutor(Executor):
|
||||
def __init__(self, *, name: str, delay_seconds: float) -> None:
|
||||
super().__init__(id=name)
|
||||
self._delay = delay_seconds
|
||||
self._name = name
|
||||
|
||||
@handler
|
||||
async def handle(self, cycle: int, ctx: WorkflowContext[StepResult]) -> None:
|
||||
await asyncio.sleep(self._delay)
|
||||
await ctx.send_message(StepResult(origin=self._name, cycle=cycle, data=f"I did {self._name.upper()[-1]}"))
|
||||
|
||||
|
||||
class FanInExecutor(Executor):
|
||||
def __init__(self, *, required_cycles: int = 3, id: str = "fanin") -> None:
|
||||
super().__init__(id=id)
|
||||
self._completed_cycles = 0
|
||||
self._required_cycles = required_cycles
|
||||
|
||||
@handler
|
||||
async def handle(self, results: list[StepResult], ctx: WorkflowContext[CommonEvents, str]) -> None:
|
||||
if not results:
|
||||
return
|
||||
cycle_number = results[0].cycle
|
||||
summary = ", ".join(f"{r.origin}: {r.data}" for r in results)
|
||||
print(f"Cycle {cycle_number} aggregate -> {summary}")
|
||||
|
||||
self._completed_cycles += 1
|
||||
if self._completed_cycles >= self._required_cycles:
|
||||
await ctx.yield_output(f"Completed {self._completed_cycles} cycles")
|
||||
return
|
||||
|
||||
await ctx.send_message(CommonEvents.C_STEP_DONE)
|
||||
|
||||
|
||||
async def run_agent_framework_workflow_example() -> str | None:
|
||||
kickoff = KickOffExecutor()
|
||||
step_a = DelayedStepExecutor(name="step_a", delay_seconds=1)
|
||||
step_b = DelayedStepExecutor(name="step_b", delay_seconds=2)
|
||||
aggregate = FanInExecutor(required_cycles=3)
|
||||
|
||||
workflow = (
|
||||
WorkflowBuilder()
|
||||
.add_edge(kickoff, step_a)
|
||||
.add_edge(kickoff, step_b)
|
||||
.add_fan_in_edges([step_a, step_b], aggregate)
|
||||
.add_edge(aggregate, kickoff)
|
||||
.set_start_executor(kickoff)
|
||||
.build()
|
||||
)
|
||||
|
||||
final_text: str | None = None
|
||||
async for event in workflow.run_stream(CommonEvents.START_PROCESS):
|
||||
if isinstance(event, WorkflowOutputEvent):
|
||||
final_text = cast(str, event.data)
|
||||
|
||||
return final_text
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
print("===== Agent Framework Workflow =====")
|
||||
af_result = await run_agent_framework_workflow_example()
|
||||
if af_result:
|
||||
print(af_result)
|
||||
else:
|
||||
print("No Agent Framework output.")
|
||||
|
||||
print("===== Semantic Kernel Process Framework =====")
|
||||
await run_semantic_kernel_process_example()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,279 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Nested process comparison between Semantic Kernel Process Framework and Agent Framework sub-workflows."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import ClassVar, cast
|
||||
|
||||
######################################################################
|
||||
# region Agent Framework imports
|
||||
######################################################################
|
||||
from agent_framework import (
|
||||
Executor,
|
||||
WorkflowBuilder,
|
||||
WorkflowContext,
|
||||
WorkflowExecutor,
|
||||
WorkflowOutputEvent,
|
||||
handler,
|
||||
)
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
######################################################################
|
||||
# region Semantic Kernel imports
|
||||
######################################################################
|
||||
from semantic_kernel import Kernel
|
||||
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
|
||||
from semantic_kernel.functions import kernel_function
|
||||
from semantic_kernel.processes.kernel_process.kernel_process import KernelProcess
|
||||
from semantic_kernel.processes.kernel_process.kernel_process_event import KernelProcessEventVisibility
|
||||
from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep
|
||||
from semantic_kernel.processes.kernel_process.kernel_process_step_context import KernelProcessStepContext
|
||||
from semantic_kernel.processes.kernel_process.kernel_process_step_state import KernelProcessStepState
|
||||
from semantic_kernel.processes.local_runtime.local_kernel_process import start
|
||||
from semantic_kernel.processes.process_builder import ProcessBuilder
|
||||
from typing_extensions import Never
|
||||
|
||||
######################################################################
|
||||
# endregion
|
||||
######################################################################
|
||||
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
|
||||
|
||||
class ProcessEvents(Enum):
|
||||
START_PROCESS = "StartProcess"
|
||||
START_INNER_PROCESS = "StartInnerProcess"
|
||||
OUTPUT_READY_PUBLIC = "OutputReadyPublic"
|
||||
OUTPUT_READY_INTERNAL = "OutputReadyInternal"
|
||||
|
||||
######################################################################
|
||||
# region Semantic Kernel nested process path
|
||||
######################################################################
|
||||
|
||||
|
||||
class StepState(BaseModel):
|
||||
last_message: str | None = None
|
||||
|
||||
|
||||
class EchoStep(KernelProcessStep[None]):
|
||||
ECHO: ClassVar[str] = "echo"
|
||||
|
||||
@kernel_function(name=ECHO)
|
||||
async def echo(self, message: str) -> str:
|
||||
print(f"[ECHO] {message}")
|
||||
return message
|
||||
|
||||
|
||||
class RepeatStep(KernelProcessStep[StepState]):
|
||||
REPEAT: ClassVar[str] = "repeat"
|
||||
|
||||
state: StepState = Field(default_factory=StepState)
|
||||
|
||||
async def activate(self, state: KernelProcessStepState[StepState]):
|
||||
self.state = state.state
|
||||
|
||||
@kernel_function(name=REPEAT)
|
||||
async def repeat(
|
||||
self,
|
||||
message: str,
|
||||
context: KernelProcessStepContext,
|
||||
count: int = 2,
|
||||
) -> None:
|
||||
output = " ".join([message] * count)
|
||||
self.state.last_message = output
|
||||
print(f"[REPEAT] {output}")
|
||||
|
||||
await context.emit_event(
|
||||
process_event=ProcessEvents.OUTPUT_READY_PUBLIC.value,
|
||||
data=output,
|
||||
visibility=KernelProcessEventVisibility.Public,
|
||||
)
|
||||
await context.emit_event(
|
||||
process_event=ProcessEvents.OUTPUT_READY_INTERNAL.value,
|
||||
data=output,
|
||||
visibility=KernelProcessEventVisibility.Internal,
|
||||
)
|
||||
|
||||
|
||||
def _create_linear_process(name: str) -> ProcessBuilder:
|
||||
process_builder = ProcessBuilder(name=name)
|
||||
echo_step = process_builder.add_step(step_type=EchoStep)
|
||||
repeat_step = process_builder.add_step(step_type=RepeatStep)
|
||||
|
||||
process_builder.on_input_event(event_id=ProcessEvents.START_PROCESS.value).send_event_to(target=echo_step)
|
||||
|
||||
echo_step.on_function_result(function_name=EchoStep.ECHO).send_event_to(
|
||||
target=repeat_step,
|
||||
parameter_name="message",
|
||||
)
|
||||
|
||||
return process_builder
|
||||
|
||||
|
||||
_semantic_kernel = Kernel()
|
||||
|
||||
|
||||
async def run_semantic_kernel_nested_process() -> None:
|
||||
_semantic_kernel.add_service(OpenAIChatCompletion(service_id="default"))
|
||||
|
||||
process_builder = _create_linear_process("Outer")
|
||||
nested_process_step = process_builder.add_step_from_process(_create_linear_process("Inner"))
|
||||
|
||||
process_builder.steps[1].on_event(ProcessEvents.OUTPUT_READY_INTERNAL.value).send_event_to(
|
||||
nested_process_step.where_input_event_is(ProcessEvents.START_PROCESS.value)
|
||||
)
|
||||
|
||||
kernel_process = process_builder.build()
|
||||
|
||||
process_handle = await start(
|
||||
process=kernel_process,
|
||||
kernel=_semantic_kernel,
|
||||
initial_event=ProcessEvents.START_PROCESS.value,
|
||||
data="Test",
|
||||
)
|
||||
process_info = await process_handle.get_state()
|
||||
|
||||
inner_process: KernelProcess | None = next(
|
||||
(s for s in process_info.steps if s.state.name == "Inner"),
|
||||
None,
|
||||
)
|
||||
if inner_process is None:
|
||||
raise RuntimeError("Inner process state missing")
|
||||
|
||||
repeat_state: KernelProcessStepState[StepState] | None = next(
|
||||
(s.state for s in inner_process.steps if s.state.name == "RepeatStep"),
|
||||
None,
|
||||
)
|
||||
if repeat_state is None or repeat_state.state is None:
|
||||
raise RuntimeError("RepeatStep state missing")
|
||||
assert repeat_state.state.last_message == "Test Test Test Test" # nosec
|
||||
|
||||
######################################################################
|
||||
# region Agent Framework nested workflow path
|
||||
######################################################################
|
||||
|
||||
|
||||
@dataclass
|
||||
class RepeatPayload:
|
||||
message: str
|
||||
count: int = 2
|
||||
|
||||
|
||||
class KickoffExecutor(Executor):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(id="kickoff")
|
||||
|
||||
@handler
|
||||
async def start(self, message: str, ctx: WorkflowContext[RepeatPayload]) -> None:
|
||||
print(f"[OUTER] Start with message: {message}")
|
||||
await ctx.send_message(RepeatPayload(message=message, count=2))
|
||||
|
||||
|
||||
class OuterEchoExecutor(Executor):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(id="outer_echo")
|
||||
|
||||
@handler
|
||||
async def echo(self, payload: RepeatPayload, ctx: WorkflowContext[RepeatPayload]) -> None:
|
||||
print(f"[OUTER ECHO] {payload.message}")
|
||||
await ctx.send_message(payload)
|
||||
|
||||
|
||||
class OuterRepeatExecutor(Executor):
|
||||
def __init__(self, *, inner_target_id: str) -> None:
|
||||
super().__init__(id="outer_repeat")
|
||||
self._inner_target_id = inner_target_id
|
||||
|
||||
@handler
|
||||
async def repeat(self, payload: RepeatPayload, ctx: WorkflowContext[RepeatPayload]) -> None:
|
||||
repeated = " ".join([payload.message] * payload.count)
|
||||
print(f"[OUTER REPEAT] {repeated}")
|
||||
await ctx.send_message(RepeatPayload(message=repeated, count=2), target_id=self._inner_target_id)
|
||||
|
||||
|
||||
class InnerEchoExecutor(Executor):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(id="inner_echo")
|
||||
|
||||
@handler
|
||||
async def echo(self, payload: RepeatPayload, ctx: WorkflowContext[RepeatPayload]) -> None:
|
||||
print(f" [INNER ECHO] {payload.message}")
|
||||
await ctx.send_message(payload)
|
||||
|
||||
|
||||
class InnerRepeatExecutor(Executor):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(id="inner_repeat")
|
||||
|
||||
@handler
|
||||
async def repeat(self, payload: RepeatPayload, ctx: WorkflowContext[Never, str]) -> None:
|
||||
repeated = " ".join([payload.message] * payload.count)
|
||||
print(f" [INNER REPEAT] {repeated}")
|
||||
await ctx.yield_output(repeated)
|
||||
|
||||
|
||||
class CollectResultExecutor(Executor):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(id="collector")
|
||||
|
||||
@handler
|
||||
async def collect(self, result: str, ctx: WorkflowContext[Never, str]) -> None:
|
||||
print(f"[COLLECTOR] Final result -> {result}")
|
||||
await ctx.yield_output(result)
|
||||
|
||||
|
||||
def _build_inner_workflow() -> WorkflowExecutor:
|
||||
inner_echo = InnerEchoExecutor()
|
||||
inner_repeat = InnerRepeatExecutor()
|
||||
|
||||
inner_workflow = WorkflowBuilder().set_start_executor(inner_echo).add_edge(inner_echo, inner_repeat).build()
|
||||
|
||||
return WorkflowExecutor(inner_workflow, id="inner_workflow")
|
||||
|
||||
|
||||
async def run_agent_framework_nested_workflow(initial_message: str) -> Sequence[str]:
|
||||
inner_executor = _build_inner_workflow()
|
||||
|
||||
kickoff = KickoffExecutor()
|
||||
outer_echo = OuterEchoExecutor()
|
||||
outer_repeat = OuterRepeatExecutor(inner_target_id=inner_executor.id)
|
||||
collector = CollectResultExecutor()
|
||||
|
||||
outer_workflow = (
|
||||
WorkflowBuilder()
|
||||
.set_start_executor(kickoff)
|
||||
.add_edge(kickoff, outer_echo)
|
||||
.add_edge(outer_echo, outer_repeat)
|
||||
.add_edge(outer_repeat, inner_executor)
|
||||
.add_edge(inner_executor, collector)
|
||||
.build()
|
||||
)
|
||||
|
||||
results: list[str] = []
|
||||
async for event in outer_workflow.run_stream(initial_message):
|
||||
if isinstance(event, WorkflowOutputEvent):
|
||||
results.append(cast(str, event.data))
|
||||
|
||||
return results
|
||||
|
||||
######################################################################
|
||||
# endregion
|
||||
######################################################################
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
print("===== Agent Framework Nested Workflow =====")
|
||||
af_results = await run_agent_framework_nested_workflow("Test")
|
||||
for index, value in enumerate(af_results, start=1):
|
||||
print(f"Result {index}: {value}")
|
||||
|
||||
print("\n===== Semantic Kernel Nested Process =====")
|
||||
await run_semantic_kernel_nested_process()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user