weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3697315162
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -412,8 +485,9 @@ def _try_get_cached_result(
A tuple of (is_hit, result_or_exception). If is_hit is True,
the second element is the cached result or an exception to
re-raise.
"""
- function_id = _compute_function_id(func)
- args_digest = _compute_args_digest(args, kwargs)
+ function_id, args_digest = _resolve_durable_identity(
Review Comment:
The new PENDING branch seems to open a recovery gap on Python, and I would
welcome a sanity check on the routing. `matchNextOrClearSubsequentCallResult`
returns `null` for a matching PENDING slot without clearing it or advancing
`currentCallIndex` (`RunnerContextImpl.java:762-771`). Java pre-checks PENDING
in `durableExecuteCompletionOnly` at `:297-300` and finalizes in place, so it
never lands there. Python's `_try_get_cached_result` reads that `null` as a
plain miss (`:494`), re-executes, and `recordCallCompletion` appends
(`RunnerContextImpl.java:813`): a two-call batch replayed serially becomes
`[PENDING(f0), PENDING(f1), SUCCEEDED(f0), SUCCEEDED(f1)]`, and each later
failover re-runs both tools and appends two more.
It is new at this head: at `9169bb2` the batch reserved
`"tool-call-<call_id>"`, which never matched the serial digest, so the mismatch
branch cleared and re-ran cleanly. No config flip is needed either, since
`len(executions)` dropping to 1 on an unresolvable tool routes a reserved batch
to the serial path (`tool_call_action.py:69`, `:108-117`); flipping back to
parallel caps the growth but leaves a stale tail. Java has the recovery test
(`RunnerContextImplDurableExecuteTest.java:267-299`); would a Python
counterpart make sense, given the parity ask in `AGENTS.md`? For the fix, the
bridge could take Java's pre-check, or the matcher could clear the slot before
returning the miss. Which fits the bridge contract better?
##########
docs/content/docs/operations/configuration.md:
##########
@@ -131,7 +131,10 @@ Here is the list of all built-in core configuration
options.
| `max-retries` | 3 | int
| Number of retries when using `ErrorHandlingStrategy.RETRY`.
|
| `retry-wait-interval` | 1 | int
| Base wait interval in seconds between retries when using
`ErrorHandlingStrategy.RETRY`. Uses exponential backoff: the actual wait time
for the Nth retry is `retry-wait-interval * 2^(N-1)` seconds. For example, with
default 1s, waits are 1s, 2s, 4s, etc. Retry count and total wait time are
reported in `ChatResponseEvent` and recorded as metrics (`retryCount`,
`retryWaitSec`) under the connection name. |
| `chat.async` | true | boolean
| Whether chat asynchronously for built-in chat action.
|
-| `tool-call.async` | true | boolean
| Whether process tool call for built-in tool call action.
|
+| `tool-call.async` | true | boolean
| Whether the built-in tool-call action runs each tool via durable async
execution.
|
+| `tool-call.parallel` | true | boolean
| When `tool-call.async` is also true (JDK 21+), run multiple tool calls from
one `ToolRequestEvent` as one parallel durable batch. Increases in-flight
external calls; after failover, unfinished tools may be submitted again —
side-effecting tools should be idempotent or provide a reconciler. Set to
`false` for serial tool execution. |
+| `tool-call.num-async-threads` | os cpu count * 2 | int
| Dedicated thread pool size for tool-call async / parallel batch execution.
Separate from `num-async-threads` so a large tool batch does not exhaust the
global async pool.
|
+| `tool-call.batch.timeout.ms` | -1 (disabled) | long
(milliseconds) | Overall timeout for one parallel tool-call batch.
Non-positive disables it. On timeout, completed slots keep their outcome and
unfinished slots fail. Timeout cancellation is best-effort; external side
effects from unfinished tool calls may still complete, so side-effecting tools
should be idempotent or provide a reconciler. |
Review Comment:
nit: two JDK wording points. The new row has no JDK caveat, though the
option is inert below Java 21: the java11 fallback takes `timeout` and ignores
it (`ContinuationActionExecutor.java:75-87`), and two of the five IT combos run
Java 17. Conversely `:135` says `tool-call.parallel` needs JDK 21+, which does
not hold for Python, whose batch runs on a `ThreadPoolExecutor`
(`flink_runner_context.py:266`). Worth a clause on each?
##########
python/flink_agents/api/runner_context.py:
##########
@@ -24,12 +25,49 @@
from flink_agents.api.metric_group import MetricGroup
from flink_agents.api.resource import Resource, ResourceType
-__all__ = ["AsyncExecutionResult", "RunnerContext"]
+__all__ = ["AsyncExecutionResult", "DurableCall", "Outcome", "RunnerContext"]
if TYPE_CHECKING:
from flink_agents.api.memory_object import MemoryObject
+@dataclass(frozen=True)
+class DurableCall:
+ """A deterministic durable call entry for batch execution."""
+
+ id: str
Review Comment:
`id` is required here but the Python runtime never reads it:
`_durable_identity` (`flink_runner_context.py:751-752`) recomputes identity
from `call.func/args/kwargs` at `:755`, `:821`, `:843`. Java treats the id as
the journal identity (`RunnerContextImpl.java:293`), so a stable `id` with
varying arguments keeps recovery on Java and loses it on Python. `DurableCall`
is new, so it is cheap to settle now. Should `id` stay on the dataclass at all?
Keeping it also pulls in the `plan` to `runtime` import
(`tool_call_action.py:34`), the first production one in the repo, against the
direction `AGENTS.md` states. Java declares the id locally
(`ToolCallAction.java:45`). Where would you want that to live?
##########
python/flink_agents/runtime/tests/test_flink_runner_context_reconcilable.py:
##########
@@ -128,27 +170,78 @@ def clearCallResultsFromCurrentIndexAndPersist(self) ->
None:
self.operations.append("clear")
self.call_results = self.call_results[: self.current_call_index]
+ def reservePendingBatch(
+ self, function_ids: list[str], args_digests: list[str]
+ ) -> None:
+ self.operations.append(f"reserve:{len(function_ids)}")
+ for function_id, digest in zip(function_ids, args_digests,
strict=True):
+ self.call_results.append(
+ _StoredCallResult(
+ function_id=function_id,
+ args_digest=digest,
+ status="PENDING",
Review Comment:
The fake's `matchNextOrClearSubsequentCallResult` (`:103-116`) has no
`status == "PENDING"` branch, so a PENDING match still returns `[True, None,
None]`, the pre-fix behaviour changed at `RunnerContextImpl.java:762-771`. A
Python regression test for the batch to serial case would pass here while
production appends. Would it help to teach the fake the new contract first, so
such a test could fail?
##########
python/flink_agents/api/runner_context.py:
##########
@@ -203,6 +241,7 @@ def durable_execute(
func: Callable[[Any], Any],
*args: Any,
reconciler: Callable[[], Any] | None = None,
+ durable_id: str | None = None,
Review Comment:
`durable_id` is on both public abstract signatures (`:244`, `:306`) and
threaded through `flink_runner_context.py`, but nothing passes it, and
`test_tool_call_action.py:361` asserts it is not passed. Java has no
counterpart. Two things that made me pause while it sits unused: `reconciler`'s
docstring marks itself reserved and not forwarded to `func` (`:285-286`) where
`durable_id`'s (`:287-290`) does not, so a tool declaring its own `durable_id`
would have it swallowed, and every third-party `RunnerContext` inherits the
abstract signature. Is a follow-up expected to use it?
##########
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:
Partly. The Python catch-all is gone (`tool_call_action.py:145-156`), but
the Java one at `ToolCallAction.java:168-177` is unchanged, so the two
languages now do different things with the same failure. Was leaving Java for
the follow-up the intent? I answered your two-layer question in the review
comment, since it bears on this.
--
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]
