Merge branch 'main' into main

Author: ryleymao

src/google/adk/agents/run_config.py

@@ -31,6 +31,24 @@
 logger = logging.getLogger('google_adk.' + __name__)
 
 
+class ToolThreadPoolConfig(BaseModel):
+  """Configuration for the tool thread pool executor.
+
+  Attributes:
+    max_workers: Maximum number of worker threads in the pool. Defaults to 4.
+  """
+
+  model_config = ConfigDict(
+      extra='forbid',
+  )
+
+  max_workers: int = Field(
+      default=4,
+      description='Maximum number of worker threads in the pool.',
+      ge=1,
+  )
+
+
 class StreamingMode(Enum):
   """Streaming modes for agent execution.
 
@@ -232,6 +250,53 @@ class RunConfig(BaseModel):
   save_live_blob: bool = False
   """Saves live video and audio data to session and artifact service."""
 
+  tool_thread_pool_config: Optional[ToolThreadPoolConfig] = None
+  """Configuration for running tools in a thread pool for live mode.
+
+  When set, tool executions will run in a separate thread pool executor
+  instead of the main event loop. When None (default), tools run in the
+  main event loop.
+
+  This helps keep the event loop responsive for:
+  - User interruptions to be processed immediately
+  - Model responses to continue being received
+
+  Both sync and async tools are supported. Async tools are run in a new event
+  loop within the background thread, which helps catch blocking I/O mistakenly
+  used inside async functions.
+
+  IMPORTANT - GIL (Global Interpreter Lock) Considerations:
+
+  Thread pool HELPS with (GIL is released):
+  - Blocking I/O: time.sleep(), network calls, file I/O, database queries
+  - C extensions: numpy, hashlib, image processing libraries
+  - Async functions containing blocking I/O (common user mistake)
+
+  Thread pool does NOT help with (GIL is held):
+  - Pure Python CPU-bound code: loops, calculations, recursive algorithms
+  - The GIL prevents true parallel execution for Python bytecode
+
+  For CPU-intensive Python code, consider alternatives:
+  - Use C extensions that release the GIL
+  - Break work into chunks with periodic `await asyncio.sleep(0)`
+  - Use multiprocessing (ProcessPoolExecutor) for true parallelism
+
+  Example:
+    ```python
+    from google.adk.agents.run_config import RunConfig, ToolThreadPoolConfig
+
+    # Enable thread pool with default settings
+    run_config = RunConfig(
+        tool_thread_pool_config=ToolThreadPoolConfig(),
+    )
+
+    # Enable thread pool with custom max_workers
+    run_config = RunConfig(
+        tool_thread_pool_config=ToolThreadPoolConfig(max_workers=8),
+    )
+    ```
+  """
+
   save_live_audio: bool = Field(
       default=False,
       deprecated=True,

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

@@ -17,7 +17,9 @@
 from __future__ import annotations
 
 import asyncio
+from concurrent.futures import ThreadPoolExecutor
 import copy
+import functools
 import inspect
 import logging
 import threading
@@ -53,6 +55,109 @@
 
 logger = logging.getLogger('google_adk.' + __name__)
 
+# Global thread pool executors for running tools in background threads.
+# This prevents blocking tools from blocking the event loop in Live API mode.
+# Key is max_workers, value is the executor.
+_TOOL_THREAD_POOLS: dict[int, ThreadPoolExecutor] = {}
+_TOOL_THREAD_POOL_LOCK = threading.Lock()
+
+
+def _get_tool_thread_pool(max_workers: int = 4) -> ThreadPoolExecutor:
+  """Gets or creates a thread pool executor for tool execution.
+
+  Args:
+    max_workers: Maximum number of worker threads in the pool.
+
+  Returns:
+    A ThreadPoolExecutor with the specified max_workers.
+  """
+  if max_workers not in _TOOL_THREAD_POOLS:
+    with _TOOL_THREAD_POOL_LOCK:
+      if max_workers not in _TOOL_THREAD_POOLS:
+        _TOOL_THREAD_POOLS[max_workers] = ThreadPoolExecutor(
+            max_workers=max_workers, thread_name_prefix='adk_tool_executor'
+        )
+  return _TOOL_THREAD_POOLS[max_workers]
+
+
+def _is_sync_tool(tool: BaseTool) -> bool:
+  """Checks if a tool's underlying function is synchronous."""
+  if not hasattr(tool, 'func'):
+    return False
+  func = tool.func
+  return not (
+      inspect.iscoroutinefunction(func)
+      or inspect.isasyncgenfunction(func)
+      or (
+          hasattr(func, '__call__')
+          and inspect.iscoroutinefunction(func.__call__)
+      )
+  )
+
+
+async def _call_tool_in_thread_pool(
+    tool: BaseTool,
+    args: dict[str, Any],
+    tool_context: ToolContext,
+    max_workers: int = 4,
+) -> Any:
+  """Runs a tool in a thread pool to avoid blocking the event loop.
+
+  For sync tools, this runs the tool's function directly in a background thread.
+  For async tools, this creates a new event loop in the background thread and
+  runs the async function there. This helps catch blocking I/O (like time.sleep,
+  network calls, file I/O) that was mistakenly used inside async functions.
+
+  Note: Due to Python's GIL, this does NOT help with pure Python CPU-bound code.
+  Thread pool only helps when the GIL is released (blocking I/O, C extensions).
+
+  Args:
+    tool: The tool to execute.
+    args: Arguments to pass to the tool.
+    tool_context: The tool context.
+    max_workers: Maximum number of worker threads in the pool.
+
+  Returns:
+    The result of running the tool.
+  """
+  from ...tools.function_tool import FunctionTool
+
+  loop = asyncio.get_running_loop()
+  executor = _get_tool_thread_pool(max_workers)
+
+  if _is_sync_tool(tool):
+    # For sync FunctionTool, call the underlying function directly
+    def run_sync_tool():
+      if isinstance(tool, FunctionTool):
+        args_to_call = tool._preprocess_args(args)
+        signature = inspect.signature(tool.func)
+        valid_params = {param for param in signature.parameters}
+        if 'tool_context' in valid_params:
+          args_to_call['tool_context'] = tool_context
+        args_to_call = {
+            k: v for k, v in args_to_call.items() if k in valid_params
+        }
+        return tool.func(**args_to_call)
+      else:
+        # For other sync tool types, we can't easily run them in thread pool
+        return None
+
+    result = await loop.run_in_executor(executor, run_sync_tool)
+    if result is not None:
+      return result
+  else:
+    # For async tools, run them in a new event loop in a background thread.
+    # This helps when async functions contain blocking I/O (common user mistake)
+    # that would otherwise block the main event loop.
+    def run_async_tool_in_new_loop():
+      # Create a new event loop for this thread
+      return asyncio.run(tool.run_async(args=args, tool_context=tool_context))
+
+    return await loop.run_in_executor(executor, run_async_tool_in_new_loop)
+
+  # Fall back to normal async execution for non-FunctionTool sync tools
+  return await tool.run_async(args=args, tool_context=tool_context)
+
 
 def generate_client_function_call_id() -> str:
   return f'{AF_FUNCTION_CALL_ID_PREFIX}{uuid.uuid4()}'
@@ -706,9 +811,19 @@ async def run_tool_and_update_queue(tool, function_args, tool_context):
         )
     }
   else:
-    function_response = await __call_tool_async(
-        tool, args=function_args, tool_context=tool_context
-    )
+    # Check if we should run tools in thread pool to avoid blocking event loop
+    thread_pool_config = invocation_context.run_config.tool_thread_pool_config
+    if thread_pool_config is not None:
+      function_response = await _call_tool_in_thread_pool(
+          tool,
+          args=function_args,
+          tool_context=tool_context,
+          max_workers=thread_pool_config.max_workers,
+      )
+    else:
+      function_response = await __call_tool_async(
+          tool, args=function_args, tool_context=tool_context
+      )
   return function_response