feat(flows): dedupe identical tool calls within an invocation

docs/guides/runners/runner/index.md

@@ -117,6 +117,7 @@ Passed per-invocation to `runner.run_async(..., run_config=RunConfig(...))`:
 | `get_session_config` | `GetSessionConfig \| None` | `None` | Fine-grained session retrieval and event window loading configuration. |
 | `model_input_context` | `list[types.Content] \| None` | `None` | Transient unpersisted context added to model input for the current invocation. |
 | `max_llm_calls` | `int` | `500` | Maximum limit on LLM calls per run execution. |
+| `dedupe_tool_calls` | `bool` | `False` | Opt-in sharing of one execution per invocation between identical tool calls (same agent, branch, tool name and arguments). Callbacks still run per call; a result that transfers, escalates or requests confirmation or auth is never shared; a reused result is marked with `custom_metadata['adk_tool_call_cache_hit']`. `LongRunningFunctionTool` calls are always deduped. |
 
 ## Advanced applications
 

src/google/adk/agents/invocation_context.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import asyncio
+import dataclasses
 from typing import Any
 
 from google.adk.platform import uuid as platform_uuid
@@ -105,6 +106,24 @@ def __init__(
     self.aborted = False
 
 
+@dataclasses.dataclass
+class _ToolCallCacheEntry:
+  """One tool execution shared by the identical calls of an invocation.
+
+  Attributes:
+    future: Resolves to the tool result once the first call for the key has
+      run the tool, or to the exception that run raised.
+    shareable: Whether identical calls may reuse the result. False when the
+      run recorded an action other than a state or artifact delta, such as a
+      transfer, an escalation or a request for authentication or
+      confirmation: that action is the effect of the call that made it, and
+      reusing the result would not replay it.
+  """
+
+  future: asyncio.Future[object]
+  shareable: bool = True
+
+
 class InvocationContext(BaseModel):
   """An invocation context represents the data of a single invocation of an agent.
 
@@ -282,6 +301,18 @@ class InvocationContext(BaseModel):
   _abort_state: _AbortState = PrivateAttr(default_factory=_AbortState)
   """Captured abort state (signal, loop, and aborted flag) shared across copies."""
 
+  _tool_call_cache: dict[tuple[Any, ...], _ToolCallCacheEntry] = PrivateAttr(
+      default_factory=dict
+  )
+  """Tool executions shared by the identical tool calls of this invocation.
+
+  Keyed by agent name, branch, tool name and canonical arguments. Which calls
+  are deduped, and how one execution is shared, is decided by the tool caller
+  in the LLM flow; the context only holds the entries. Like ``_abort_state``, the dict is
+  the very same object in every ``model_copy()`` clone of this context, so an
+  identical call made by a sub-agent finds what its ancestor already ran.
+  """
+
   @override
   def model_post_init(self, __context: Any) -> None:
     super().model_post_init(__context)

src/google/adk/agents/run_config.py

@@ -243,6 +243,32 @@ class RunConfig(BaseModel):
     - Less than or equal to 0: This allows for unbounded number of llm calls.
   """
 
+  dedupe_tool_calls: bool = False
+  """Whether identical tool calls run only once per invocation.
+
+  When enabled, a tool call whose name and arguments match one the same agent
+  already made in this invocation, on the same agent branch, reuses that
+  call's result instead of running the tool again. Models sometimes re-emit a
+  call while a slow or expensive tool is still running, or repeat it in a
+  later step; deduping saves the repeated execution and keeps the two
+  responses consistent. The cache lives for one invocation; a resumed
+  invocation starts with an empty one.
+
+  Only the tool execution is shared. Before-tool and after-tool callbacks still
+  run for every call, every call gets its own function response event, and a
+  duplicate receives its own copy of the result as the tool returned it. State
+  and artifact deltas come from the first execution only. A result is not
+  shared when its run recorded any other action (a transfer, an escalation, a
+  request for authentication or confirmation...), since that action is the
+  effect of the call that made it; a failed execution is not reused either. A
+  reused result is marked with ``custom_metadata['adk_tool_call_cache_hit'] =
+  True`` on its function response event (on the merged event when the
+  responses of one step are merged).
+
+  ``LongRunningFunctionTool`` calls are always deduped, whatever this setting
+  is.
+  """
+
   custom_metadata: Optional[dict[str, Any]] = None
   """Custom metadata for the current invocation."""
 

src/google/adk/flows/llm_flows/tools/_batch_executor.py

@@ -101,6 +101,14 @@ def merge_parallel_function_response_events(
 
   merged_actions = EventActions.model_validate(merged_actions_data)
 
+  # Metadata is merged too, so what a call's own event says about its response
+  # (such as that the result was reused from an identical call) is still said
+  # once the responses of a step are combined.
+  merged_metadata: dict[str, Any] = {}
+  for event in function_response_events:
+    if event.custom_metadata:
+      merged_metadata.update(event.custom_metadata)
+
   # Create the new merged event
   merged_event = Event(
       invocation_id=base_event.invocation_id,
@@ -109,6 +117,7 @@ def merge_parallel_function_response_events(
       content=types.Content(role='user', parts=merged_parts),
       actions=merged_actions,
       live_session_id=base_event.live_session_id,
+      custom_metadata=merged_metadata or None,
   )
 
   # Use the base_event as the timestamp