joeyutong commented on code in PR #924:
URL: https://github.com/apache/flink-agents/pull/924#discussion_r3759009444
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java:
##########
@@ -368,16 +370,27 @@ public ChatMessage call() throws Exception {
for (int attempt = 0; attempt < numRetries + 1; attempt++) {
try {
- response =
- chatAsync
- ? ctx.durableExecuteAsync(callable)
- : ctx.durableExecute(callable);
+ ExecutionReporters.started(ctx,
ExecutionReporter.EntityTypes.LLM, model);
Review Comment:
Added `entityMetadata.model` to Java and Python LLM lifecycle reports. It is
the configured/requested model or deployment identifier, while `entityName`
remains the ChatModel Resource name; start and terminal reports use the same
metadata. I left provider out because the current ChatModel abstraction does
not expose a stable provider identity across integrations, and token usage
remains outside lifecycle matching metadata for now.
##########
python/flink_agents/cli/trace_tree.py:
##########
@@ -93,28 +162,15 @@ def read_event_records(
for log_file in log_files:
for record in read_json_objects(log_file, warnings):
- event_id: str | None = None
- invalid_reason: str | None = None
- if not isinstance(record, dict):
- invalid_reason = "record must be a JSON object"
- else:
- event = record.get("event")
- event_type = record.get("eventType")
- if not isinstance(event, dict):
- invalid_reason = "field 'event' must be a JSON object"
- elif not isinstance(event.get("id"), str) or not event["id"]:
- invalid_reason = "field 'event.id' must be a non-empty
string"
- elif not isinstance(event_type, str) or not event_type:
- invalid_reason = "field 'eventType' must be a non-empty
string"
- else:
- event_id = event["id"]
- for field_name in ("upstreamEventId",
"upstreamActionName"):
- field_value = event.get(field_name)
- if field_value is not None and not
isinstance(field_value, str):
- invalid_reason = (
- f"field 'event.{field_name}' must be a string
or null"
- )
- break
+ if (
+ isinstance(record, dict)
+ and record.get("eventType") in EXECUTION_LIFECYCLE_EVENT_TYPES
Review Comment:
Documented the four lifecycle types as framework-reserved and moved Python
consumers to shared constants. Trace Tree now ignores only records whose type,
expected status, and execution identity match the lifecycle shape; a
reserved-name business Event is retained and emits `RESERVED_EVENT_TYPE` with
its Event ID.
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java:
##########
@@ -368,16 +370,27 @@ public ChatMessage call() throws Exception {
for (int attempt = 0; attempt < numRetries + 1; attempt++) {
try {
- response =
- chatAsync
- ? ctx.durableExecuteAsync(callable)
- : ctx.durableExecute(callable);
+ ExecutionReporters.started(ctx,
ExecutionReporter.EntityTypes.LLM, model);
+ try {
+ response =
+ chatAsync
+ ? ctx.durableExecuteAsync(callable)
+ : ctx.durableExecute(callable);
+ Objects.requireNonNull(response, "ChatModel returned a
null response.");
+ } catch (Exception modelError) {
+ ExecutionReporters.failed(
+ ctx,
+ ExecutionReporter.EntityTypes.LLM,
+ model,
+ modelError,
+
ExecutionReporter.ProblemCategories.MODEL_CALL_FAILED);
+ throw modelError;
+ }
+ ExecutionReporters.succeeded(ctx,
ExecutionReporter.EntityTypes.LLM, model);
recordChatTokenMetrics(chatModel, response);
Review Comment:
Agreed. This behavior predates the trace change and remains unchanged in
this PR; I will track durable-replay token overcounting as follow-up work.
##########
python/flink_agents/cli/trace_tree.py:
##########
@@ -82,6 +90,67 @@ def read_json_objects(path: Path, warnings: list[dict[str,
Any]]) -> Iterator[An
yield record
+def _normalize_event_record(
+ record: Any,
+) -> tuple[dict[str, Any] | None, str | None, str | None]:
+ """Normalize flat and legacy Event Log records for lineage
reconstruction."""
+ if not isinstance(record, dict):
+ return None, None, "record must be a JSON object"
+
+ event_type = record.get("eventType")
+ flat_record = "eventId" in record or "eventAttributes" in record
+ if flat_record:
+ event_id_value = record.get("eventId")
+ event_attributes = record.get("eventAttributes")
+ upstream_event_id = record.get("upstreamEventId")
+ upstream_action_name = record.get("upstreamActionName")
+ event_id_field = "eventId"
+ attributes_field = "eventAttributes"
+ lineage_prefix = ""
+ else:
+ event = record.get("event")
+ if not isinstance(event, dict):
+ return None, None, "field 'event' must be a JSON object"
+ event_id_value = event.get("id")
+ event_attributes = event.get("attributes")
+ upstream_event_id = event.get("upstreamEventId")
+ upstream_action_name = event.get("upstreamActionName")
+ event_id_field = "event.id"
+ attributes_field = "event.attributes"
+ lineage_prefix = "event."
+
+ if not isinstance(event_id_value, str) or not event_id_value:
+ return None, None, f"field '{event_id_field}' must be a non-empty
string"
+ if not isinstance(event_type, str) or not event_type:
+ return None, None, "field 'eventType' must be a non-empty string"
+ if not isinstance(event_attributes, dict):
+ return None, None, f"field '{attributes_field}' must be a JSON object"
Review Comment:
Fixed for both flat and legacy records: once a valid Event ID has been
parsed, later shape-validation warnings retain it. Added coverage for invalid
`eventType` and invalid attributes.
--
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]
