feat: Add ReflectRetryPlugin to reflect from errors and retry when tool/model errors

PiperOrigin-RevId: 783503154

Author: pwwpche

src/google/adk/flows/llm_flows/base_llm_flow.py

@@ -513,45 +513,61 @@ async def _call_llm_async(
     # Calls the LLM.
     llm = self.__get_llm(invocation_context)
     with tracer.start_as_current_span('call_llm'):
-      if invocation_context.run_config.support_cfc:
-        invocation_context.live_request_queue = LiveRequestQueue()
-        async for llm_response in self.run_live(invocation_context):
-          # Runs after_model_callback if it exists.
-          if altered_llm_response := await self._handle_after_model_callback(
-              invocation_context, llm_response, model_response_event
-          ):
-            llm_response = altered_llm_response
-          # only yield partial response in SSE streaming mode
-          if (
-              invocation_context.run_config.streaming_mode == StreamingMode.SSE
-              or not llm_response.partial
-          ):
-            yield llm_response
-          if llm_response.turn_complete:
-            invocation_context.live_request_queue.close()
-      else:
-        # Check if we can make this llm call or not. If the current call pushes
-        # the counter beyond the max set value, then the execution is stopped
-        # right here, and exception is thrown.
-        invocation_context.increment_llm_call_count()
-        async for llm_response in llm.generate_content_async(
-            llm_request,
-            stream=invocation_context.run_config.streaming_mode
-            == StreamingMode.SSE,
+      try:
+        if (
+            invocation_context.run_config
+            and invocation_context.run_config.support_cfc
         ):
-          trace_call_llm(
-              invocation_context,
-              model_response_event.id,
+          invocation_context.live_request_queue = LiveRequestQueue()
+          async for llm_response in self.run_live(invocation_context):
+            # Runs after_model_callback if it exists.
+            if altered_llm_response := await self._handle_after_model_callback(
+                invocation_context, llm_response, model_response_event
+            ):
+              llm_response = altered_llm_response
+            # only yield partial response in SSE streaming mode
+            if (
+                invocation_context.run_config.streaming_mode
+                == StreamingMode.SSE
+                or not llm_response.partial
+            ):
+              yield llm_response
+            if llm_response.turn_complete:
+              invocation_context.live_request_queue.close()
+        else:
+          # Check if we can make this llm call or not. If the current call pushes
+          # the counter beyond the max set value, then the execution is stopped
+          # right here, and exception is thrown.
+          invocation_context.increment_llm_call_count()
+          async for llm_response in llm.generate_content_async(
               llm_request,
-              llm_response,
-          )
-          # Runs after_model_callback if it exists.
-          if altered_llm_response := await self._handle_after_model_callback(
-              invocation_context, llm_response, model_response_event
+              stream=invocation_context.run_config.streaming_mode
+              == StreamingMode.SSE,
           ):
-            llm_response = altered_llm_response
+            trace_call_llm(
+                invocation_context,
+                model_response_event.id,
+                llm_request,
+                llm_response,
+            )
+            # Runs after_model_callback if it exists.
+            if altered_llm_response := await self._handle_after_model_callback(
+                invocation_context, llm_response, model_response_event
+            ):
+              llm_response = altered_llm_response
 
-          yield llm_response
+            yield llm_response
+      except Exception as model_error:
+        if (
+            invocation_context.run_config
+            and invocation_context.run_config.support_cfc
+            and invocation_context.live_request_queue
+        ):
+          invocation_context.live_request_queue.close()
+        error_response = await self._handle_model_error(
+            invocation_context, llm_request, model_response_event, model_error
+        )
+        yield error_response
 
   async def _handle_before_model_callback(
       self,
@@ -592,6 +608,29 @@ async def _handle_before_model_callback(
       if callback_response:
         return callback_response
 
+  async def _handle_model_error(
+      self,
+      invocation_context: InvocationContext,
+      llm_request: LlmRequest,
+      model_response_event: Event,
+      model_error: Exception,
+  ) -> LlmResponse:
+    """Handle model errors through plugin system."""
+    callback_context = CallbackContext(
+        invocation_context, event_actions=model_response_event.actions
+    )
+    error_response = (
+        await invocation_context.plugin_manager.run_on_model_error_callback(
+            callback_context=callback_context,
+            llm_request=llm_request,
+            error=model_error,
+        )
+    )
+    if error_response is not None:
+      return error_response
+    else:
+      raise model_error
+
   async def _handle_after_model_callback(
       self,
       invocation_context: InvocationContext,

src/google/adk/flows/llm_flows/functions.py

@@ -176,9 +176,21 @@ async def handle_function_calls_async(
 
       # Step 3: Otherwise, proceed calling the tool normally.
       if function_response is None:
-        function_response = await __call_tool_async(
-            tool, args=function_args, tool_context=tool_context
-        )
+        try:
+          function_response = await __call_tool_async(
+              tool, args=function_args, tool_context=tool_context
+          )
+        except Exception as tool_error:
+          error_response = await invocation_context.plugin_manager.run_on_tool_error_callback(
+              tool=tool,
+              tool_args=function_args,
+              tool_context=tool_context,
+              error=tool_error,
+          )
+          if error_response is not None:
+            function_response = error_response
+          else:
+            raise tool_error
 
       # Step 4: Check if plugin after_tool_callback overrides the function
       # response.

src/google/adk/plugins/base_plugin.py

@@ -265,6 +265,31 @@ async def after_model_callback(
     """
     pass
 
+  async def on_model_error_callback(
+      self,
+      *,
+      callback_context: CallbackContext,
+      llm_request: LlmRequest,
+      error: Exception,
+  ) -> Optional[LlmResponse]:
+    """Callback executed when a model call encounters an error.
+
+    This callback provides an opportunity to handle model errors gracefully,
+    potentially providing alternative responses or recovery mechanisms.
+
+    Args:
+      callback_context: The context for the current agent call.
+      llm_request: The request that was sent to the model when the error
+        occurred.
+      error: The exception that was raised during model execution.
+
+    Returns:
+      An optional LlmResponse. If an LlmResponse is returned, it will be used
+      instead of propagating the error. Returning `None` allows the original
+      error to be raised.
+    """
+    pass
+
   async def before_tool_callback(
       self,
       *,
@@ -315,3 +340,29 @@ async def after_tool_callback(
       result.
     """
     pass
+
+  async def on_tool_error_callback(
+      self,
+      *,
+      tool: BaseTool,
+      tool_args: dict[str, Any],
+      tool_context: ToolContext,
+      error: Exception,
+  ) -> Optional[dict]:
+    """Callback executed when a tool call encounters an error.
+
+    This callback provides an opportunity to handle tool errors gracefully,
+    potentially providing alternative responses or recovery mechanisms.
+
+    Args:
+      tool: The tool instance that encountered an error.
+      tool_args: The arguments that were passed to the tool.
+      tool_context: The context specific to the tool execution.
+      error: The exception that was raised during tool execution.
+
+    Returns:
+      An optional dictionary. If a dictionary is returned, it will be used as
+      the tool response instead of propagating the error. Returning `None`
+      allows the original error to be raised.
+    """
+    pass

src/google/adk/plugins/plugin_manager.py

@@ -48,6 +48,8 @@
     "after_tool_callback",
     "before_model_callback",
     "after_model_callback",
+    "on_tool_error_callback",
+    "on_model_error_callback",
 ]
 
 logger = logging.getLogger("google_adk." + __name__)
@@ -195,6 +197,21 @@ async def run_after_tool_callback(
         result=result,
     )
 
+  async def run_on_model_error_callback(
+      self,
+      *,
+      callback_context: CallbackContext,
+      llm_request: LlmRequest,
+      error: Exception,
+  ) -> Optional[LlmResponse]:
+    """Runs the `on_model_error_callback` for all plugins."""
+    return await self._run_callbacks(
+        "on_model_error_callback",
+        callback_context=callback_context,
+        llm_request=llm_request,
+        error=error,
+    )
+
   async def run_before_model_callback(
       self, *, callback_context: CallbackContext, llm_request: LlmRequest
   ) -> Optional[LlmResponse]:
@@ -215,6 +232,23 @@ async def run_after_model_callback(
         llm_response=llm_response,
     )
 
+  async def run_on_tool_error_callback(
+      self,
+      *,
+      tool: BaseTool,
+      tool_args: dict[str, Any],
+      tool_context: ToolContext,
+      error: Exception,
+  ) -> Optional[dict]:
+    """Runs the `on_tool_error_callback` for all plugins."""
+    return await self._run_callbacks(
+        "on_tool_error_callback",
+        tool=tool,
+        tool_args=tool_args,
+        tool_context=tool_context,
+        error=error,
+    )
+
   async def _run_callbacks(
       self, callback_name: PluginCallbackName, **kwargs: Any
   ) -> Optional[Any]: