weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3671217142
##########
api/src/main/java/org/apache/flink/agents/api/agents/AgentExecutionOptions.java:
##########
@@ -42,9 +44,43 @@ public class AgentExecutionOptions {
public static final ConfigOption<Boolean> CHAT_ASYNC =
new ConfigOption<>("chat.async", Boolean.class, true);
+ /** Whether the built-in tool-call action runs each tool via durable async
execution. */
public static final ConfigOption<Boolean> TOOL_CALL_ASYNC =
new ConfigOption<>("tool-call.async", Boolean.class, true);
+ /**
+ * Whether multiple tool calls from one {@code ToolRequestEvent} run as
one parallel durable
+ * batch when {@link #TOOL_CALL_ASYNC} is also enabled (JDK 21+).
+ *
+ * <p>Default is {@code true}. A parallel batch raises the number of
in-flight external calls;
+ * after failover, tools whose results were not yet persisted may be
submitted again.
+ * Side-effecting tools should be idempotent or provide a {@code
reconciler()}. Set to {@code
+ * false} to keep serial async or sync tool execution.
+ */
+ public static final ConfigOption<Boolean> TOOL_CALL_PARALLEL =
+ new ConfigOption<>("tool-call.parallel", Boolean.class, true);
+
+ /**
+ * Size of the dedicated thread pool used for tool-call async and parallel
batch execution.
+ *
+ * <p>Separate from {@link #NUM_ASYNC_THREADS} so a large tool batch does
not exhaust the global
+ * async pool.
+ */
+ public static final ConfigOption<Integer> TOOL_CALL_NUM_ASYNC_THREADS =
+ new ConfigOption<>(
+ "tool-call.num-async-threads",
+ Integer.class,
+ Runtime.getRuntime().availableProcessors() * 2);
+
+ /**
+ * Overall timeout for one parallel tool-call batch.
+ *
+ * <p>Non-positive values disable the timeout. When the deadline elapses,
unfinished slots are
+ * failed; slots that already completed keep their success or failure
outcome.
+ */
+ public static final ConfigOption<Duration> TOOL_CALL_BATCH_TIMEOUT =
+ new ConfigOption<>("tool-call.batch.timeout", Duration.class,
Duration.ofMillis(-1));
Review Comment:
The type change buys more than the YAML fix, for what it's worth —
`AgentConfiguration.get()` has a `Long` branch at
`AgentConfiguration.java:142`, and `Long` is Jackson-native, so the
plan-serialization and Python routes resolve along with it.
Two smaller things while the option is in flux.
`short-term-memory.state-ttl.ms` (`AgentExecutionOptions.java:88`) is the
existing `Long`-millis option in this class and puts the unit in the key —
since a bare `Long` drops the unit `Duration` carried, would
`tool-call.batch.timeout.ms` be worth matching to it while the key is still
unreleased?
And a few `Duration` leftovers will probably want sweeping with the same
change: the parity-harness entries added for it
(`check_java_python_config_options_parity.py:45` and the
`_java_duration_to_millis` branch at `:99`) are covered by `"java.lang.Long":
int` at `:40` once the type moves, and the config table still lists the type as
`Duration` with a `-1ms` default (`configuration.md:137`). Any reason to keep
those around once the type moves?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/context/JavaRunnerContextImpl.java:
##########
@@ -75,6 +102,159 @@ private <T> T durableExecuteAsyncWithReconcile(
callable, reconcileCallable, () ->
executeAsyncCallable(callable));
}
+ @Override
+ public <T> List<Outcome<T>>
durableExecuteAllAsync(List<DurableCallable<T>> callables)
+ throws Exception {
+ if (callables.isEmpty()) {
+ return List.of();
+ }
+ if (durableExecutionContext == null) {
+ return executeAllWithoutDurableState(callables);
+ }
+
+ String argsDigest = "";
+ int base = durableExecutionContext.getCurrentCallIndex();
+ BatchExecutionPlan<T> plan = buildBatchExecutionPlan(callables, base,
argsDigest);
+
+ reservePendingBatchIfNeeded(callables, argsDigest, plan);
+
+ List<Outcome<T>> executed = executeOutcomeSuppliers(plan.suppliers);
+ finalizeExecutedOutcomes(callables, base, argsDigest, plan, executed);
+
+ advanceCallIndexBy(callables.size());
+ return plan.outcomes;
+ }
+
+ private <T> BatchExecutionPlan<T> buildBatchExecutionPlan(
+ List<DurableCallable<T>> callables, int base, String argsDigest)
throws Exception {
+ BatchExecutionPlan<T> plan = new
BatchExecutionPlan<>(callables.size());
+ for (int i = 0; i < callables.size(); i++) {
+ DurableCallable<T> callable = callables.get(i);
+ CallResult current = getCallResultAt(base + i);
+ if (current == null) {
+ markReservationStart(plan, i);
+ addExecutableCall(plan, i, callable::call);
+ continue;
+ }
+ if (!current.matches(callable.getId(), argsDigest)) {
+ clearCallResultsFromAndPersist(base + i);
+ plan.needsReservation = true;
+ plan.executionStart = i;
+ addExecutableCall(plan, i, callable::call);
+ appendRemainingExecutions(callables, plan, i + 1);
+ break;
+ }
+ if (current.isPending()) {
+ Callable<T> reconcileCallable = callable.reconciler();
+ Callable<T> executionCallable =
+ reconcileCallable != null ? reconcileCallable :
callable::call;
+ addExecutableCall(plan, i, executionCallable);
+ } else {
+ plan.outcomes.add(
+ readTerminalOutcomeAt(
+ base + i, callable.getId(), argsDigest,
callable.getResultClass()));
+ }
+ }
+ return plan;
+ }
+
+ private <T> void markReservationStart(BatchExecutionPlan<T> plan, int
callIndex) {
+ plan.needsReservation = true;
+ if (plan.executionStart < 0) {
+ plan.executionStart = callIndex;
+ }
+ }
+
+ private <T> void addExecutableCall(
+ BatchExecutionPlan<T> plan, int callIndex, Callable<T>
executionCallable) {
+ plan.outcomes.add(null);
+ plan.suppliers.add(executionCallable);
+ plan.executableCallIndexes.add(callIndex);
+ }
+
+ private <T> void appendRemainingExecutions(
+ List<DurableCallable<T>> callables, BatchExecutionPlan<T> plan,
int startIndex) {
+ for (int i = startIndex; i < callables.size(); i++) {
+ DurableCallable<T> remaining = callables.get(i);
+ addExecutableCall(plan, i, remaining::call);
+ }
+ }
+
+ private <T> void reservePendingBatchIfNeeded(
+ List<DurableCallable<T>> callables, String argsDigest,
BatchExecutionPlan<T> plan) {
+ if (!plan.needsReservation) {
+ return;
+ }
+ List<String> ids = new ArrayList<>();
+ for (DurableCallable<T> callable :
+ callables.subList(plan.executionStart, callables.size())) {
+ ids.add(callable.getId());
+ }
+ reservePendingBatch(ids, argsDigest);
Review Comment:
That plan sounds right. One part I wasn't sure it reaches: the read side is
status-blind independently of how the slot got written. `CallResult.matches`
(`CallResult.java:160`) compares only `functionId` and `argsDigest`, so a
reserved PENDING slot would still read as a hit at the match site even once
`finalizeCallAt` owns the write side. Does the serial path want a status guard
there too, or does your fix remove the route that reaches it with a PENDING
slot in the first place?
--
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]
