weiqingy commented on code in PR #924:
URL: https://github.com/apache/flink-agents/pull/924#discussion_r3745291890
##########
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:
This skip drops the record and writes nothing, so it is the one place the
reader discards something without leaving a warning behind. If a user's own
Event happens to use one of these type strings it disappears from the tree, and
its children then report `MISSING_PARENT` (`:316-324`) for a parent that is
sitting right there in the log.
Nothing reserves the `_` prefix today: `Event.java:82-84` only checks for
null or empty, and Python's `event.py:59-71` has no check at all. So
`_execution_*` is a convention, the same way `_input_event` is
(`EventUtil.java:27-29` matches that one on the raw string too). None of that
changes in this PR. What does change is that these names now live in two
languages: `EntityTypes`, `ProblemCategories` and `ToolExecutionMetadataKeys`
all got mirrored into Python, but these four are typed out again at `:26-33`
and listed a third time in `monitoring.md:395-398`, with nothing keeping the
copies in step.
So is the namespace part of the contract? If it is, would a shared constant
plus a warning on this skip be worth adding?
##########
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:
`recordChatTokenMetrics` sits outside the durable boundary. On a cache hit,
`durableExecuteCompletionOnly` returns at `RunnerContextImpl.java:384` and
never reaches the real call at `:390`, but the cached `ChatMessage` still
carries its `promptTokens` / `completionTokens`. So the documented counters
(`monitoring.md:50-51`) go up for a model call that never happened. Python does
the same (`chat_model_action.py:377-388`).
The window is small. A finished action is skipped whole at
`ActionExecutionOperator.java:362-383` and never gets here, so this is only the
replay of an action that did not finish, and only when
`actionStateStoreBackend` is set (default `null`,
`AgentConfigOptions.java:65-66`).
It also predates this PR: `main:375` is the same line in the same spot, so
there is nothing to change here. Noting it for a follow-up issue, since these
counters are user-facing.
##########
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:
Discussion #929 asks for the resolved model, the provider, and per-call
token usage to land in `entityMetadata`. Today neither report carries any: this
line and `:389` both pass just the model, so an LLM record has no
`entityMetadata` at all, and `entityName` is only the alias
`_default_chat_model`. One line further down at `:390` the model name and the
token counts are both in hand, and they go only to the cumulative counters.
Python is the same (`chat_model_action.py:375`, `:377-388`).
Model and provider look easy, since both are known before the call, so the
start and the end report would carry the same value. Usage is the tricky one:
`entityMetadata` is part of what matches an end report back to its start
(`RunnerContextImpl.java:264-286`, `ReportedExecutionKey.java:56-67`), and
`ExecutionReporter.java:28-29` asks it to stay the same across both. A number
known only after the response would not match, so the end report would come out
under a fresh `executionId` instead of closing the one it started.
The record shape is being settled here, so would it be worth carrying model
and provider now, and letting usage find its own home later?
##########
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:
Nit: `event_id_value` is checked at `:122`, but this return and the one at
`:125` both pass `None` as the id, so a record with a perfectly good `eventId`
shows up as `MALFORMED_RECORD` with nothing to search on. This runs after both
shape branches close, so it hits flat and legacy records alike. The lineage
check just below at `:134-138` does pass the id through.
--
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]
