weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3671218164
##########
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:
Handling it inside `durableExecuteAllAsync` so the caller only reads
outcomes is a cleaner seam, agreed. Where would you draw the line for failures
that aren't a tool's fault? A state-store write failure mapped into N per-tool
outcomes would let the action complete normally, with the agent carrying on as
though "all tools failed" were a real result. Would it be worth letting those
keep propagating, and mapping only per-call execution failures into outcomes?
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -636,6 +721,110 @@ def wrapped_func(*a: Any, **kw: Any) -> Any:
return wrapped_func
+ def _call_matches(
+ self, current: _PersistedCallResult, call: DurableCall, args_digest:
str
+ ) -> bool:
+ return current.function_id == call.id and current.args_digest ==
args_digest
+
+ def _read_terminal_outcome(self, current: _PersistedCallResult) -> Outcome:
+ if current.exception_payload is not None:
+ return
Outcome.failure(cloudpickle.loads(current.exception_payload))
+ if current.result_payload is None:
+ return Outcome.success(None)
+ return Outcome.success(cloudpickle.loads(current.result_payload))
+
+ def _callable_for_durable_call(self, call: DurableCall) -> Callable[[],
Any]:
+ kwargs = call.kwargs or {}
+ return partial(call.func, *call.args, **kwargs)
+
+ def _prepare_batch_execution(self, calls: list[DurableCall]) ->
_BatchExecutionPlan:
+ args_digest = ""
Review Comment:
Happy either way, but I think two separable things are bundled under
`functionId`, and only one of them is the piece we agreed to defer.
The narrow one is the divergence this PR introduces, which feels in scope
here. `_build_executions` already mints
`DurableCall(id=f"tool-call-{call_id}")` (`tool_call_action.py:134`), but
`_execute_sequentially` passes only `call.func` (`:174`, `:180`), so the id is
dropped and the runtime falls back to `_compute_function_id` →
`module.qualname`, the same string for every tool. Java has no such split —
`buildCallables` mints one callable and both paths use it. Threading the id
that's already built through the serial path looks contained.
The broader one is using a unique per-call id as recovery identity — the
"same args, different result" case we split out to its own issue. Worth
flagging that this PR already crosses that line on Java: `getId()` returns the
constant `"tool-call"` on `main`, and `"tool-call-" + id` here, for the serial
path as well as the batch.
That carries an upgrade cost either way. An action checkpointed
mid-tool-loop on the current release and restored on this build recorded
`functionId="tool-call"`, so `matchNextOrClearSubsequentCallResult`
(`RunnerContextImpl.java:654`) won't match, clears from the current index, and
the already-completed tool calls run again — surfaced in the log as
"Non-deterministic call detected", which would point an operator at the wrong
cause.
So my leaning would be to keep the serial-path id threading here as a parity
fix and let the separate issue carry the semantics change. If you'd rather land
both together, that works too — the piece worth writing down either way is the
cross-version recovery behaviour. What's your read?
##########
runtime/src/test/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImplDurableExecuteAsyncTest.java:
##########
@@ -240,12 +397,61 @@ public <T> T executeAsync(ContinuationContext context,
Supplier<T> supplier) {
return supplier.get();
}
+ @Override
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ java.time.Duration timeout) {
+ executeAllAsyncCallCount++;
+ executeAllAsyncBatchSizes.add(suppliers.size());
+ if (useTimeoutCollection) {
+ return collectTimedOutOutcomes(suppliers);
+ }
+ List<Outcome<T>> outcomes = new
java.util.ArrayList<>(suppliers.size());
+ for (Callable<T> supplier : suppliers) {
+ try {
+ outcomes.add(Outcome.success(supplier.call()));
+ } catch (Exception e) {
+ outcomes.add(Outcome.failure(e));
+ }
+ }
+ return outcomes;
+ }
+
+ private <T> List<Outcome<T>> collectTimedOutOutcomes(List<Callable<T>>
suppliers) {
Review Comment:
Sounds good, and sequencing it after the config fix makes sense. One gap the
e2e won't close though: the stub at `:404` ignores the `timeout` argument it's
handed, so that assertion passes regardless of the deadline. Worth tightening
at the same time?
--
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]
