Merge pull request #1401 from MervinPraison/claude/issue-1392-20260416-0916

feat: implement parallel tool execution (Gap 2) with backward compatibility

Author: MervinPraison

src/praisonai-agents/praisonaiagents/agent/agent.py

@@ -768,6 +768,14 @@ def __init__(
                 alternative="use 'execution=ExecutionConfig(rate_limiter=obj)' instead",
                 stacklevel=3
             )
+        if parallel_tool_calls is not None:
+            warn_deprecated_param(
+                "parallel_tool_calls",
+                since="1.0.0",
+                removal="2.0.0",
+                alternative="use 'execution=ExecutionConfig(parallel_tool_calls=True)' instead",
+                stacklevel=3
+            )
         if verification_hooks is not None:
             warn_deprecated_param(
                 "verification_hooks",
@@ -943,13 +951,17 @@ def __init__(
                 allow_code_execution = True
             if _exec_config.code_mode != "safe":
                 code_execution_mode = _exec_config.code_mode
+            # Get parallel_tool_calls from ExecutionConfig
+            parallel_tool_calls = _exec_config.parallel_tool_calls
             # Budget guard extraction
             _max_budget = getattr(_exec_config, 'max_budget', None)
             _on_budget_exceeded = getattr(_exec_config, 'on_budget_exceeded', 'stop') or 'stop'
         else:
             max_iter, max_rpm, max_execution_time, max_retry_limit = 20, None, None, 2
             _max_budget = None
             _on_budget_exceeded = 'stop'
+            # Default parallel_tool_calls when no ExecutionConfig provided
+            parallel_tool_calls = False
 
         # ─────────────────────────────────────────────────────────────────────
         # Resolve TEMPLATES param - FAST PATH
@@ -1440,6 +1452,8 @@ def __init__(
             self.self_reflect = True if self_reflect is None else self_reflect
 
         self.instructions = instructions
+        # Gap 2: Store parallel tool calls setting for ToolCallExecutor selection
+        self.parallel_tool_calls = parallel_tool_calls
         # Check for model name in environment variable if not provided
         self._using_custom_llm = False
         # Flag to track if final result has been displayed to prevent duplicates

src/praisonai-agents/praisonaiagents/agent/chat_mixin.py

@@ -1250,6 +1250,7 @@ def _chat_impl(self, prompt, temperature, tools, output_json, output_pydantic, r
                         task_description=task_description,
                         task_id=task_id,
                         execute_tool_fn=self.execute_tool,
+                        parallel_tool_calls=getattr(self.execution, "parallel_tool_calls", False),
                         reasoning_steps=reasoning_steps,
                         stream=stream
                     )
@@ -1719,6 +1720,7 @@ async def _achat_impl(self, prompt, temperature, tools, output_json, output_pyda
                         task_description=task_description,
                         task_id=task_id,
                         execute_tool_fn=self.execute_tool_async,
+                        parallel_tool_calls=getattr(self.execution, "parallel_tool_calls", False),
                         reasoning_steps=reasoning_steps,
                         stream=stream
                     )
@@ -2248,7 +2250,8 @@ def _start_stream(self, prompt: str, **kwargs) -> Generator[str, None, None]:
                         task_name=kwargs.get('task_name'),
                         task_description=kwargs.get('task_description'),
                         task_id=kwargs.get('task_id'),
-                        execute_tool_fn=self.execute_tool
+                        execute_tool_fn=self.execute_tool,
+                        parallel_tool_calls=getattr(self.execution, "parallel_tool_calls", False)
                     ):
                         response_content += chunk
                         yield chunk

src/praisonai-agents/praisonaiagents/config/feature_configs.py

@@ -735,6 +735,11 @@ class ExecutionConfig:
     # Action when budget exceeded: "stop" (default) raises BudgetExceededError,
     # "warn" logs warning but continues, or callable(total_cost, max_budget).
     on_budget_exceeded: Any = "stop"
+    
+    # Parallel tool execution (Gap 2): Enable parallel execution of batched LLM tool calls
+    # When True, multiple tool calls from LLM are executed concurrently instead of sequentially
+    # Default False preserves existing behavior for backward compatibility
+    parallel_tool_calls: bool = False
 
     def to_dict(self) -> Dict[str, Any]:
         """Convert to dictionary."""
@@ -749,6 +754,7 @@ def to_dict(self) -> Dict[str, Any]:
             "context_compaction": self.context_compaction,
             "max_context_tokens": self.max_context_tokens,
             "max_budget": self.max_budget,
+            "parallel_tool_calls": self.parallel_tool_calls,
         }
 
 

src/praisonai-agents/praisonaiagents/llm/llm.py

@@ -15,6 +15,8 @@
 import time
 import json
 import xml.etree.ElementTree as ET
+# Gap 2: Tool call execution imports
+from ..tools.call_executor import ToolCall, create_tool_call_executor
 # Display functions - lazy loaded to avoid importing rich at startup
 # These are only needed when output=verbose
 _display_module = None
@@ -1649,6 +1651,7 @@ def get_response(
         task_description: Optional[str] = None,
         task_id: Optional[str] = None,
         execute_tool_fn: Optional[Callable] = None,
+        parallel_tool_calls: bool = False,  # Gap 2: Enable parallel tool execution
         stream: bool = True,
         stream_callback: Optional[Callable] = None,
         emit_events: bool = False,
@@ -1893,26 +1896,47 @@ def _prepare_return_value(text: str) -> Union[str, tuple]:
                                 "tool_calls": serializable_tool_calls,
                             })
 
-                            tool_results = []
+                            # Execute tool calls using ToolCallExecutor (Gap 2: parallel or sequential)
+                            is_ollama = self._is_ollama_provider()
+                            tool_calls_batch = []
+                            
+                            # Prepare batch of ToolCall objects
                             for tool_call in tool_calls:
-                                function_name, arguments, tool_call_id = self._extract_tool_call_info(tool_call)
-
-                                logging.debug(f"[RESPONSES_API] Executing tool {function_name} with args: {arguments}")
-                                tool_result = execute_tool_fn(function_name, arguments, tool_call_id=tool_call_id)
+                                function_name, arguments, tool_call_id = self._extract_tool_call_info(tool_call, is_ollama=is_ollama)
+                                tool_calls_batch.append(ToolCall(
+                                    function_name=function_name,
+                                    arguments=arguments,
+                                    tool_call_id=tool_call_id,
+                                    is_ollama=is_ollama
+                                ))
+                            
+                            # Create appropriate executor based on parallel_tool_calls setting
+                            executor = create_tool_call_executor(parallel=parallel_tool_calls)
+                            
+                            # Execute batch
+                            tool_results_batch = executor.execute_batch(tool_calls_batch, execute_tool_fn)
+                            
+                            tool_results = []
+                            for tool_call_obj, tool_result_obj in zip(tool_calls_batch, tool_results_batch):
+                                if tool_result_obj.error is not None:
+                                    raise tool_result_obj.error
+                                tool_result = tool_result_obj.result
                                 tool_results.append(tool_result)
                                 accumulated_tool_results.append(tool_result)
 
+                                logging.debug(f"[RESPONSES_API] Executed tool {tool_result_obj.function_name} with result: {tool_result}")
+
                                 if verbose:
-                                    display_message = f"Agent {agent_name} called function '{function_name}' with arguments: {arguments}\n"
+                                    display_message = f"Agent {agent_name} called function '{tool_call_obj.function_name}' with arguments: {tool_call_obj.arguments}\n"
                                     display_message += f"Function returned: {tool_result}" if tool_result else "Function returned no output"
                                     _get_display_functions()['display_tool_call'](display_message, console=self.console)
 
                                 result_str = json.dumps(tool_result) if tool_result else "empty"
                                 _get_display_functions()['execute_sync_callback'](
                                     'tool_call',
-                                    message=f"Calling function: {function_name}",
-                                    tool_name=function_name,
-                                    tool_input=arguments,
+                                    message=f"Calling function: {tool_call_obj.function_name}",
+                                    tool_name=tool_call_obj.function_name,
+                                    tool_input=tool_call_obj.arguments,
                                     tool_output=result_str[:200] if result_str else None,
                                 )
 
@@ -1927,7 +1951,7 @@ def _prepare_return_value(text: str) -> Union[str, tuple]:
                                     content = json.dumps(tool_result)
                                 messages.append({
                                     "role": "tool",
-                                    "tool_call_id": tool_call_id,
+                                    "tool_call_id": tool_result_obj.tool_call_id,
                                     "content": content,
                                 })
 
@@ -3142,6 +3166,7 @@ def get_response_stream(
         task_description: Optional[str] = None,
         task_id: Optional[str] = None,
         execute_tool_fn: Optional[Callable] = None,
+        parallel_tool_calls: bool = False,  # Gap 2: Enable parallel tool execution
         **kwargs
     ):
         """Generator that yields real-time response chunks from the LLM.
@@ -3167,6 +3192,7 @@ def get_response_stream(
             task_description: Optional task description for logging
             task_id: Optional task ID for logging
             execute_tool_fn: Optional function for executing tools
+            parallel_tool_calls: If True, execute batched LLM tool calls in parallel (default False)
             **kwargs: Additional parameters
             
         Yields:
@@ -3301,26 +3327,44 @@ def get_response_stream(
                                 "tool_calls": serializable_tool_calls
                             })
 
-                        # Execute tool calls and add results to conversation
+                        # Execute tool calls using ToolCallExecutor (Gap 2: parallel or sequential)
+                        is_ollama = self._is_ollama_provider()
+                        tool_calls_batch = []
+                        
+                        # Prepare batch of ToolCall objects
                         for tool_call in tool_calls:
-                            is_ollama = self._is_ollama_provider()
                             function_name, arguments, tool_call_id = self._extract_tool_call_info(tool_call, is_ollama)
-                            
-                            try:
-                                # Execute the tool (pass tool_call_id for event correlation)
-                                tool_result = execute_tool_fn(function_name, arguments, tool_call_id=tool_call_id)
-                                
-                                # Add tool result to messages
-                                tool_message = self._create_tool_message(function_name, tool_result, tool_call_id, is_ollama)
-                                messages.append(tool_message)
-                                
-                            except Exception as e:
-                                logging.error(f"Tool execution error for {function_name}: {e}")
-                                # Add error message to conversation
-                                error_message = self._create_tool_message(
-                                    function_name, f"Error executing tool: {e}", tool_call_id, is_ollama
+                            tool_calls_batch.append(ToolCall(
+                                function_name=function_name,
+                                arguments=arguments, 
+                                tool_call_id=tool_call_id,
+                                is_ollama=is_ollama
+                            ))
+                        
+                        # Create appropriate executor based on parallel_tool_calls setting
+                        executor = create_tool_call_executor(parallel=parallel_tool_calls)
+                        
+                        # Execute batch and add results to conversation
+                        tool_results = executor.execute_batch(tool_calls_batch, execute_tool_fn)
+                        
+                        for tool_result in tool_results:
+                            if tool_result.error is None:
+                                # Successful execution
+                                tool_message = self._create_tool_message(
+                                    tool_result.function_name, 
+                                    tool_result.result, 
+                                    tool_result.tool_call_id, 
+                                    tool_result.is_ollama
+                                )
+                            else:
+                                # Error during execution (already logged by executor)
+                                tool_message = self._create_tool_message(
+                                    tool_result.function_name,
+                                    tool_result.result,  # Contains error message
+                                    tool_result.tool_call_id,
+                                    tool_result.is_ollama
                                 )
-                                messages.append(error_message)
+                            messages.append(tool_message)
 
                         # Continue conversation after tool execution - get follow-up response
                         try:
@@ -5462,4 +5506,4 @@ def _generate_tool_definition(self, function_or_name) -> Optional[Dict]:
             }
         }
         logging.debug(f"Generated tool definition: {tool_def}")
-        return tool_def
+        return tool_def