weiqingy commented on code in PR #1114: URL: https://github.com/apache/flink-agents/pull/1114#discussion_r4052248299
########## api/src/main/java/org/apache/flink/agents/api/subagent/InputSchemas.java: ########## @@ -0,0 +1,89 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.agents.api.subagent; + +import com.fasterxml.jackson.databind.JsonMappingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; + +import javax.annotation.Nullable; + +/** + * Renders the input type a sub-agent declares as the JSON Schema a chat model is told about, so + * that a sub-agent which types its arguments does not also have to spell out their schema. + * + * <p>Rendering goes through the same Jackson generator {@code ReActAgent} renders a POJO output + * schema with, which keeps the two type-to-schema paths in this module on one implementation and + * adds no dependency. + */ +final class InputSchemas { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private InputSchemas() {} + + /** + * The schema of {@code type}, or {@code null} when the type states no shape a model could build + * a call from. That is {@link Object}, the type a sub-agent declares when it declares none, and + * any type that does not render as a JSON object, because the parameters of a callable must be + * one. + * + * @throws IllegalArgumentException if rendering the type fails, which is a declaration mistake + * worth failing on rather than dropping silently. + */ + @Nullable + static String fromType(@Nullable Class<?> type) { + if (type == null || type == Object.class) { + return null; + } + JsonNode schema = render(type); + return "object".equals(schema.path("type").asText()) ? schema.toString() : null; + } + + private static JsonNode render(Class<?> type) { + try { + return MAPPER.generateJsonSchema(type).getSchemaNode(); Review Comment: Thanks, the top-level fields match Python now. I tried a few more input types and found two cases where Java and Python still give different schemas: - A nested object gets no `required` list in Java (`InputSchemas.java:130-133`). The model might then skip a field Java needs, and the sub-agent would get a null. A `byte[]` inside a nested object also still shows up as an array of `byte`. - Some fields are marked required when they shouldn't be. Take `boolean isActive`: the schema calls it `active`, but the check at `:141` looks for `isActive`, doesn't find it, and marks it required. The same happens to a field renamed with `@JsonProperty` and to a getter with no field. What do you think about matching on the schema name and running the same step on nested objects? If nested types feel out of scope for this PR, narrowing contract 12 to top-level fields would work too. ########## 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: With the new operational metrics on main, a sub-agent call is reported as a tool named `_subagent_<name>`. The tool metrics only keep names of registered tools (`ToolExecutionMetricRecorder.java:64-67`), and a `_subagent_` name can never be one. So every sub-agent call is counted as `tool=unknown`, together with tool names the model made up. These calls also never report a start time, so they get no latency. Python looks the same. For someone watching a dashboard, sub-agent calls and bad tool calls would look alike, and what are likely the slowest calls would be missing from latency. How would you like sub-agent calls to show up? For example, they could report a start time and get their own group. I found this by reading the code and haven't run it, so please correct me if I missed a path. ########## api/src/main/java/org/apache/flink/agents/api/subagent/SubagentSetup.java: ########## @@ -19,22 +19,105 @@ package org.apache.flink.agents.api.subagent; import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; import org.apache.flink.agents.api.context.RunnerContext; import org.apache.flink.agents.api.resource.ResourceType; import org.apache.flink.agents.api.resource.SerializableResource; +import javax.annotation.Nullable; + /** * Caller-facing definition of a sub-agent, registered in the agent plan as an {@code AGENT} * resource. */ public abstract class SubagentSetup extends SerializableResource { + /** + * Prefix of the callable name a sub-agent is exposed to a chat model under. Tools are forbidden + * to register under this prefix, so a prefixed callable name unambiguously addresses a + * sub-agent and the executing side routes it to the {@code AGENT} namespace. + */ + public static final String CALLABLE_NAME_PREFIX = "_subagent_"; Review Comment: Thanks, the description matches the code now. nit: the API section says "One thing changes" for existing callers. A repeated tool name now also fails `open()` (`BaseChatModelSetup.java:121`), where before it was accepted. Would it be worth listing that too, so someone upgrading isn't surprised? -- 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]
