wehub-resource-sync
266 lines
11 KiB
Python
266 lines
11 KiB
Python
# Copyright (c) Microsoft. All rights reserved.
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
import uuid
|
|
from collections.abc import Callable
|
|
from queue import Queue
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from pydantic import Field
|
|
|
|
from semantic_kernel.exceptions import KernelException
|
|
from semantic_kernel.exceptions.process_exceptions import ProcessEventUndefinedException
|
|
from semantic_kernel.kernel import Kernel
|
|
from semantic_kernel.processes.const import END_PROCESS_ID
|
|
from semantic_kernel.processes.kernel_process.kernel_process_state import KernelProcessState
|
|
from semantic_kernel.processes.kernel_process.kernel_process_step_info import KernelProcessStepInfo
|
|
from semantic_kernel.processes.local_runtime.local_event import (
|
|
KernelProcessEvent,
|
|
KernelProcessEventVisibility,
|
|
LocalEvent,
|
|
)
|
|
from semantic_kernel.processes.local_runtime.local_message import LocalMessage
|
|
from semantic_kernel.processes.local_runtime.local_message_factory import LocalMessageFactory
|
|
from semantic_kernel.processes.local_runtime.local_step import LocalStep
|
|
from semantic_kernel.utils.feature_stage_decorator import experimental
|
|
|
|
if TYPE_CHECKING:
|
|
from semantic_kernel.processes.kernel_process.kernel_process import KernelProcess
|
|
|
|
logger: logging.Logger = logging.getLogger(__name__)
|
|
|
|
|
|
@experimental
|
|
class LocalProcess(LocalStep):
|
|
"""A local process that contains a collection of steps."""
|
|
|
|
kernel: Kernel
|
|
steps: list[LocalStep] = Field(default_factory=list)
|
|
step_infos: list[KernelProcessStepInfo] = Field(default_factory=list)
|
|
process: "KernelProcess"
|
|
initialize_task: bool | None = False
|
|
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,
|
|
process: "KernelProcess",
|
|
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] = {
|
|
"step_info": process,
|
|
"kernel": kernel,
|
|
"parent_process_id": parent_process_id,
|
|
"step_infos": list(process.steps),
|
|
"process": process,
|
|
"initialize_task": False,
|
|
}
|
|
|
|
if max_supersteps is not None:
|
|
args["max_supersteps"] = max_supersteps
|
|
|
|
if factories:
|
|
args["factories"] = factories
|
|
|
|
super().__init__(**args)
|
|
|
|
def ensure_initialized(self):
|
|
"""Ensures the process is initialized lazily (synchronously)."""
|
|
if not self.initialize_task:
|
|
self.initialize_process()
|
|
self.initialize_task = True
|
|
|
|
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(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.
|
|
|
|
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)
|
|
await self.start(keep_alive=False)
|
|
if self.process_task:
|
|
await self.process_task
|
|
|
|
async def stop(self):
|
|
"""Stops a running process."""
|
|
if not self.process_task or self.process_task.done():
|
|
return # Task is already finished or hasn't started
|
|
|
|
self.process_task.cancel()
|
|
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self.process_task
|
|
|
|
async def initialize_step(self):
|
|
"""Initializes the step."""
|
|
# The process does not need any further initialization
|
|
pass
|
|
|
|
async def send_message(self, process_event: KernelProcessEvent):
|
|
"""Sends a message to the process."""
|
|
if process_event is None:
|
|
raise ProcessEventUndefinedException("The process event must be specified.")
|
|
self.external_event_queue.put(process_event)
|
|
|
|
async def get_process_info(self) -> "KernelProcess":
|
|
"""Gets the process information."""
|
|
return await self.to_kernel_process()
|
|
|
|
def initialize_process(self):
|
|
"""Initializes the input and output edges for the process and initializes the steps."""
|
|
from semantic_kernel.processes.kernel_process.kernel_process import KernelProcess
|
|
|
|
# Initialize the input and output edges for the process
|
|
self.output_edges = {kvp[0]: list(kvp[1]) for kvp in self.process.edges.items()}
|
|
|
|
# Initialize the steps within this process
|
|
for step in self.step_infos:
|
|
local_step = None
|
|
|
|
# The current step should already have a name.
|
|
assert step.state and step.state.name is not None # nosec
|
|
|
|
if isinstance(step, KernelProcess):
|
|
# The process will only have an Id if it's already been executed.
|
|
if not step.state.id:
|
|
step.state.id = str(uuid.uuid4().hex)
|
|
|
|
# Create a new LocalProcess for the step
|
|
process = LocalProcess(
|
|
process=step,
|
|
kernel=self.kernel,
|
|
factories=self.factories,
|
|
parent_process_id=self.id,
|
|
)
|
|
|
|
local_step = process
|
|
else:
|
|
# The current step should already have an Id.
|
|
assert step.state and step.state.id is not None # nosec
|
|
|
|
# Create a LocalStep for the step
|
|
local_step = LocalStep( # type: ignore
|
|
step_info=step,
|
|
kernel=self.kernel,
|
|
factories=self.factories,
|
|
parent_process_id=self.id,
|
|
)
|
|
|
|
# Add the local step to the list of steps
|
|
self.steps.append(local_step)
|
|
|
|
async def handle_message(self, message: LocalMessage):
|
|
"""Handles a LocalMessage that has been sent to the process."""
|
|
if not message.target_event_id:
|
|
error_message = (
|
|
"Internal Process Error: The target event id must be specified when sending a message to a step."
|
|
)
|
|
logger.error(error_message)
|
|
raise KernelException(error_message)
|
|
|
|
event_id = message.target_event_id
|
|
if event_id in self.output_edges:
|
|
nested_event = KernelProcessEvent(
|
|
id=event_id, data=message.target_event_data, visibility=KernelProcessEventVisibility.Internal
|
|
)
|
|
await self.run_once(nested_event)
|
|
|
|
async def internal_execute(self, max_supersteps: int = 100, keep_alive: bool = True):
|
|
"""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)
|
|
for step in self.steps:
|
|
await self.enqueue_step_messages(step, message_channel)
|
|
|
|
messages_to_process: list[LocalMessage] = []
|
|
while not message_channel.empty():
|
|
messages_to_process.append(message_channel.get())
|
|
|
|
if not messages_to_process and (not keep_alive or self.external_event_queue.empty()):
|
|
break
|
|
|
|
message_tasks = []
|
|
for message in messages_to_process:
|
|
if message.destination_id == END_PROCESS_ID:
|
|
break
|
|
|
|
destination_step = next(step for step in self.steps if step.id == message.destination_id)
|
|
message_tasks.append(destination_step.handle_message(message))
|
|
|
|
await asyncio.gather(*message_tasks)
|
|
|
|
except Exception as ex:
|
|
logger.error(f"An error occurred while running the process: {ex}.")
|
|
raise
|
|
|
|
async def to_kernel_process(self) -> "KernelProcess":
|
|
"""Builds a KernelProcess from the current LocalProcess."""
|
|
from semantic_kernel.processes.kernel_process.kernel_process import KernelProcess
|
|
|
|
process_state = KernelProcessState(name=self.name, version=self.step_state.version, id=self.id)
|
|
step_tasks = [step.to_kernel_process_step_info() for step in self.steps]
|
|
steps = await asyncio.gather(*step_tasks)
|
|
return KernelProcess(state=process_state, steps=steps, edges=self.output_edges)
|
|
|
|
async def to_kernel_process_step_info(self) -> "KernelProcessStepInfo":
|
|
"""Extracts the current state of the step and returns it as a KernelProcessStepInfo."""
|
|
return await self.to_kernel_process()
|
|
|
|
def enqueue_external_messages(self, message_channel: Queue[LocalMessage]):
|
|
"""Processes external events that have been sent to the process."""
|
|
while not self.external_event_queue.empty():
|
|
external_event: KernelProcessEvent = self.external_event_queue.get_nowait()
|
|
if external_event.id in self.output_edges:
|
|
edges = self.output_edges[external_event.id]
|
|
for edge in edges:
|
|
message = LocalMessageFactory.create_from_edge(edge, external_event.data)
|
|
message_channel.put(message)
|
|
|
|
async def enqueue_step_messages(self, step: LocalStep, message_channel: Queue[LocalMessage]):
|
|
"""Processes events emitted by the given step and enqueues them."""
|
|
all_step_events = step.get_all_events()
|
|
for step_event in all_step_events:
|
|
# must come first because emitting the step event modifies its namespace
|
|
for edge in step.get_edge_for_event(step_event.id):
|
|
message = LocalMessageFactory.create_from_edge(edge, step_event.data)
|
|
message_channel.put(message)
|
|
|
|
if step_event.visibility == KernelProcessEventVisibility.Public:
|
|
if isinstance(step_event, KernelProcessEvent):
|
|
await self.emit_event(step_event) # type: ignore
|
|
elif isinstance(step_event, LocalEvent):
|
|
await self.emit_local_event(step_event) # type: ignore
|
|
|
|
def dispose(self):
|
|
"""Clean up resources."""
|
|
if self.process_task:
|
|
self.process_task.cancel()
|