weiqingy commented on code in PR #924:
URL: https://github.com/apache/flink-agents/pull/924#discussion_r3654253842
##########
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:
`started` here and `succeeded` at `:389` bracket `ctx.durableExecute(...)`,
but on replay that returns the cached result and short-circuits
(`RunnerContextImpl.java:383-385`) while the action body still re-runs. Your
own `testEarlierCheckpointReplayKeepsDurableState:1039` pins that:
`DURABLE_CALL_COUNTER` stays at 1 while the output is produced again.
So after a fine-grained recovery the log gets a complete `started` plus
`succeeded` pair, with a fresh `executionId`, for a model call that never
happened. `ToolCallAction` has the same shape. Since counting LLM calls is the
first thing anyone does with this log, cost dashboards would over-report spend
and under-report latency.
What makes me read this as unintended: the Action level already has
`ExecutionLifecycleEvents.executionReused()` for exactly this case, but nothing
below Action can emit it, since `ExecutionReporter` has no such method.
Could a cached durable result surface a reused signal that the reporters
emit instead of `started`/`succeeded`? Or does the reporting want to move
inside the durable boundary? And if it's out of scope here, would naming it in
the monitoring doc be enough for now?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/FileEventLogger.java:
##########
@@ -199,24 +207,18 @@ public void append(EventContext context, Event event)
throws Exception {
}
ObjectNode rootNode = (ObjectNode) tree;
- // Truncate the event subtree at STANDARD level.
+ // Truncate event attributes at STANDARD level.
if (level == EventLogLevel.STANDARD && truncator != null) {
- JsonNode eventNode = rootNode.get("event");
- if (eventNode instanceof ObjectNode) {
- boolean truncated = truncator.truncate((ObjectNode) eventNode);
+ JsonNode attributesNode = rootNode.get("eventAttributes");
Review Comment:
Passing `rootNode.get("eventAttributes")` into `truncator.truncate(...)`
narrows truncation in two ways. Same block at `Slf4jEventLogger.java:179-181`,
so both loggers are affected.
The protections now point at the wrong names. `JsonTruncator` is untouched
by this PR and still skips `PROTECTED_FIELDS = {eventType, id, attributes}` at
`isTopLevel` (`:55-56`, `:117-119`). Those were envelope names; under the flat
shape they are user attribute names, so an attribute called `id` escapes
`event-log.standard.max-string-length` at STANDARD.
`JsonTruncatorTest.testProtectedFields:136-161` still pins the old shape and
passes, so CI won't catch it. Worth noting #923 adds `upstreamEventId` and
`upstreamActionName` to that same set, which won't take effect under the flat
shape either.
Separately, `entityMetadata` sits as a sibling of `eventAttributes`
(`EventLogRecordJsonSerializer.java:73`), so the truncator never sees it. It's
fed by the public `ToolExecutionMetadataProvider` hook, and
`LoadSkillTool.java:68-80` copies the model-supplied `name` and `path` straight
in, so an LLM-controlled value lands unbounded.
Would running the truncator over the whole record, with the new framework
field names protected, close both at once? That was my first instinct, though I
may be missing why the narrowing was deliberate. Either way, would
`testProtectedFields` want re-pointing at what the production path now passes
in?
##########
plan/src/test/java/org/apache/flink/agents/plan/actions/ChatModelActionRetryTest.java:
##########
@@ -127,6 +128,129 @@ void chatSucceedsWithoutRetry_retryCountIsZero() throws
Exception {
verify(mockActionMetricGroup, never()).getSubGroup(anyString(),
anyString());
}
+ @Test
+ void chatReportsLlmExecution() throws Exception {
+ RunnerContext reportingCtx = reportingRunnerContext();
+ BaseChatModelSetup chatModel =
configureReportingChatContext(reportingCtx);
+ when(chatModel.chat(any(), any(), any()))
+ .thenReturn(new ChatMessage(MessageRole.ASSISTANT, "hello"));
+
+ ChatModelAction.chat(
+ UUID.randomUUID(),
+ "test-model",
+ List.of(new ChatMessage(MessageRole.USER, "hi")),
+ Map.of(),
+ null,
+ reportingCtx);
+
+ ExecutionReporter reporter = (ExecutionReporter) reportingCtx;
+ verify(reporter)
+ .reportExecutionStarted(ExecutionReporter.EntityTypes.LLM,
"test-model", Map.of());
+ verify(reporter)
+ .reportExecutionSucceeded(
+ ExecutionReporter.EntityTypes.LLM, "test-model",
Map.of());
+ }
+
+ @Test
+ void chatReportsStructuredOutputParserExecution() throws Exception {
+ RunnerContext reportingCtx = reportingRunnerContext();
+ BaseChatModelSetup chatModel =
configureReportingChatContext(reportingCtx);
+ when(chatModel.chat(any(), any(), any()))
+ .thenReturn(new ChatMessage(MessageRole.ASSISTANT,
"{\"answer\":\"42\"}"));
+
+ ChatModelAction.chat(
+ UUID.randomUUID(),
+ "test-model",
+ List.of(new ChatMessage(MessageRole.USER, "hi")),
+ Map.of(),
+ Map.class,
+ reportingCtx);
+
+ ExecutionReporter reporter = (ExecutionReporter) reportingCtx;
+ verify(reporter)
+ .reportExecutionStarted(
+ ExecutionReporter.EntityTypes.PARSER,
Agent.STRUCTURED_OUTPUT, Map.of());
+ verify(reporter)
+ .reportExecutionSucceeded(
+ ExecutionReporter.EntityTypes.PARSER,
Agent.STRUCTURED_OUTPUT, Map.of());
+ }
+
+ @Test
+ void chatRetriesStructuredOutputParseErrorWithoutFailingLlm() throws
Exception {
+ RunnerContext reportingCtx = reportingRunnerContext();
+ BaseChatModelSetup chatModel =
configureReportingChatContext(reportingCtx);
+ when(reportingCtx.getConfig())
+ .thenReturn(readableConfig(Agent.ErrorHandlingStrategy.RETRY,
1, 0));
+ when(chatModel.chat(any(), any(), any()))
+ .thenReturn(
+ new ChatMessage(MessageRole.ASSISTANT, "not-json"),
+ new ChatMessage(MessageRole.ASSISTANT,
"{\"answer\":\"42\"}"));
+
+ ChatModelAction.chat(
+ UUID.randomUUID(),
+ "test-model",
+ List.of(new ChatMessage(MessageRole.USER, "hi")),
+ Map.of(),
+ Map.class,
+ reportingCtx);
+
+ verify(chatModel, times(2)).chat(any(), any(), any());
+
+ ExecutionReporter reporter = (ExecutionReporter) reportingCtx;
+ verify(reporter, times(2))
+ .reportExecutionStarted(ExecutionReporter.EntityTypes.LLM,
"test-model", Map.of());
+ verify(reporter, times(2))
+ .reportExecutionSucceeded(
+ ExecutionReporter.EntityTypes.LLM, "test-model",
Map.of());
+ verify(reporter)
+ .reportExecutionFailed(
+ eq(ExecutionReporter.EntityTypes.PARSER),
+ eq(Agent.STRUCTURED_OUTPUT),
+ eq(Map.of()),
+ any(Exception.class),
+
eq(ExecutionReporter.ProblemCategories.MODEL_OUTPUT_PARSE_ERROR));
+ verify(reporter, never())
+ .reportExecutionFailed(
+ eq(ExecutionReporter.EntityTypes.LLM),
+ eq("test-model"),
+ eq(Map.of()),
+ any(Throwable.class),
+ any());
+ }
+
+ @Test
+ void chatReportsEachRetriedModelInvocation() throws Exception {
Review Comment:
Anchoring here since this is where the contract gets pinned:
`chatReportsEachRetriedModelInvocation` asserts `times(2)` on
`reportExecutionStarted`. `started` sits inside the retry loop
(`ChatModelAction.java:371-373`), and Python matches
(`chat_model_action.py:352-375`), so both languages emit one execution per
attempt.
On the design discussion the contract landed on one execution per framework
model invocation, with retries collapsing into it and Parser recorded
separately. Parser is implemented that way (`:457-474`); the retry half went
the other way.
Downstream, a `retryCount` of 3 on a successful request reads as 4 LLM
executions at a 25% success rate, and nothing in the record separates "attempt
2" from "a second call". `ChatResponseEvent` already carries `retryCount`, so
the two views disagree.
Which way did you land in the end? Two shapes if either helps: hoist
`started`/`succeeded` outside the loop and put attempts on the single terminal,
or keep per-attempt and add an ordinal plus a shared correlation id. Either
way, would it be worth pinning in `ExecutionReporter`'s javadoc, now that it's
permanent public API?
--
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]
