diff --git a/docs/wayflowcore/source/core/changelog.rst b/docs/wayflowcore/source/core/changelog.rst index d75c2c85..aa40b121 100644 --- a/docs/wayflowcore/source/core/changelog.rst +++ b/docs/wayflowcore/source/core/changelog.rst @@ -7,6 +7,13 @@ WayFlow |current_version| Improvements ^^^^^^^^^^^^ +* **Parallel tool calling** + + Added the ``parallel_tool_calls`` LLM setting. Independent server-side tool + calls in one response can now execute concurrently, while client-side and + confirmation-requiring tools remain sequential. The documentation includes + guidance for tools with side effects or shared state. + * **Configurable additional object properties** ``ObjectProperty`` now supports JSON Schema ``additionalProperties`` configuration. diff --git a/docs/wayflowcore/source/core/howtoguides/howto_build_assistants_with_tools.rst b/docs/wayflowcore/source/core/howtoguides/howto_build_assistants_with_tools.rst index 73b0ceb5..c1cd0cf7 100644 --- a/docs/wayflowcore/source/core/howtoguides/howto_build_assistants_with_tools.rst +++ b/docs/wayflowcore/source/core/howtoguides/howto_build_assistants_with_tools.rst @@ -266,6 +266,50 @@ The only difference is that the file path is provided as a conversation message :start-after: # .. start-##_Creating_and_running_an_agent_with_a_client_tool :end-before: # .. end-##_Creating_and_running_an_agent_with_a_client_tool + +Parallel tool calling +===================== + +When an LLM returns several tool requests in one response, WayFlow can execute +the requests concurrently. Parallel tool calling is enabled by default and is +configured on the LLM with the ``parallel_tool_calls`` flag: + +.. code-block:: python + + llm = OpenAIModel( + model_id="gpt-4o", + generation_config=LlmGenerationConfig(temperature=0), + parallel_tool_calls=True, + ) + +The flag controls WayFlow's tool-execution behavior; it does not make a model +emit multiple tool requests. The model must support and choose to return a +batch of tool requests. If parallel tool calling is disabled, requests in a +batch are executed sequentially. + +WayFlow executes a batch in parallel only when all of its requests are +server-side tools. Batches containing a ``ClientTool`` are executed +sequentially, because a client tool yields control to the client and may have +side effects that must remain ordered. This also means that tool requests +should not depend on another request in the same batch. If a server-side tool +uses shared mutable state, a shared database connection, or has ordering or +side-effect requirements, document those limitations in the tool description +and disable parallel tool calling when necessary: + +.. code-block:: python + + llm = OpenAIModel( + model_id="gpt-4o", + generation_config=LlmGenerationConfig(temperature=0), + parallel_tool_calls=False, + ) + +Tool descriptions should make dependencies and side effects clear to the LLM. +For example, describe whether a tool is read-only, whether it mutates shared +state, and whether it must be called after another tool. When parallel tool +calling is enabled, only independent tool requests should be emitted in the +same batch. + Agent Spec Exporting/Loading ============================ diff --git a/wayflowcore/src/wayflowcore/executors/_agentexecutor.py b/wayflowcore/src/wayflowcore/executors/_agentexecutor.py index 79fbc9ac..5dd7cc80 100644 --- a/wayflowcore/src/wayflowcore/executors/_agentexecutor.py +++ b/wayflowcore/src/wayflowcore/executors/_agentexecutor.py @@ -4,6 +4,7 @@ # (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License # (UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option. +import asyncio import logging import os import warnings @@ -52,6 +53,7 @@ from wayflowcore.planning import ExecutionPlan from wayflowcore.property import JsonSchemaParam, Property, StringProperty, _validate_strict_outputs from wayflowcore.tools import ClientTool, Tool, ToolRequest, ToolResult +from wayflowcore.tools.servertools import ServerTool from wayflowcore.tools.tools import _descriptors_to_json_schema_map, _sanitize_tool_name from wayflowcore.tracing.span import AgentExecutionSpan @@ -1057,6 +1059,77 @@ async def _process_tool_call( should_yield = False return None, should_yield + @staticmethod + def _get_parallel_server_tool_requests( + config: Agent, state: AgentConversationExecutionState + ) -> List[ToolRequest]: + """Return queued requests that can safely be executed concurrently.""" + if not config.llm.parallel_tool_calls or len(state.tool_call_queue) < 2: + return [] + + tools_by_name = {tool.name: tool for tool in (state.current_retrieved_tools or [])} + queued_requests = list(state.tool_call_queue) + queued_tools = [tools_by_name.get(tool_request.name) for tool_request in queued_requests] + if not all( + isinstance(tool, ServerTool) and not tool.requires_confirmation + for tool in queued_tools + ): + return [] + return queued_requests + + @staticmethod + async def _process_parallel_tool_calls( + agent_config: Agent, + agent_state: AgentConversationExecutionState, + conversation: "AgentConversation", + tool_requests: List[ToolRequest], + ) -> Optional[ExecutionStatus]: + """Execute an all-server-tool batch concurrently.""" + # Remove the batch before starting it so a failure cannot cause a + # request to be executed twice if the conversation is resumed. + agent_state.tool_call_queue.clear() + + results = await asyncio.gather( + *( + AgentConversationExecutor._execute_next_subcall( + config=agent_config, + conversation=conversation, + state=agent_state, + tool_request=tool_request, + messages=conversation.message_list, + ) + for tool_request in tool_requests + ), + return_exceptions=True, + ) + + raised_exceptions = [result for result in results if isinstance(result, Exception)] + if raised_exceptions: + raise raised_exceptions[0] + + yielding_results = [ + result + for result in results + if isinstance(result, ExecutionStatus) and result._requires_yielding + ] + if yielding_results: + # Server tools that are interrupted (for example by an AuthInterrupt) do not + # append a tool result message. Restore those requests so that resuming the + # conversation retries them. Requests which already have a result must stay + # removed from the queue, otherwise they would be executed twice. + completed_tool_request_ids = { + message.tool_result.tool_request_id + for message in conversation.message_list.messages + if message.tool_result is not None + } + agent_state.tool_call_queue.extend( + tool_request + for tool_request in tool_requests + if tool_request.tool_request_id not in completed_tool_request_ids + ) + return yielding_results[0] + return None + @staticmethod async def _execute_agent( conversation: "AgentConversation", @@ -1095,7 +1168,24 @@ async def _execute_agent( agent_state.current_tool_request = None elif len(agent_state.tool_call_queue) > 0: - agent_state._get_current_tool_request() + parallel_tool_requests = ( + AgentConversationExecutor._get_parallel_server_tool_requests( + config=agent_config, state=agent_state + ) + ) + if parallel_tool_requests: + execution_status = ( + await AgentConversationExecutor._process_parallel_tool_calls( + agent_config=agent_config, + agent_state=agent_state, + conversation=conversation, + tool_requests=parallel_tool_requests, + ) + ) + if execution_status is not None: + return execution_status + else: + agent_state._get_current_tool_request() elif agent_state.curr_iter >= agent_config.max_iterations: if len(agent_config.output_descriptors) > 0: default_outputs = _fill_submit_result_defaults( diff --git a/wayflowcore/src/wayflowcore/models/geminimodel.py b/wayflowcore/src/wayflowcore/models/geminimodel.py index c9957ee9..a7a642d1 100644 --- a/wayflowcore/src/wayflowcore/models/geminimodel.py +++ b/wayflowcore/src/wayflowcore/models/geminimodel.py @@ -129,6 +129,7 @@ def __init__( name: Optional[str] = None, description: Optional[str] = None, retry_policy: Optional[RetryPolicy] = None, + parallel_tool_calls: bool = True, ) -> None: self.auth = auth self.proxy = proxy @@ -143,6 +144,7 @@ def __init__( generation_config=generation_config, supports_structured_generation=supports_structured_generation, supports_tool_calling=supports_tool_calling, + parallel_tool_calls=parallel_tool_calls, __metadata_info__=__metadata_info__, id=id, name=name, @@ -289,6 +291,7 @@ def config(self) -> Dict[str, Any]: ), "supports_structured_generation": self.supports_structured_generation, "supports_tool_calling": self.supports_tool_calling, + **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}), "auth": self._serialize_auth_config(self.auth), "generation_config": ( self.generation_config.to_dict() if self.generation_config is not None else None diff --git a/wayflowcore/src/wayflowcore/models/llmmodel.py b/wayflowcore/src/wayflowcore/models/llmmodel.py index 27010c45..b2857c37 100644 --- a/wayflowcore/src/wayflowcore/models/llmmodel.py +++ b/wayflowcore/src/wayflowcore/models/llmmodel.py @@ -115,6 +115,7 @@ def __init__( id: Optional[str] = None, name: Optional[str] = None, description: Optional[str] = None, + parallel_tool_calls: bool = True, ): """ Base class for LLM models. @@ -137,6 +138,10 @@ def __init__( Whether the model supports tool calling or not. When set to `None`, the model will be prompted with a tool and it will check it can use the tool. + parallel_tool_calls: + Whether independent server-side tool calls returned in the same LLM + response should be executed concurrently. Client-side tools and + tools requiring confirmation are always handled sequentially. id: ID of the component. name: @@ -178,6 +183,7 @@ def __init__( if supports_tool_calling is not None else _fetch_tool_calling_support(self) ) + self.parallel_tool_calls = parallel_tool_calls @property def default_chat_template(self) -> "PromptTemplate": diff --git a/wayflowcore/src/wayflowcore/models/ocigenaimodel.py b/wayflowcore/src/wayflowcore/models/ocigenaimodel.py index e3e031e4..2a1acc76 100644 --- a/wayflowcore/src/wayflowcore/models/ocigenaimodel.py +++ b/wayflowcore/src/wayflowcore/models/ocigenaimodel.py @@ -155,6 +155,7 @@ def __init__( auth_profile: Optional[str] = "DEFAULT", api_type: OciAPIType = OciAPIType.OCI, conversation_store_id: Optional[str] = None, + parallel_tool_calls: bool = True, ) -> None: """ Model powered by OCIGenAI. @@ -293,6 +294,7 @@ def __init__( generation_config=generation_config, supports_structured_generation=True, supports_tool_calling=True, + parallel_tool_calls=parallel_tool_calls, __metadata_info__=__metadata_info__, ) @@ -614,6 +616,7 @@ def config(self) -> Dict[str, Any]: "serving_mode": self.serving_mode.value, "compartment_id": self.compartment_id, "provider": self.provider.value, + **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}), } @property diff --git a/wayflowcore/src/wayflowcore/models/ollamamodel.py b/wayflowcore/src/wayflowcore/models/ollamamodel.py index e305ede3..302929d2 100644 --- a/wayflowcore/src/wayflowcore/models/ollamamodel.py +++ b/wayflowcore/src/wayflowcore/models/ollamamodel.py @@ -47,6 +47,7 @@ def __init__( name: Optional[str] = None, description: Optional[str] = None, retry_policy: Optional[RetryPolicy] = None, + parallel_tool_calls: bool = True, ) -> None: """ Model powered by a locally hosted Ollama server. @@ -119,6 +120,7 @@ def __init__( ca_file=ca_file, generation_config=generation_config, supports_tool_calling=supports_tool_calling, + parallel_tool_calls=parallel_tool_calls, supports_structured_generation=supports_structured_generation, __metadata_info__=__metadata_info__, id=id, @@ -155,6 +157,7 @@ def config(self) -> Dict[str, Any]: "generation_config": ( self.generation_config.to_dict() if self.generation_config is not None else None ), + **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}), } @property diff --git a/wayflowcore/src/wayflowcore/models/openaicompatiblemodel.py b/wayflowcore/src/wayflowcore/models/openaicompatiblemodel.py index 9fb6af1e..276d038a 100644 --- a/wayflowcore/src/wayflowcore/models/openaicompatiblemodel.py +++ b/wayflowcore/src/wayflowcore/models/openaicompatiblemodel.py @@ -59,6 +59,7 @@ def __init__( name: Optional[str] = None, description: Optional[str] = None, retry_policy: Optional[RetryPolicy] = None, + parallel_tool_calls: bool = True, ) -> None: """ Model to use remote LLM endpoints that use OpenAI-compatible chat APIs. @@ -135,6 +136,7 @@ def __init__( generation_config=generation_config, supports_structured_generation=supports_structured_generation, supports_tool_calling=supports_tool_calling, + parallel_tool_calls=parallel_tool_calls, __metadata_info__=__metadata_info__, id=id, name=name, @@ -261,6 +263,7 @@ def config(self) -> Dict[str, Any]: ), "supports_structured_generation": self.supports_structured_generation, "supports_tool_calling": self.supports_tool_calling, + **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}), "generation_config": ( self.generation_config.to_dict() if self.generation_config is not None else None ), diff --git a/wayflowcore/src/wayflowcore/models/openaimodel.py b/wayflowcore/src/wayflowcore/models/openaimodel.py index 05227576..7628f83c 100644 --- a/wayflowcore/src/wayflowcore/models/openaimodel.py +++ b/wayflowcore/src/wayflowcore/models/openaimodel.py @@ -29,6 +29,7 @@ def __init__( id: Optional[str] = None, name: Optional[str] = None, description: Optional[str] = None, + parallel_tool_calls: bool = True, ) -> None: """ Model powered by OpenAI. @@ -91,6 +92,7 @@ def __init__( proxy=proxy, api_key=api_key if api_key is not None else os.environ[OPEN_API_KEY], generation_config=generation_config, + parallel_tool_calls=parallel_tool_calls, supports_structured_generation=True, supports_tool_calling=True, api_type=api_type, @@ -113,4 +115,5 @@ def config(self) -> Dict[str, Any]: "generation_config": ( self.generation_config.to_dict() if self.generation_config is not None else None ), + **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}), } diff --git a/wayflowcore/src/wayflowcore/models/vllmmodel.py b/wayflowcore/src/wayflowcore/models/vllmmodel.py index 9bc97ccf..b94a5b2d 100644 --- a/wayflowcore/src/wayflowcore/models/vllmmodel.py +++ b/wayflowcore/src/wayflowcore/models/vllmmodel.py @@ -44,6 +44,7 @@ def __init__( id: Optional[str] = None, name: Optional[str] = None, description: Optional[str] = None, + parallel_tool_calls: bool = True, ) -> None: """ Model powered by a model hosted with VLLM server. @@ -127,6 +128,7 @@ def __init__( generation_config=generation_config, supports_structured_generation=supports_structured_generation, supports_tool_calling=supports_tool_calling, + parallel_tool_calls=parallel_tool_calls, api_type=api_type, retry_policy=retry_policy, __metadata_info__=__metadata_info__, @@ -148,6 +150,7 @@ def config(self) -> Dict[str, Any]: "generation_config": ( self.generation_config.to_dict() if self.generation_config is not None else None ), + **({"parallel_tool_calls": False} if not self.parallel_tool_calls else {}), } @property diff --git a/wayflowcore/tests/integration/test_agent.py b/wayflowcore/tests/integration/test_agent.py index 44c3965d..649a053e 100644 --- a/wayflowcore/tests/integration/test_agent.py +++ b/wayflowcore/tests/integration/test_agent.py @@ -16,16 +16,19 @@ from wayflowcore import Conversation from wayflowcore.agent import Agent, CallerInputMode +from wayflowcore.auth import AuthChallengeRequest from wayflowcore.contextproviders import ContextProvider, ToolContextProvider from wayflowcore.contextproviders.constantcontextprovider import ConstantContextProvider from wayflowcore.controlconnection import ControlFlowEdge from wayflowcore.dataconnection import DataFlowEdge +from wayflowcore.exceptions import AuthInterrupt from wayflowcore.executors._agentexecutor import ( _SUBMIT_TOOL_NAME, _TALK_TO_USER_INPUT_PARAM, _TALK_TO_USER_TOOL_NAME, ) from wayflowcore.executors.executionstatus import ( + AuthChallengeRequestStatus, FinishedStatus, ToolExecutionConfirmationStatus, ToolRequestStatus, @@ -120,6 +123,85 @@ def create_basic_agent(llm: LlmModel, **kwargs) -> Agent: ) +def test_parallel_server_tool_calls_are_executed_concurrently() -> None: + active_calls = 0 + max_active_calls = 0 + + @tool + async def parallel_tool(value: Annotated[int, "value to return"]) -> str: + """Return a value after a short asynchronous operation.""" + nonlocal active_calls, max_active_calls + active_calls += 1 + max_active_calls = max(max_active_calls, active_calls) + await asyncio.sleep(0.01) + active_calls -= 1 + return str(value) + + llm = DummyModel() + agent = Agent( + llm=llm, + tools=[parallel_tool], + custom_instruction="Use the available tools.", + can_finish_conversation=False, + ) + conversation = agent.start_conversation(messages="run both tools") + + with patch_llm( + llm, + outputs=[ + [ + ToolRequest(name="parallel_tool", args={"value": 1}), + ToolRequest(name="parallel_tool", args={"value": 2}), + ], + "done", + ], + patch_internal=True, + ): + conversation.execute() + + assert max_active_calls == 2 + + +def test_parallel_server_tool_calls_can_be_disabled() -> None: + active_calls = 0 + max_active_calls = 0 + + @tool + async def sequential_tool(value: Annotated[int, "value to return"]) -> str: + """Return a value after a short asynchronous operation.""" + nonlocal active_calls, max_active_calls + active_calls += 1 + max_active_calls = max(max_active_calls, active_calls) + await asyncio.sleep(0.01) + active_calls -= 1 + return str(value) + + llm = DummyModel() + llm.parallel_tool_calls = False + agent = Agent( + llm=llm, + tools=[sequential_tool], + custom_instruction="Use the available tools.", + can_finish_conversation=False, + ) + conversation = agent.start_conversation(messages="run both tools") + + with patch_llm( + llm, + outputs=[ + [ + ToolRequest(name="sequential_tool", args={"value": 1}), + ToolRequest(name="sequential_tool", args={"value": 2}), + ], + "done", + ], + patch_internal=True, + ): + conversation.execute() + + assert max_active_calls == 1 + + def _make_tool_request_message( name: str, args: dict, tool_request_id: str = "request_id" ) -> Message: @@ -2506,6 +2588,74 @@ def test_agent_handles_parallel_tool_calls_with_canonicalized_native_template(): _check_each_tool_request_is_followed_by_single_matching_tool_result(conversation.get_messages()) +def test_auth_interrupt_during_parallel_tool_calls_is_resumable(): + llm = DummyModel() + auth_available = False + auth_tool_calls = 0 + successful_tool_calls = 0 + + @tool + def auth_tool() -> str: + """A tool that requires authentication on its first call.""" + nonlocal auth_tool_calls + auth_tool_calls += 1 + if not auth_available: + return AuthInterrupt( + AuthChallengeRequestStatus( + challenge=AuthChallengeRequest( + resource_uri="https://example.com", + issuer=[], + authorization_url="https://example.com/authorize", + ), + client_transport_id="test-transport", + ) + ) # type: ignore[return-value] + return "authenticated" + + @tool + def successful_tool() -> str: + """A tool that completes while the other tool is interrupted.""" + nonlocal successful_tool_calls + successful_tool_calls += 1 + return "successful" + + agent = Agent(llm=llm, tools=[auth_tool, successful_tool], max_iterations=5) + conversation = agent.start_conversation(messages="Call both tools") + + with patch_llm( + llm, + outputs=[ + [ + ToolRequest(name="auth_tool", args={}, tool_request_id="auth-call"), + ToolRequest(name="successful_tool", args={}, tool_request_id="success-call"), + ] + ], + patch_internal=True, + ): + status = conversation.execute() + + assert isinstance(status, AuthChallengeRequestStatus) + assert [request.tool_request_id for request in conversation.state.tool_call_queue] == [ + "auth-call" + ] + assert auth_tool_calls == 1 + assert successful_tool_calls == 1 + + auth_available = True + with patch_llm(llm, outputs=["done"], patch_internal=True): + conversation.execute() + + tool_result_ids = [ + message.tool_result.tool_request_id + for message in conversation.get_messages() + if message.tool_result is not None + ] + assert tool_result_ids.count("auth-call") == 1 + assert tool_result_ids.count("success-call") == 1 + assert auth_tool_calls == 2 + assert successful_tool_calls == 1 + + def test_exception_during_parallel_tool_calls_with_agent(remotely_hosted_llm): sub_agent = Agent( llm=remotely_hosted_llm,