fix: preserve tool call metadata and compatibility in parallel executor

Agent-Logs-Url: https://github.com/MervinPraison/PraisonAI/sessions/67d22169-f2e6-4727-9804-065fd54f9f78

Co-authored-by: MervinPraison <454862+MervinPraison@users.noreply.github.com>

Author: Copilot

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

@@ -16,7 +16,7 @@
 import json
 import xml.etree.ElementTree as ET
 # Gap 2: Tool call execution imports
-from ..tools.call_executor import ToolCall, ToolResult, create_tool_call_executor
+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
@@ -1918,14 +1918,16 @@ def _prepare_return_value(text: str) -> Union[str, tuple]:
 
                             tool_results = []
                             for tool_result_obj in 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 '{tool_result_obj.function_name}' with arguments: {tool_result_obj.arguments if hasattr(tool_result_obj, 'arguments') else 'N/A'}\n"
+                                    display_message = f"Agent {agent_name} called function '{tool_result_obj.function_name}' with arguments: {tool_result_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)
 
@@ -1934,7 +1936,7 @@ def _prepare_return_value(text: str) -> Union[str, tuple]:
                                     'tool_call',
                                     message=f"Calling function: {tool_result_obj.function_name}",
                                     tool_name=tool_result_obj.function_name,
-                                    tool_input=tool_result_obj.arguments if hasattr(tool_result_obj, 'arguments') else {},
+                                    tool_input=tool_result_obj.arguments,
                                     tool_output=result_str[:200] if result_str else None,
                                 )
 
@@ -1949,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,
                                 })
 
@@ -5504,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

src/praisonai-agents/praisonaiagents/tools/call_executor.py

@@ -11,12 +11,11 @@
 - Thread-safe with bounded workers
 """
 
-import asyncio
 import concurrent.futures
 import logging
-from typing import Any, Callable, Dict, List, Optional, Protocol, Union
+from typing import Any, Callable, Dict, List, Optional, Protocol
 from dataclasses import dataclass
-from threading import BoundedSemaphore
+from ..trace.context_events import copy_context_to_callable
 
 logger = logging.getLogger(__name__)
 
@@ -34,6 +33,7 @@ class ToolCall:
 class ToolResult:
     """Result of executing a single tool call."""
     function_name: str
+    arguments: Dict[str, Any]
     result: Any
     tool_call_id: str
     is_ollama: bool
@@ -85,6 +85,7 @@ def execute_batch(
                 )
                 results.append(ToolResult(
                     function_name=tool_call.function_name,
+                    arguments=tool_call.arguments,
                     result=result,
                     tool_call_id=tool_call.tool_call_id,
                     is_ollama=tool_call.is_ollama
@@ -93,6 +94,7 @@ def execute_batch(
                 logger.error(f"Tool execution error for {tool_call.function_name}: {e}")
                 results.append(ToolResult(
                     function_name=tool_call.function_name,
+                    arguments=tool_call.arguments,
                     result=f"Error executing tool: {e}",
                     tool_call_id=tool_call.tool_call_id,
                     is_ollama=tool_call.is_ollama,
@@ -120,7 +122,6 @@ def __init__(self, max_workers: int = 5):
             max_workers: Maximum concurrent tool executions (default 5)
         """
         self.max_workers = max_workers
-        self._semaphore = BoundedSemaphore(max_workers)
 
     def execute_batch(
         self,
@@ -138,34 +139,35 @@ def execute_batch(
 
         def _execute_single_tool(tool_call: ToolCall) -> ToolResult:
             """Execute a single tool call with error handling."""
-            with self._semaphore:  # Respect max_workers bound
-                try:
-                    result = execute_tool_fn(
-                        tool_call.function_name,
-                        tool_call.arguments, 
-                        tool_call.tool_call_id
-                    )
-                    return ToolResult(
-                        function_name=tool_call.function_name,
-                        result=result,
-                        tool_call_id=tool_call.tool_call_id,
-                        is_ollama=tool_call.is_ollama
-                    )
-                except Exception as e:
-                    logger.error(f"Tool execution error for {tool_call.function_name}: {e}")
-                    return ToolResult(
-                        function_name=tool_call.function_name,
-                        result=f"Error executing tool: {e}",
-                        tool_call_id=tool_call.tool_call_id,
-                        is_ollama=tool_call.is_ollama,
-                        error=e
-                    )
+            try:
+                result = execute_tool_fn(
+                    tool_call.function_name,
+                    tool_call.arguments,
+                    tool_call.tool_call_id
+                )
+                return ToolResult(
+                    function_name=tool_call.function_name,
+                    arguments=tool_call.arguments,
+                    result=result,
+                    tool_call_id=tool_call.tool_call_id,
+                    is_ollama=tool_call.is_ollama
+                )
+            except Exception as e:
+                logger.error(f"Tool execution error for {tool_call.function_name}: {e}")
+                return ToolResult(
+                    function_name=tool_call.function_name,
+                    arguments=tool_call.arguments,
+                    result=f"Error executing tool: {e}",
+                    tool_call_id=tool_call.tool_call_id,
+                    is_ollama=tool_call.is_ollama,
+                    error=e
+                )
 
         # Use ThreadPoolExecutor for sync tools
         with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
             # Submit all tool calls
             future_to_index = {
-                executor.submit(_execute_single_tool, tool_call): i 
+                executor.submit(copy_context_to_callable(_execute_single_tool), tool_call): i
                 for i, tool_call in enumerate(tool_calls)
             }
 
@@ -192,4 +194,4 @@ def create_tool_call_executor(parallel: bool = False, max_workers: int = 5) -> T
     if parallel:
         return ParallelToolCallExecutor(max_workers=max_workers)
     else:
-        return SequentialToolCallExecutor()
+        return SequentialToolCallExecutor()

src/praisonai-agents/test_parallel_tools.py

@@ -91,6 +91,7 @@ def mock_execute_tool(name: str, args: dict, tool_call_id: str = None) -> str:
     assert len(seq_results) == len(par_results)
     for i, (seq_result, par_result) in enumerate(zip(seq_results, par_results)):
         assert seq_result.function_name == par_result.function_name
+        assert seq_result.arguments == par_result.arguments
         assert seq_result.tool_call_id == par_result.tool_call_id
         print(f"  Result {i+1}: {seq_result.function_name} -> {seq_result.result}")
 
@@ -194,4 +195,4 @@ def main():
     print("reducing latency for I/O-bound workflows while maintaining backward compatibility.")
 
 if __name__ == "__main__":
-    main()
+    main()