wenjin272 commented on code in PR #631:
URL: https://github.com/apache/flink-agents/pull/631#discussion_r3093851969
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/context/PythonRunnerContextImpl.java:
##########
@@ -48,16 +47,18 @@ public PythonRunnerContextImpl(
super(agentMetricGroup, mailboxThreadChecker, agentPlan,
resourceCache, jobIdentifier);
}
- @Override
- public void sendEvent(Event event) {
- Preconditions.checkState(
- event instanceof PythonEvent, "PythonRunnerContext only accept
Python event.");
- super.sendEvent(event);
- }
-
- public void sendEvent(String type, byte[] event, String eventJsonStr) {
- // this method will be invoked by PythonActionExecutor's python
interpreter.
- sendEvent(new PythonEvent(event, type, eventJsonStr));
+ /**
+ * Sends an event from Python to Java.
+ *
+ * <p>All Python events are serialized as JSON. The event is deserialized
into a Java {@link
+ * Event} and dispatched normally.
+ *
+ * @param eventJson JSON string with at least a {@code "type"} field
+ * @throws IOException if JSON parsing fails
+ */
+ public void sendUnifiedEvent(String eventJson) throws IOException {
Review Comment:
I think the name `sendUnifiedEvent` is a bit ambiguous. I suggest renaming
it to `sendEventJson` or `sentPythonEvent`.
##########
python/flink_agents/api/events/chat_event.py:
##########
@@ -36,9 +41,57 @@ class ChatRequestEvent(Event):
The expected output schema of the chat model final response. Optional.
"""
- model: str
- messages: List[ChatMessage]
- output_schema: OutputSchema | None = None
+ EVENT_TYPE: ClassVar[str] = "_chat_request_event"
+
+ def __init__(
+ self,
+ model: str,
+ messages: List[ChatMessage],
+ output_schema: OutputSchema | None = None,
+ ) -> None:
+ """Create a ChatRequestEvent."""
+ super().__init__(
+ type=ChatRequestEvent.EVENT_TYPE,
+ attributes={
+ "model": model,
+ "messages": messages,
+ "output_schema": output_schema,
+ },
+ )
+
+ @classmethod
+ @override
+ def from_event(cls, event: Event) -> "ChatRequestEvent":
+ assert "model" in event.attributes
+ assert "messages" in event.attributes
+ messages_raw = event.attributes["messages"]
+ messages = [
+ ChatMessage.model_validate(m) if isinstance(m, dict) else m
+ for m in messages_raw
+ ]
+ output_schema_raw = event.attributes.get("output_schema")
+ if isinstance(output_schema_raw, dict):
+ output_schema_raw = OutputSchema.model_validate(output_schema_raw)
+ return ChatRequestEvent(
+ model=event.attributes["model"],
+ messages=messages,
+ output_schema=output_schema_raw,
+ )
+
+ @property
+ def model(self) -> str:
+ """Return the chat model name."""
+ return self.attributes["model"]
Review Comment:
Shall we `return self.get_attr("model")`?
##########
python/flink_agents/api/events/chat_event.py:
##########
@@ -36,9 +41,57 @@ class ChatRequestEvent(Event):
The expected output schema of the chat model final response. Optional.
"""
- model: str
- messages: List[ChatMessage]
- output_schema: OutputSchema | None = None
+ EVENT_TYPE: ClassVar[str] = "_chat_request_event"
+
+ def __init__(
+ self,
+ model: str,
+ messages: List[ChatMessage],
+ output_schema: OutputSchema | None = None,
+ ) -> None:
+ """Create a ChatRequestEvent."""
+ super().__init__(
+ type=ChatRequestEvent.EVENT_TYPE,
+ attributes={
+ "model": model,
+ "messages": messages,
+ "output_schema": output_schema,
Review Comment:
We need to ensure consistency between the built-in events on the Python and
Java sides, so we must make the same changes on the Java side.
##########
api/src/main/java/org/apache/flink/agents/api/annotation/Action.java:
##########
@@ -31,23 +31,40 @@
* <p>This annotation specifies which event types the action should respond
to. The annotated method
* will be triggered when any of the specified event types occur.
*
+ * <p>Events can be specified either as concrete Event subclasses via {@link
#listenEvents()}, or as
+ * plain type strings via {@link #listenEventTypes()}. At least one of the two
must be non-empty.
+ *
* <p>Example usage:
*
* <pre>{@code
+ * // Class-based (existing pattern)
* @Action(listenEvents = {InputEvent.class, CustomEvent.class})
- * public void handleEvents(Event event) {
- * // Action logic here
- * }
+ * public void handleEvents(Event event) { ... }
+ *
+ * // String-based (unified events)
+ * @Action(listenEventTypes = {"MyCustomEvent", "AnotherEvent"})
+ * public void handleUnifiedEvents(Event event) { ... }
+ *
+ * // Mixed
+ * @Action(listenEvents = {InputEvent.class}, listenEventTypes =
{"MyCustomEvent"})
+ * public void handleMixed(Event event) { ... }
* }</pre>
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Action {
/**
- * List of event types that this action should respond to. At least one
event type must be
- * specified.
+ * List of event classes that this action should respond to.
*
* @return Array of Event classes that this action listens to
*/
- Class<? extends Event>[] listenEvents();
+ Class<? extends Event>[] listenEvents() default {};
Review Comment:
On the Python side, actions only support listening to event string
identifiers; on the Java side, we should align with this behavior.
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -19,36 +19,76 @@
package org.apache.flink.agents.api;
import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
/** Base class for all event types in the system. */
-public abstract class Event {
+public class Event {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
private final UUID id;
+ private final String type;
private final Map<String, Object> attributes;
/** The timestamp of the source record. */
private Long sourceTimestamp;
Review Comment:
I think we can remove this field to keep the align between python `Event`
and java `Event`
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -19,36 +19,76 @@
package org.apache.flink.agents.api;
import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
/** Base class for all event types in the system. */
-public abstract class Event {
+public class Event {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
private final UUID id;
+ private final String type;
private final Map<String, Object> attributes;
/** The timestamp of the source record. */
private Long sourceTimestamp;
+ /** Unified event with user-defined type and attributes. */
+ public Event(String type, Map<String, Object> attributes) {
+ this(UUID.randomUUID(), type, attributes);
+ }
+
+ /** Unified event with user-defined type and empty attributes. */
+ public Event(String type) {
+ this(type, new HashMap<>());
+ }
+
+ /** Subclasses that don't set an explicit type. */
public Event() {
- this(UUID.randomUUID(), new HashMap<>());
+ this(UUID.randomUUID(), null, new HashMap<>());
+ }
+
+ /** Subclasses with id and attributes but no explicit type. */
+ public Event(UUID id, Map<String, Object> attributes) {
+ this(id, null, attributes);
}
@JsonCreator
public Event(
@JsonProperty("id") UUID id,
+ @JsonProperty("type") String type,
@JsonProperty("attributes") Map<String, Object> attributes) {
this.id = id;
- this.attributes = attributes;
+ this.type = type;
+ this.attributes = attributes != null ? attributes : new HashMap<>();
}
public UUID getId() {
return id;
}
+ /** Returns the event type string used for routing. */
+ @JsonIgnore
+ public String getType() {
+ return type != null ? type : this.getClass().getName();
Review Comment:
I think we can directly require that the `type` is not null, which would
simplify the implementation of Event.
##########
python/flink_agents/api/events/event.py:
##########
@@ -30,27 +30,71 @@
from pyflink.common import Row
-class Event(BaseModel, ABC, extra="allow"):
+def _reconstruct_row_if_needed(data: Any) -> Any:
+ """Recursively reconstruct pyflink Row objects from their JSON-serialized
dicts.
+
+ Row objects are serialized as ``{"type": "Row", "values": [...], "fields":
[...]}``.
+ This helper walks dicts and lists to convert any such representation back
+ into a ``pyflink.common.Row``.
+ """
+ if isinstance(data, dict):
+ if data.get("type") == "Row" and "values" in data:
+ fields = data.get("fields")
+ values = data["values"]
+ if fields:
+ return Row(**dict(zip(fields, values, strict=False)))
+ return Row(*values)
+ return {k: _reconstruct_row_if_needed(v) for k, v in data.items()}
+ if isinstance(data, list):
+ return [_reconstruct_row_if_needed(item) for item in data]
+ return data
+
+
+class Event(BaseModel, extra="allow"):
"""Base class for all event types in the system.
+ This class serves dual purposes:
+
+ - **Unified events**: Instantiated directly with a user-defined ``type``
+ string and arbitrary key-value ``attributes``. No subclassing required.
+ - **Subclassed events**: Concrete subclasses (e.g., :class:`InputEvent`)
+ set a fixed ``type`` string and store data in ``attributes``.
+
Event allows extra properties, but these must be BaseModel instances or
JSON
serializable.
Attributes:
----------
id : UUID
Unique identifier for the event, generated deterministically based on
- event content. This ensures events with identical content have the same
- ID, which is critical for ActionStateStore divergence detection.
+ event content.
+ type : str | None
+ Event type string used for routing.
+ attributes : Dict[str, Any]
+ Key-value properties for the event data.
"""
+ _type_registry: ClassVar[Dict[str, Type["Event"]]] = {}
Review Comment:
It appears that this variable is intended to handle the deserialization of
subclasses, but it is not currently in use. I think the Event class doesn't
need to handle this; deserialization of subclass objects can be managed within
the subclass's from_event method. Because subclasses may have certain
non-trivial fields that require special handling during deserialization.
##########
python/flink_agents/api/tests/test_decorators.py:
##########
@@ -25,25 +23,25 @@
def test_action_decorator() -> None: # noqa D103
- @action(InputEvent)
+ @action("_input_event")
Review Comment:
I think we can use InputEvent.EVENT_TYPE, and the same applies to other
places where @action is used.
##########
python/flink_agents/api/events/chat_event.py:
##########
Review Comment:
We need to ensure consistency between the chat event on the Python and Java
sides, so we must make the same changes on the Java side. Other built-in events
also need to be aligned.
##########
python/flink_agents/runtime/python_java_utils.py:
##########
@@ -46,36 +45,44 @@
def convert_to_python_object(bytesObject: bytes) -> Any:
- """Used for deserializing Python objects."""
+ """Used for deserializing Python objects (e.g. durable execution
results)."""
return cloudpickle.loads(bytesObject)
-def _build_event_log_string(event: InputEvent | Event, event_type: str) -> str:
- try:
- payload = json.loads(event.model_dump_json())
- payload["eventType"] = event_type
- payload.setdefault("attributes", {})
- return json.dumps(payload)
- except Exception:
- return str(event)
+def convert_json_to_python_event(event_json: str) -> Event:
+ """Deserialize a JSON string into a typed Python Event object.
+
+ Called from Java via PythonActionExecutor to convert a Java Event
+ (serialized as JSON) into a Python Event for action dispatch.
+ If a subclass is registered for the event's type string, the event
+ is automatically reconstructed as that subclass via ``from_event``.
+ """
+ event = Event.from_json(event_json)
+ event_class = Event._type_registry.get(event.type)
+ if event_class is not None and hasattr(event_class, "from_event"):
+ return event_class.from_event(event)
Review Comment:
Similar to my comment in `Event`, the framework does not need to be aware of
the subclass types of `Event`. This would require the framework to handle many
details of serialization and deserialization.
Fro subclass types of `Event`, user need use `SubClass.from_event()` to
reconstruct the specific event object in action, like what you do in built-in
tool call action and chat action.
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -289,18 +224,23 @@ def set_long_term_memory(self, ltm:
InternalBaseLongTermMemory) -> None:
def send_event(self, event: Event) -> None:
"""Send an event to the agent for processing.
+ All events are serialized as JSON and sent via ``sendUnifiedEvent``
+ so that any language can reconstruct them.
+
Parameters
----------
event : Event
The event to be processed by the agent system.
"""
- class_path =
f"{event.__class__.__module__}.{event.__class__.__qualname__}"
- event_bytes = cloudpickle.dumps(event)
- event_json_str = _build_event_log_string(event, class_path)
+ event_json = event.model_dump_json()
try:
- self._j_runner_context.sendEvent(class_path, event_bytes,
event_json_str)
+ self._j_runner_context.sendUnifiedEvent(event_json)
except Exception as e:
- err_msg = "Failed to send event " + class_path + " to runner
context"
+ err_msg = (
+ "Failed to send event '"
+ + event.get_type()
+ + "' to runner context"
Review Comment:
I think we can also add the event json in `err_msg`.
--
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]
