weiqingy commented on code in PR #1114:
URL: https://github.com/apache/flink-agents/pull/1114#discussion_r4088073480
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java:
##########
@@ -384,6 +426,180 @@ private static void recordInlineResponse(
}
}
+ private static void dispatchAgentExecution(
+ ToolCallExecution execution,
+ RunnerContext ctx,
+ Map<String, Boolean> success,
+ Map<String, String> error,
+ Map<String, ToolResponse> responses)
+ throws InterruptedException {
+ try {
+ // submit() and await() already run through durable execution
inside the setup, so
+ // wrapping the call again here would nest durable cursors.
+ SubagentResult result = execution.agent.submit(ctx,
execution.agentArguments).await();
+ recordAgentResult(execution, result, ctx, success, error,
responses);
+ } catch (InterruptedException e) {
+ // A cancellation, not a sub-agent failure: propagate it exactly
like the tool paths do
+ // (#1111) so the caller skips sendEvent instead of folding the
cancellation into a
+ // tool-error response and driving a further chat call off it.
+ Thread.currentThread().interrupt();
+ throw e;
+ } catch (Exception e) {
+ recordAgentFailure(execution, e, ctx, success, error, responses);
+ } catch (StackOverflowError e) {
+ // Normalizing a result nested deeper than the stack allows
overflows it; the cycle
+ // guard reports a plain cycle earlier, so what reaches here is a
result too deep to
+ // walk. A StackOverflowError is an Error, so it would escape the
catch above and fail
+ // the job; fold it into the same failed delegation the model can
read and correct.
+ recordAgentFailure(
+ execution,
+ new RuntimeException(
+ "Sub-agent result is cyclic or too deeply nested
to normalize as"
+ + " JSON",
+ e),
+ ctx,
+ success,
+ error,
+ responses);
+ }
+ }
+
+ /**
+ * Runs the sub-agent calls concurrently rather than one by one: every
{@code submit} is issued
+ * first, so the async setups start their remote runs together, and each
future is then awaited.
+ * Submitting and awaiting both follow {@code agentExecutions} order,
which is the deterministic
+ * tool-call order, so the setup's id allocator hands out the same ids on
every replay and the
+ * durable keys stay stable. The per-call try/catch keeps the isolation
the serial path had: one
+ * sub-agent failing, at submit or at await, is recorded and reported
without stopping the rest.
+ *
+ * <p>A cancellation is the one thing that is not isolated: like the tool
batch (#1111), an
+ * {@link InterruptedException} propagates so the caller skips sendEvent,
and every handle
+ * submitted but no longer going to be awaited is cancelled first, so the
concurrent dispatch
+ * leaves no in-flight remote run dangling on the way out.
+ */
+ private static void dispatchAgentExecutions(
+ List<ToolCallExecution> agentExecutions,
+ RunnerContext ctx,
+ Map<String, Boolean> success,
+ Map<String, String> error,
+ Map<String, ToolResponse> responses)
+ throws InterruptedException {
+ // submit() runs through durable execution inside the setup, so it is
not wrapped here.
+ List<ToolCallExecution> submitted = new
ArrayList<>(agentExecutions.size());
+ List<SubagentFuture> futures = new ArrayList<>(agentExecutions.size());
+ for (ToolCallExecution execution : agentExecutions) {
+ try {
+ futures.add(execution.agent.submit(ctx,
execution.agentArguments));
+ submitted.add(execution);
+ } catch (InterruptedException e) {
+ // Cancelled mid-submit: everything submitted so far is now
never going to be
+ // awaited, so cancel it before propagating like the tool
paths (#1111).
+ cancelFrom(futures, 0);
+ Thread.currentThread().interrupt();
+ throw e;
+ } catch (Exception e) {
+ recordAgentFailure(execution, e, ctx, success, error,
responses);
+ }
+ }
+ for (int i = 0; i < futures.size(); i++) {
+ ToolCallExecution execution = submitted.get(i);
+ try {
+ SubagentResult result = futures.get(i).await();
+ recordAgentResult(execution, result, ctx, success, error,
responses);
+ } catch (InterruptedException e) {
+ // Cancelled mid-await: this handle and every later one were
submitted but will not
+ // be awaited, so cancel them before propagating like the tool
paths (#1111).
+ cancelFrom(futures, i);
+ Thread.currentThread().interrupt();
+ throw e;
+ } catch (Exception e) {
+ recordAgentFailure(execution, e, ctx, success, error,
responses);
+ } catch (StackOverflowError e) {
+ // As in the serial path: an overflow while normalizing one
result is an Error that
+ // would otherwise escape and fail the job mid-batch, so it is
folded into that
+ // call's failed delegation and the remaining handles are
still awaited.
+ recordAgentFailure(
+ execution,
+ new RuntimeException(
+ "Sub-agent result is cyclic or too deeply
nested to normalize"
+ + " as JSON",
+ e),
+ ctx,
+ success,
+ error,
+ responses);
+ }
+ }
+ }
+
+ /**
+ * Requests cancellation of every handle from {@code from} onward, e.g. on
a mid-batch cancel.
+ */
+ private static void cancelFrom(List<SubagentFuture> futures, int from) {
+ for (int i = from; i < futures.size(); i++) {
+ futures.get(i).cancel();
+ }
+ }
+
+ private static void recordAgentFailure(
+ ToolCallExecution execution,
+ Exception e,
+ RunnerContext ctx,
+ Map<String, Boolean> success,
+ Map<String, String> error,
+ Map<String, ToolResponse> responses) {
+ recordExecutionException(execution, e, success, error, responses);
+ ExecutionReporters.failed(
+ ctx,
+ ExecutionReporter.EntityTypes.TOOL,
+ execution.name,
+ execution.entityMetadata,
+ e,
+ ExecutionReporter.ProblemCategories.TOOL_CALL_FAILED);
+ }
+
+ private static void recordAgentResult(
+ ToolCallExecution execution,
+ SubagentResult result,
+ RunnerContext ctx,
+ Map<String, Boolean> success,
+ Map<String, String> error,
+ Map<String, ToolResponse> responses)
+ throws Exception {
+ if (result.isSuccess()) {
+ success.put(execution.id, true);
+ responses.put(
+ execution.id,
+ ToolResponse.success(
+ ToolResultUtils.toChatMessageContent(
+ ToolResultUtils.normalizeAgentResult(
+ result.getResult(),
execution.agent.getResultType()))));
+ ExecutionReporters.succeeded(
Review Comment:
Thanks for resolving the comments. Nothing further from me.
--
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]
