You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2025/06/25 11:57:36 UTC
(flink-agents) branch main updated: [api][python] Introduce Python API for Workflow, Event and Action.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 774e663 [api][python] Introduce Python API for Workflow, Event and Action.
774e663 is described below
commit 774e66325ed30ac850efe467568ecf0508d1b4f7
Author: WenjinXie <we...@gmail.com>
AuthorDate: Thu Jun 19 10:05:27 2025 +0800
[api][python] Introduce Python API for Workflow, Event and Action.
---
.../{plan/workflow_plan.py => api/decorators.py} | 47 +++---
python/flink_agents/api/execution_enviroment.py | 80 +++++++++
.../workflow_plan.py => api/runner_context.py} | 32 ++--
python/flink_agents/api/tests/test_decorators.py | 60 +++++++
.../{plan/tests/test_action.py => api/workflow.py} | 33 +---
python/flink_agents/examples/workflow_example.py | 69 ++++++++
python/flink_agents/plan/action.py | 3 +-
python/flink_agents/plan/function.py | 1 +
python/flink_agents/plan/tests/test_action.py | 5 +-
.../flink_agents/plan/tests/test_workflow_plan.py | 58 +++++++
python/flink_agents/plan/workflow_plan.py | 38 +++++
.../runtime/local_execution_environment.py | 86 ++++++++++
python/flink_agents/runtime/local_runner.py | 180 +++++++++++++++++++++
.../test_action.py => runtime/tests/__init__.py} | 29 ----
.../tests/test_local_execution_environment.py | 82 ++++++++++
.../workflow_runner.py} | 37 ++---
16 files changed, 712 insertions(+), 128 deletions(-)
diff --git a/python/flink_agents/plan/workflow_plan.py b/python/flink_agents/api/decorators.py
similarity index 55%
copy from python/flink_agents/plan/workflow_plan.py
copy to python/flink_agents/api/decorators.py
index e54d0e5..58e902e 100644
--- a/python/flink_agents/plan/workflow_plan.py
+++ b/python/flink_agents/api/decorators.py
@@ -15,37 +15,36 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
-
-from typing import Dict, List, Type
-
-from pydantic import BaseModel
+from typing import Callable, Tuple, Type
from flink_agents.api.event import Event
-from flink_agents.plan.action import Action
-class WorkflowPlan(BaseModel):
- """Workflow plan compiled from user defined workflow.
+def action(*listen_events: Tuple[Type[Event], ...]) -> Callable:
+ """Decorator for marking a function as a workflow action.
- Attributes:
+ Parameters
----------
- actions : Dict[Type[Event], List[Action]]
- Mapping of event types to the list of Actions that listen to them.
+ listen_events : list[Type[Event]]
+ List of event types that this action should respond to.
+
+ Returns:
+ -------
+ Callable
+ Decorator function that marks the target function with event listeners.
+
+ Raises:
+ ------
+ AssertionError
+ If no events are provided to listen to.
"""
+ assert len(listen_events) > 0, 'action must have at least one event type to listen to'
- actions: Dict[Type[Event], List[Action]]
-
- def get_actions(self, event_type: Type[Event]) -> List[Action]:
- """Get actions that listen to the specified event type.
+ for event in listen_events:
+ assert issubclass(event, Event), 'action must only listen to event types.'
- Parameters
- ----------
- event_type : Type[Event]
- The event type to query.
+ def decorator(func: Callable) -> Callable:
+ func._listen_events = listen_events
+ return func
- Returns:
- -------
- list[Action]
- List of Actions that will respond to this event type.
- """
- return self.actions[event_type]
+ return decorator
diff --git a/python/flink_agents/api/execution_enviroment.py b/python/flink_agents/api/execution_enviroment.py
new file mode 100644
index 0000000..e77f610
--- /dev/null
+++ b/python/flink_agents/api/execution_enviroment.py
@@ -0,0 +1,80 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+import importlib
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List
+
+from flink_agents.api.workflow import Workflow
+
+
+class AgentsExecutionEnvironment(ABC):
+ """Base class for workflow execution environment."""
+
+ @staticmethod
+ def get_execution_environment(**kwargs: Dict[str, Any]) -> 'AgentsExecutionEnvironment':
+ """Get agents execution environment.
+
+ Currently, this method only returns LocalExecutionEnvironment. After
+ implement other AgentsExecutionEnvironments, this method will return
+ appropriate environment according to configuration.
+
+ Returns:
+ -------
+ AgentsExecutionEnvironment
+ Environment for workflow execution.
+ """
+ return importlib.import_module(
+ "flink_agents.runtime.local_execution_environment"
+ ).get_execution_environment(**kwargs)
+
+ @abstractmethod
+ def from_list(self, input: List[Dict[str, Any]]) -> 'AgentsExecutionEnvironment':
+ """Set input for agents. Used for local execution.
+
+ Parameters
+ ----------
+ input : list
+ Receive a list as input. The element in the list should be a dict like
+ {'key': Any, 'value': Any} or {'value': Any} , extra field will be ignored.
+ """
+
+ @abstractmethod
+ def apply(self, workflow: Workflow) -> 'AgentsExecutionEnvironment':
+ """Set workflow of execution environment.
+
+ Parameters
+ ----------
+ workflow : Workflow
+ The workflow user defined to run in execution environment.
+ """
+
+ @abstractmethod
+ def to_list(self) -> List[Dict[str, Any]]:
+ """Get outputs of workflow execution. Used for local execution.
+
+ The element in the list is a dict like {'key': output}.
+
+ Returns:
+ -------
+ list
+ Outputs of workflow execution.
+ """
+
+ @abstractmethod
+ def execute(self) -> None:
+ """Execute agents."""
diff --git a/python/flink_agents/plan/workflow_plan.py b/python/flink_agents/api/runner_context.py
similarity index 58%
copy from python/flink_agents/plan/workflow_plan.py
copy to python/flink_agents/api/runner_context.py
index e54d0e5..4823ffb 100644
--- a/python/flink_agents/plan/workflow_plan.py
+++ b/python/flink_agents/api/runner_context.py
@@ -15,37 +15,23 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
-
-from typing import Dict, List, Type
-
-from pydantic import BaseModel
+from abc import ABC, abstractmethod
from flink_agents.api.event import Event
-from flink_agents.plan.action import Action
-class WorkflowPlan(BaseModel):
- """Workflow plan compiled from user defined workflow.
+class RunnerContext(ABC):
+ """Abstract base class providing context for workflow execution.
- Attributes:
- ----------
- actions : Dict[Type[Event], List[Action]]
- Mapping of event types to the list of Actions that listen to them.
+ This context provides access to event handling.
"""
- actions: Dict[Type[Event], List[Action]]
-
- def get_actions(self, event_type: Type[Event]) -> List[Action]:
- """Get actions that listen to the specified event type.
+ @abstractmethod
+ def send_event(self, event: Event) -> None:
+ """Send an event to the workflow for processing.
Parameters
----------
- event_type : Type[Event]
- The event type to query.
-
- Returns:
- -------
- list[Action]
- List of Actions that will respond to this event type.
+ event : Event
+ The event to be processed by the workflow system.
"""
- return self.actions[event_type]
diff --git a/python/flink_agents/api/tests/test_decorators.py b/python/flink_agents/api/tests/test_decorators.py
new file mode 100644
index 0000000..7d20127
--- /dev/null
+++ b/python/flink_agents/api/tests/test_decorators.py
@@ -0,0 +1,60 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+from typing import List
+
+import pytest
+
+from flink_agents.api.decorators import action
+from flink_agents.api.event import Event, InputEvent, OutputEvent
+from flink_agents.api.runner_context import RunnerContext
+
+
+def test_action_decorator() -> None: #noqa D103
+ @action(InputEvent)
+ def forward_action(event: Event, ctx: RunnerContext) -> None:
+ input = event.input
+ ctx.send_event(OutputEvent(output=input))
+
+ assert hasattr(forward_action, '_listen_events')
+ listen_events = forward_action._listen_events
+ assert listen_events == (InputEvent,)
+
+def test_action_decorator_listen_multi_events() -> None: #noqa D103
+ @action(InputEvent, OutputEvent)
+ def forward_action(event: Event, ctx: RunnerContext) -> None:
+ input = event.input
+ ctx.send_event(OutputEvent(output=input))
+
+ assert hasattr(forward_action, '_listen_events')
+ listen_events = forward_action._listen_events
+ assert listen_events == (InputEvent, OutputEvent)
+
+def test_action_decorator_listen_no_event() -> None: #noqa D103
+ with pytest.raises(AssertionError):
+ @action()
+ def forward_action(event: Event, ctx: RunnerContext) -> None:
+ input = event.input
+ ctx.send_event(OutputEvent(output=input))
+
+def test_action_decorator_listen_non_event_type() -> None: #noqa D103
+ with pytest.raises(AssertionError):
+ @action(List)
+ def forward_action(event: Event, ctx: RunnerContext) -> None:
+ input = event.input
+ ctx.send_event(OutputEvent(output=input))
+
diff --git a/python/flink_agents/plan/tests/test_action.py b/python/flink_agents/api/workflow.py
similarity index 55%
copy from python/flink_agents/plan/tests/test_action.py
copy to python/flink_agents/api/workflow.py
index e47a4ae..d0300ea 100644
--- a/python/flink_agents/plan/tests/test_action.py
+++ b/python/flink_agents/api/workflow.py
@@ -15,32 +15,11 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
-import pytest
-
-from flink_agents.api.event import InputEvent
-from flink_agents.plan.action import Action
-from flink_agents.plan.function import PythonFunction
-
-
-def legal_signature(event: InputEvent) -> None: # noqa: D103
- pass
-
-def illegal_signature(value: int) -> None: # noqa: D103
- pass
-
-def test_action_signature_legal() -> None: # noqa: D103
- Action(
- name="legal",
- exec=PythonFunction.from_callable(legal_signature),
- listen_event_types=[InputEvent],
- )
-
-def test_action_signature_illegal() -> None: # noqa: D103
- with pytest.raises(TypeError):
- Action(
- name="illegal",
- exec=PythonFunction.from_callable(illegal_signature),
- listen_event_types=[InputEvent],
- )
+from abc import ABC
+class Workflow(ABC): # noqa: B024
+ """Base class for defining workflow logic. Currently, Workflow is
+ an empty interface, we may add interactive method like add_action(),
+ add_model() later.
+ """
diff --git a/python/flink_agents/examples/workflow_example.py b/python/flink_agents/examples/workflow_example.py
new file mode 100644
index 0000000..89cdb15
--- /dev/null
+++ b/python/flink_agents/examples/workflow_example.py
@@ -0,0 +1,69 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+from typing import Any
+
+from flink_agents.api.decorators import action
+from flink_agents.api.event import Event, InputEvent, OutputEvent
+from flink_agents.api.execution_enviroment import AgentsExecutionEnvironment
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.workflow import Workflow
+
+
+class MyEvent(Event): #noqa D101
+ value: Any
+
+#TODO: Replace this workflow with more practical example.
+class MyWorkflow(Workflow):
+ """An example of Workflow to show the basic usage.
+
+ Currently, this workflow doesn't really make sense, and it's mainly for developing
+ validation.
+ """
+ @action(InputEvent)
+ @staticmethod
+ def first_action(event: Event, ctx: RunnerContext): #noqa D102
+ input = event.input
+ content = input + ' first_action'
+ ctx.send_event(MyEvent(value=content))
+ ctx.send_event(OutputEvent(output=content))
+
+ @action(MyEvent)
+ @staticmethod
+ def second_action(event: Event, ctx: RunnerContext): #noqa D102
+ input = event.value
+ content = input + ' second_action'
+ ctx.send_event(OutputEvent(output=content))
+
+
+if __name__ == "__main__":
+ env = AgentsExecutionEnvironment.get_execution_environment()
+
+ input_list = []
+ workflow = MyWorkflow()
+
+ output_list = env.from_list(input_list).apply(workflow).to_list()
+
+
+ input_list.append({'key': 'bob', 'value': 'The message from bob'})
+ input_list.append({'k': 'john', 'v': 'The message from john'})
+ input_list.append({'value': 'The message from unknown'}) # will automatically generate a new unique key
+
+ env.execute()
+
+ for output in output_list:
+ print(output)
diff --git a/python/flink_agents/plan/action.py b/python/flink_agents/plan/action.py
index 9b471d4..c52da8e 100644
--- a/python/flink_agents/plan/action.py
+++ b/python/flink_agents/plan/action.py
@@ -20,6 +20,7 @@ from typing import List, Type
from pydantic import BaseModel
from flink_agents.api.event import Event
+from flink_agents.api.runner_context import RunnerContext
from flink_agents.plan.function import Function
@@ -53,5 +54,5 @@ class Action(BaseModel):
"""Action will check function signature when init."""
super().__init__(name=name, exec=exec, listen_event_types=listen_event_types)
#TODO: Update expected signature after import State and Context.
- exec.check_signature(Event)
+ exec.check_signature(Event, RunnerContext)
diff --git a/python/flink_agents/plan/function.py b/python/flink_agents/plan/function.py
index ea2f35a..cf1f3e2 100644
--- a/python/flink_agents/plan/function.py
+++ b/python/flink_agents/plan/function.py
@@ -120,6 +120,7 @@ class PythonFunction(Function):
def __get_func(self) -> Callable:
if self.__func is None:
module = importlib.import_module(self.module)
+ #TODO: support function of inner class.
if "." in self.qualname:
# Handle class methods (e.g., 'ClassName.method')
classname, methodname = self.qualname.rsplit(".", 1)
diff --git a/python/flink_agents/plan/tests/test_action.py b/python/flink_agents/plan/tests/test_action.py
index e47a4ae..2bb400f 100644
--- a/python/flink_agents/plan/tests/test_action.py
+++ b/python/flink_agents/plan/tests/test_action.py
@@ -18,14 +18,15 @@
import pytest
from flink_agents.api.event import InputEvent
+from flink_agents.api.runner_context import RunnerContext
from flink_agents.plan.action import Action
from flink_agents.plan.function import PythonFunction
-def legal_signature(event: InputEvent) -> None: # noqa: D103
+def legal_signature(event: InputEvent, ctx: RunnerContext) -> None: # noqa: D103
pass
-def illegal_signature(value: int) -> None: # noqa: D103
+def illegal_signature(value: int, ctx: RunnerContext) -> None: # noqa: D103
pass
def test_action_signature_legal() -> None: # noqa: D103
diff --git a/python/flink_agents/plan/tests/test_workflow_plan.py b/python/flink_agents/plan/tests/test_workflow_plan.py
new file mode 100644
index 0000000..62d316f
--- /dev/null
+++ b/python/flink_agents/plan/tests/test_workflow_plan.py
@@ -0,0 +1,58 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+import pytest
+
+from flink_agents.api.decorators import action
+from flink_agents.api.event import Event, InputEvent, OutputEvent
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.workflow import Workflow
+from flink_agents.plan.function import PythonFunction
+from flink_agents.plan.workflow_plan import WorkflowPlan
+
+
+class TestWorkflow(Workflow): #noqa D101
+ @action(InputEvent)
+ @staticmethod
+ def increment(event: Event, ctx: RunnerContext) -> None: #noqa D102
+ value = event.input
+ value += 1
+ ctx.send_event(OutputEvent(output=value))
+
+def test_from_workflow(): #noqa D102
+ workflow = TestWorkflow()
+ workflow_plan = WorkflowPlan.from_workflow(workflow)
+ actions = workflow_plan.get_actions(InputEvent)
+ assert len(actions) == 1
+ action = actions[0]
+ assert action.name == "increment"
+ func = action.exec
+ assert isinstance(func, PythonFunction)
+ assert func.module == "flink_agents.plan.tests.test_workflow_plan"
+ assert func.qualname == "TestWorkflow.increment"
+ assert action.listen_event_types == [InputEvent]
+
+class InvalidWorkflow(Workflow): #noqa D101
+ @action(InputEvent)
+ @staticmethod
+ def invalid_signature_action(event: Event) -> None: #noqa D102
+ pass
+
+def test_to_workflow_invalid_signature() -> None: #noqa D103
+ workflow = InvalidWorkflow()
+ with pytest.raises(TypeError):
+ WorkflowPlan.from_workflow(workflow)
diff --git a/python/flink_agents/plan/workflow_plan.py b/python/flink_agents/plan/workflow_plan.py
index e54d0e5..2f0a1bb 100644
--- a/python/flink_agents/plan/workflow_plan.py
+++ b/python/flink_agents/plan/workflow_plan.py
@@ -21,7 +21,9 @@ from typing import Dict, List, Type
from pydantic import BaseModel
from flink_agents.api.event import Event
+from flink_agents.api.workflow import Workflow
from flink_agents.plan.action import Action
+from flink_agents.plan.function import PythonFunction
class WorkflowPlan(BaseModel):
@@ -35,6 +37,19 @@ class WorkflowPlan(BaseModel):
actions: Dict[Type[Event], List[Action]]
+ @staticmethod
+ def from_workflow(workflow: Workflow) -> "WorkflowPlan":
+ """Build a WorkflowPlan from user defined workflow."""
+ actions = {}
+ for action in _get_actions(workflow):
+ assert action.name not in actions, f"Duplicate action name: {action.name}"
+ for event_type in action.listen_event_types:
+ if event_type not in actions:
+ actions[event_type] = []
+ actions[event_type].append(action)
+ return WorkflowPlan(actions=actions)
+
+
def get_actions(self, event_type: Type[Event]) -> List[Action]:
"""Get actions that listen to the specified event type.
@@ -49,3 +64,26 @@ class WorkflowPlan(BaseModel):
List of Actions that will respond to this event type.
"""
return self.actions[event_type]
+
+def _get_actions(workflow: Workflow) -> List[Action]:
+ """Extract all registered workflow actions from a workflow.
+
+ Parameters
+ ----------
+ workflow : Workflow
+ The workflow to be analyzed.
+
+ Returns:
+ -------
+ List[Action]
+ List of Action defined in the workflow.
+ """
+ actions = []
+ for name, value in workflow.__class__.__dict__.items():
+ if isinstance(value, staticmethod) and hasattr(value, '_listen_events'):
+ actions.append(Action(name=name, exec=PythonFunction.from_callable(value.__func__),
+ listen_event_types=value._listen_events))
+ elif callable(value) and hasattr(value, '_listen_events'):
+ actions.append(Action(name=name, exec=PythonFunction.from_callable(value),
+ listen_event_types=value._listen_events))
+ return actions
diff --git a/python/flink_agents/runtime/local_execution_environment.py b/python/flink_agents/runtime/local_execution_environment.py
new file mode 100644
index 0000000..b0276a3
--- /dev/null
+++ b/python/flink_agents/runtime/local_execution_environment.py
@@ -0,0 +1,86 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+from typing import Any, Dict, List
+
+from flink_agents.api.execution_enviroment import AgentsExecutionEnvironment
+from flink_agents.api.workflow import Workflow
+from flink_agents.runtime.local_runner import LocalRunner
+
+
+class LocalExecutionEnvironment(AgentsExecutionEnvironment):
+ """Implementation of AgentsExecutionEnvironment for local execution environment."""
+
+ __input: List[Dict[str, Any]]
+ __output: List[Any]
+ __runner: LocalRunner = None
+ __executed: bool = False
+
+ def __init__(self) -> None:
+ """Init empty output list."""
+ self.__output = []
+
+ def from_list(self, input: list) -> 'AgentsExecutionEnvironment':
+ """Set input list of execution environment."""
+ self.__input = input
+ return self
+
+ def apply(self, workflow: Workflow) -> 'AgentsExecutionEnvironment':
+ """Create local runner to execute given workflow.
+
+ Doesn't support apply multiple workflows.
+ """
+ if self.__runner is not None:
+ err_msg = "LocalExecutionEnvironment doesn't support apply multiple workflows."
+ raise RuntimeError(err_msg)
+ self.__runner = LocalRunner(workflow)
+ return self
+
+ def to_list(self) -> list:
+ """Get output list of execution environment."""
+ return self.__output
+
+ def execute(self) -> None:
+ """Execute agents.
+
+ Doesn't support execute multiple times.
+ """
+ if self.__executed:
+ err_msg = "LocalExecutionEnvironment doesn't support execute multiple times."
+ raise RuntimeError(err_msg)
+ self.__executed = True
+ for input in self.__input:
+ self.__runner.run(**input)
+ outputs = self.__runner.get_outputs()
+ for output in outputs:
+ self.__output.append(output)
+
+
+def get_execution_environment(**kwargs: Dict[str, Any]) -> AgentsExecutionEnvironment:
+ """Factory function to create a local agents execution environment.
+
+ Parameters
+ ----------
+ **kwargs : Dict[str, Any]
+ The dict of parameters to configure the execution environment.
+
+ Returns:
+ -------
+ AgentsExecutionEnvironment
+ A configured agents execution environment instance.
+ """
+ return LocalExecutionEnvironment()
diff --git a/python/flink_agents/runtime/local_runner.py b/python/flink_agents/runtime/local_runner.py
new file mode 100644
index 0000000..bf1825f
--- /dev/null
+++ b/python/flink_agents/runtime/local_runner.py
@@ -0,0 +1,180 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+import logging
+import uuid
+from collections import deque
+from typing import Any, Dict, List
+
+from typing_extensions import override
+
+from flink_agents.api.event import Event, InputEvent, OutputEvent
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.workflow import Workflow
+from flink_agents.plan.workflow_plan import WorkflowPlan
+from flink_agents.runtime.workflow_runner import WorkflowRunner
+
+logging.basicConfig(
+ level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+class LocalRunnerContext(RunnerContext):
+ """Implementation of WorkflowRunnerContext for local workflow execution.
+
+ Attributes:
+ ----------
+ __workflow_plan : WorkflowPlan
+ Internal workflow plan for this context.
+ __key : Any
+ Unique identifier for the context, correspond to the key in flink KeyedStream.
+ events : deque[Event]
+ Queue of events to be processed in this context.
+ outputs : deque[Any]
+ Queue of outputs generated by workflow execution.
+ """
+
+ __workflow_plan: WorkflowPlan
+ __key: Any
+ events: deque[Event]
+
+ def __init__(self, workflow_plan: WorkflowPlan, key: Any) -> None:
+ """Initialize a new context with the given workflow and key.
+
+ Parameters
+ ----------
+ workflow : Workflow
+ Workflow plan used for this context.
+ key : Any
+ Unique context identifier, which is corresponding to the key in flink
+ KeyedStream.
+ """
+ self.__workflow_plan = workflow_plan
+ self.__key = key
+ self.events = deque()
+
+ @property
+ def key(self) -> Any:
+ """Get the unique identifier for this context.
+
+ Returns:
+ -------
+ Any
+ The unique identifier for this context.
+ """
+ return self.__key
+
+ @override
+ def send_event(self, event: Event) -> None:
+ """Send an event to the context's event queue and log it.
+
+ Parameters
+ ----------
+ event : Event
+ The event to be added to the queue.
+ """
+ logger.info("key: %s, send_event: %s", self.__key, event)
+ self.events.append(event)
+
+
+class LocalRunner(WorkflowRunner):
+ """Workflow runner implementation for local execution, which is
+ convenient for debugging.
+
+ Attributes:
+ ----------
+ __workflow_plan : WorkflowPlan
+ Internal workflow plan.
+ __keyed_contexts : dict[Any, LocalRunnerContext]
+ Dictionary of active contexts indexed by key.
+ __outputs:
+ Outputs generated by workflow execution.
+ """
+
+ __workflow_plan: WorkflowPlan
+ __keyed_contexts: Dict[Any, LocalRunnerContext]
+ __outputs: List[Dict[str, Any]]
+
+ def __init__(self, workflow: Workflow) -> None:
+ """Initialize the runner with the provided workflow.
+
+ Parameters
+ ----------
+ workflow : flink_agent_api.workflow.Workflow
+ The workflow class to convert and run.
+ """
+ self.__workflow_plan = WorkflowPlan.from_workflow(workflow)
+ self.__keyed_contexts = {}
+ self.__outputs = []
+
+ @override
+ def run(self, **data: Dict[str, Any]) -> Any:
+ """Execute the workflow to process the given data.
+
+ Parameters
+ ----------
+ **data : dict[str, Any]
+ input record from upstream.
+
+ Returns:
+ -------
+ key
+ The key of the input that was processed.
+ """
+ if 'key' in data:
+ key = data['key']
+ elif 'k' in data:
+ key = data['k']
+ else:
+ key = uuid.uuid4()
+
+ if key not in self.__keyed_contexts:
+ self.__keyed_contexts[key] = LocalRunnerContext(self.__workflow_plan, key)
+ context = self.__keyed_contexts[key]
+
+ if 'value' in data:
+ input_event = InputEvent(input=data['value'])
+ elif 'v' in data:
+ input_event = InputEvent(input=data['v'])
+ else:
+ msg = "Input data must be dict has 'v' or 'value' field"
+ raise RuntimeError(msg)
+
+ context.send_event(input_event)
+
+ while len(context.events) > 0:
+ event = context.events.popleft()
+ if isinstance(event, OutputEvent):
+ self.__outputs.append({key: event.output})
+ continue
+ for action in self.__workflow_plan.get_actions(type(event)):
+ logger.info(
+ "key: %s, performing action: %s", key, action.name
+ )
+ action.exec(event, context)
+ return key
+
+ def get_outputs(self) -> List[Dict[str, Any]]:
+ """Get the outputs generated by workflow execution.
+
+ Returns:
+ -------
+ List[Dict[str, Any]]
+ The workflow execution outputs.
+ """
+ return self.__outputs
diff --git a/python/flink_agents/plan/tests/test_action.py b/python/flink_agents/runtime/tests/__init__.py
similarity index 55%
copy from python/flink_agents/plan/tests/test_action.py
copy to python/flink_agents/runtime/tests/__init__.py
index e47a4ae..e154fad 100644
--- a/python/flink_agents/plan/tests/test_action.py
+++ b/python/flink_agents/runtime/tests/__init__.py
@@ -15,32 +15,3 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
-import pytest
-
-from flink_agents.api.event import InputEvent
-from flink_agents.plan.action import Action
-from flink_agents.plan.function import PythonFunction
-
-
-def legal_signature(event: InputEvent) -> None: # noqa: D103
- pass
-
-def illegal_signature(value: int) -> None: # noqa: D103
- pass
-
-def test_action_signature_legal() -> None: # noqa: D103
- Action(
- name="legal",
- exec=PythonFunction.from_callable(legal_signature),
- listen_event_types=[InputEvent],
- )
-
-def test_action_signature_illegal() -> None: # noqa: D103
- with pytest.raises(TypeError):
- Action(
- name="illegal",
- exec=PythonFunction.from_callable(illegal_signature),
- listen_event_types=[InputEvent],
- )
-
-
diff --git a/python/flink_agents/runtime/tests/test_local_execution_environment.py b/python/flink_agents/runtime/tests/test_local_execution_environment.py
new file mode 100644
index 0000000..7c70c67
--- /dev/null
+++ b/python/flink_agents/runtime/tests/test_local_execution_environment.py
@@ -0,0 +1,82 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+import pytest
+
+from flink_agents.api.decorators import action
+from flink_agents.api.event import Event, InputEvent, OutputEvent
+from flink_agents.api.execution_enviroment import AgentsExecutionEnvironment
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.workflow import Workflow
+
+
+class TestWorkflow1(Workflow): # noqa: D101
+ @action(InputEvent)
+ @staticmethod
+ def increment(event: Event, ctx: RunnerContext): #noqa D102
+ input = event.input
+ value = input + 1
+ ctx.send_event(OutputEvent(output=value))
+
+class TestWorkflow2(Workflow): # noqa: D101
+ @action(InputEvent)
+ @staticmethod
+ def decrease(event: Event, ctx: RunnerContext): #noqa D102
+ input = event.input
+ value = input - 1
+ ctx.send_event(OutputEvent(output=value))
+
+def test_local_execution_environment() -> None: # noqa: D103
+ env = AgentsExecutionEnvironment.get_execution_environment()
+
+ input_list = []
+ workflow = TestWorkflow1()
+
+ output_list = env.from_list(input_list).apply(workflow).to_list()
+
+ input_list.append({'key': 'bob', 'value': 1})
+ input_list.append({'k': 'john', 'v': 2})
+
+ env.execute()
+
+ assert output_list == [{'bob': 2}, {'john': 3}]
+
+def test_local_execution_environment_apply_multi_workflows() -> None: # noqa: D103
+ env = AgentsExecutionEnvironment.get_execution_environment()
+
+ input_list = []
+ workflow1 = TestWorkflow1()
+ workflow2 = TestWorkflow2()
+
+ with pytest.raises(RuntimeError):
+ env.from_list(input_list).apply(workflow1).apply(workflow2).to_list()
+
+
+def test_local_execution_environment_execute_multi_times() -> None: # noqa: D103
+ env = AgentsExecutionEnvironment.get_execution_environment()
+
+ input_list = []
+ workflow = TestWorkflow1()
+
+ env.from_list(input_list).apply(workflow).to_list()
+
+ input_list.append({'key': 'bob', 'value': 1})
+ input_list.append({'k': 'john', 'v': 2})
+
+ env.execute()
+ with pytest.raises(RuntimeError):
+ env.execute()
diff --git a/python/flink_agents/plan/workflow_plan.py b/python/flink_agents/runtime/workflow_runner.py
similarity index 57%
copy from python/flink_agents/plan/workflow_plan.py
copy to python/flink_agents/runtime/workflow_runner.py
index e54d0e5..250f7e7 100644
--- a/python/flink_agents/plan/workflow_plan.py
+++ b/python/flink_agents/runtime/workflow_runner.py
@@ -15,37 +15,30 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+from abc import ABC, abstractmethod
+from typing import Any, Dict
-from typing import Dict, List, Type
-from pydantic import BaseModel
+class WorkflowRunner(ABC):
+ """Abstract base class defining the interface for workflow execution.
-from flink_agents.api.event import Event
-from flink_agents.plan.action import Action
-
-
-class WorkflowPlan(BaseModel):
- """Workflow plan compiled from user defined workflow.
-
- Attributes:
- ----------
- actions : Dict[Type[Event], List[Action]]
- Mapping of event types to the list of Actions that listen to them.
+ Concrete implementations must implement the `run` method to handle workflow
+ execution logic specific to their use case.
"""
- actions: Dict[Type[Event], List[Action]]
-
- def get_actions(self, event_type: Type[Event]) -> List[Action]:
- """Get actions that listen to the specified event type.
+ @abstractmethod
+ def run(self, **data: Dict[str, Any]) -> Any:
+ """Execute the workflow and return the key of input.
Parameters
----------
- event_type : Type[Event]
- The event type to query.
+ **data : dict[str, Any]
+ input record from upstream.
Returns:
-------
- list[Action]
- List of Actions that will respond to this event type.
+ Any
+ The key of the input that was processed, will be automatically
+ generated if necessary.
"""
- return self.actions[event_type]
+