da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3665417925
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ToolCallAction.java:
##########
@@ -79,55 +103,160 @@ public static void processToolRequest(Event event,
RunnerContext ctx) {
diagnosticError = e.getMessage();
}
- if (tool != null) {
- try {
- // Framework-owned injected args must win over
model-provided values so hidden
- // context such as tenant ids cannot be spoofed by a tool
call payload.
- mergedArguments.putAll(resolveInjectedArguments(tool,
ctx));
- ToolResponse response;
- final Tool toolRef = tool;
- final Map<String, Object> callArguments = mergedArguments;
- DurableCallable<ToolResponse> callable =
- new DurableCallable<>() {
- @Override
- public String getId() {
- return "tool-call";
- }
-
- @Override
- public Class<ToolResponse> getResultClass() {
- return ToolResponse.class;
- }
-
- @Override
- public ToolResponse call() throws Exception {
- return toolRef.call(new
ToolParameters(callArguments));
- }
- };
- response =
- toolCallAsync
- ? ctx.durableExecuteAsync(callable)
- : ctx.durableExecute(callable);
- success.put(id, response.isSuccess());
- responses.put(id, response);
- if (!response.isSuccess() && response.getError() != null) {
- error.put(id, response.getError());
- }
- } catch (Exception e) {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s
execute failed.", name)));
- error.put(id, e.getMessage());
- }
- } else {
- success.put(id, false);
- responses.put(
- id, ToolResponse.error(String.format("Tool %s does not
exist.", name)));
- error.put(id, diagnosticError != null ? diagnosticError :
"Tool does not exist.");
+ if (tool == null) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s does not
exist.", name)),
+ diagnosticError != null ? diagnosticError : "Tool does
not exist.",
+ success,
+ error,
+ responses);
+ continue;
}
+
+ try {
+ // Framework-owned injected args must win over model-provided
values so hidden
+ // context such as tenant ids cannot be spoofed by a tool call
payload.
+ mergedArguments.putAll(resolveInjectedArguments(tool, ctx));
+ } catch (Exception e) {
+ recordInlineResponse(
+ id,
+ ToolResponse.error(String.format("Tool %s execute
failed.", name)),
+ e.getMessage(),
+ success,
+ error,
+ responses);
+ continue;
+ }
+
+ final Tool toolRef = tool;
+ final Map<String, Object> callArguments = mergedArguments;
+ DurableCallable<ToolResponse> callable =
+ new DurableCallable<>() {
+ @Override
+ public String getId() {
+ return "tool-call-" + id;
+ }
+
+ @Override
+ public Class<ToolResponse> getResultClass() {
+ return ToolResponse.class;
+ }
+
+ @Override
+ public ToolResponse call() throws Exception {
+ return toolRef.call(new
ToolParameters(callArguments));
+ }
+ };
+ executions.add(new ToolCallExecution(id, name, callable));
+ }
+ return executions;
+ }
+
+ private static void executeParallel(
+ List<ToolCallExecution> executions,
+ RunnerContext ctx,
+ Map<String, Boolean> success,
+ Map<String, String> error,
+ Map<String, ToolResponse> responses) {
+ List<DurableCallable<ToolResponse>> callables = new
ArrayList<>(executions.size());
+ for (ToolCallExecution execution : executions) {
+ callables.add(execution.callable);
+ }
+ try {
+ List<Outcome<ToolResponse>> outcomes =
ctx.durableExecuteAllAsync(callables);
+ for (int i = 0; i < outcomes.size(); i++) {
+ recordOutcome(executions.get(i), outcomes.get(i), success,
error, responses);
+ }
+ } catch (Exception e) {
Review Comment:
I'll handle all exceptions inside `durableExecuteAllAsync`, so the caller
only needs to process the `outcome` without additional error handling. This
keeps the outer logic cleaner.
--
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]
