xintongsong commented on code in PR #48:
URL: https://github.com/apache/flink-agents/pull/48#discussion_r2194804828
##########
python/flink_agents/runtime/local_execution_environment.py:
##########
@@ -17,50 +17,92 @@
#################################################################################
from typing import Any, Dict, List
-from flink_agents.api.execution_enviroment import AgentsExecutionEnvironment
+from pyflink.common import TypeInformation
+from pyflink.datastream import DataStream, KeySelector
+from pyflink.table import Schema, StreamTableEnvironment, Table
+
+from flink_agents.api.execution_environment import (
+ AgentBuilder,
+ 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."""
+class LocalAgentBuilder(AgentBuilder):
+ """LocalAgentBuilder for building agent instance."""
+ __env: "LocalExecutionEnvironment"
__input: List[Dict[str, Any]]
__output: List[Any]
__runner: LocalRunner = None
__executed: bool = False
- def __init__(self) -> None:
+ def __init__(
+ self, env: "LocalExecutionEnvironment", input: List[Dict[str, Any]]
+ ) -> None:
"""Init empty output list."""
- self.__output = []
-
- def from_list(self, input: list) -> 'AgentsExecutionEnvironment':
- """Set input list of execution environment."""
+ self.__env = env
self.__input = input
- return self
+ self.__output = []
- def apply(self, workflow: Workflow) -> 'AgentsExecutionEnvironment':
+ def apply(self, workflow: Workflow) -> AgentBuilder:
"""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."
+ err_msg = "LocalAgentBuilder doesn't support apply multiple
workflows."
Review Comment:
Do we support multiple agents in the same job? If yes, `env.set_agent`
doesn't make sense. If not, we should check in the env, not here.
##########
python/flink_agents/api/execution_environment.py:
##########
@@ -0,0 +1,172 @@
+################################################################################
+# 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, Optional
+
+from pyflink.common import TypeInformation
+from pyflink.datastream import DataStream, KeySelector,
StreamExecutionEnvironment
+from pyflink.table import Schema, StreamTableEnvironment, Table
+
+from flink_agents.api.workflow import Workflow
+
+
+class AgentBuilder(ABC):
+ """Builder for integrating agent with input and output."""
+
+ @abstractmethod
+ def apply(self, workflow: Workflow) -> "AgentBuilder":
+ """Set workflow of AgentBuilder.
+
+ Parameters
+ ----------
+ workflow : Workflow
+ The workflow user defined to run in execution environment.
+ """
+
+ @abstractmethod
+ def to_list(self) -> List[Dict[str, Any]]:
+ """Get output list of agent execution.
+
+ The element in the list is a dict like {'key': output}.
+
+ Returns:
+ -------
+ list
+ Outputs of agent execution.
+ """
+
+ @abstractmethod
+ def to_datastream(self) -> DataStream:
+ """Get output datastream of agent execution.
+
+ Returns:
+ -------
+ DataStream
+ Output datastream of agent execution.
+ """
+
+ # TODO: auto generate output_type.
+ @abstractmethod
+ def to_table(self, schema: Schema, output_type: TypeInformation) -> Table:
+ """Get output table of agent execution.
+
+ Parameters
+ ----------
+ schema : Schema
+ Indicate schema of the output table.
+ output_type : TypeInformation
+ Indicate schema corresponding type information.
+
+ Returns:
+ -------
+ Table
+ Output table of agent execution.
+ """
+
+
+class AgentsExecutionEnvironment(ABC):
+ """Base class for workflow execution environment."""
+
+ @classmethod
+ def get_execution_environment(
Review Comment:
Why is this a class method, rather than a static method?
##########
python/flink_agents/runtime/local_execution_environment.py:
##########
@@ -17,50 +17,92 @@
#################################################################################
from typing import Any, Dict, List
-from flink_agents.api.execution_enviroment import AgentsExecutionEnvironment
+from pyflink.common import TypeInformation
+from pyflink.datastream import DataStream, KeySelector
+from pyflink.table import Schema, StreamTableEnvironment, Table
+
+from flink_agents.api.execution_environment import (
+ AgentBuilder,
+ 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."""
+class LocalAgentBuilder(AgentBuilder):
+ """LocalAgentBuilder for building agent instance."""
+ __env: "LocalExecutionEnvironment"
__input: List[Dict[str, Any]]
__output: List[Any]
__runner: LocalRunner = None
__executed: bool = False
- def __init__(self) -> None:
+ def __init__(
+ self, env: "LocalExecutionEnvironment", input: List[Dict[str, Any]]
+ ) -> None:
"""Init empty output list."""
- self.__output = []
-
- def from_list(self, input: list) -> 'AgentsExecutionEnvironment':
- """Set input list of execution environment."""
+ self.__env = env
self.__input = input
- return self
+ self.__output = []
- def apply(self, workflow: Workflow) -> 'AgentsExecutionEnvironment':
+ def apply(self, workflow: Workflow) -> AgentBuilder:
"""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."
+ err_msg = "LocalAgentBuilder doesn't support apply multiple
workflows."
raise RuntimeError(err_msg)
self.__runner = LocalRunner(workflow)
return self
- def to_list(self) -> list:
+ def to_list(self) -> List[Dict[str, Any]]:
"""Get output list of execution environment."""
+ self.__env.set_agent(self.__input, self.__output, self.__runner)
Review Comment:
Why not setting in `apply`? What if the agent does not have any output?
##########
python/flink_agents/api/event.py:
##########
@@ -19,30 +19,19 @@
from typing import Any
from uuid import UUID, uuid4
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, Field
class Event(BaseModel, ABC, extra="allow"):
- """Base class for all event types in the system. Event allow extra
properties, but
- these properties are required isinstance of BaseModel, or json
serializable.
+ """Base class for all event types in the system.
Attributes:
----------
id : UUID
Unique identifier for the event, automatically generated using uuid4.
"""
- id: UUID = Field(default_factory=uuid4)
-
- @model_validator(mode='after')
- def validate_extra(self) -> 'Event':
- """Ensure init fields is serializable."""
- self.model_dump_json()
- return self
- def __setattr__(self, name: str, value: Any) -> None:
- super().__setattr__(name, value)
- # Ensure added property can be serialized.
- self.model_dump_json()
Review Comment:
Not sure about removing these. If this is because some types currently won't
pass the validation, we can bypass those types and leave a todo item.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
