diff --git a/src/pipecat/adapters/base_llm_adapter.py b/src/pipecat/adapters/base_llm_adapter.py index 3e8037629..7d210d1f5 100644 --- a/src/pipecat/adapters/base_llm_adapter.py +++ b/src/pipecat/adapters/base_llm_adapter.py @@ -15,6 +15,7 @@ from typing import Any, Dict, Generic, List, Optional, TypeVar from loguru import logger +from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.processors.aggregators.llm_context import ( LLMContext, @@ -48,6 +49,20 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]): def __init__(self): """Initialize the adapter.""" self._warned_system_instruction = False + self._builtin_tools: List[FunctionSchema] = [] + + @property + def builtin_tools(self) -> List[FunctionSchema]: + """Built-in tools automatically merged into every inference request. + + Mixins (e.g. ``AsyncToolCancellationLLMServiceMixin``) append their + tool schemas here so that the tools are injected transparently without + the user having to add them to their ``ToolsSchema``. + + Returns: + Mutable list of ``FunctionSchema`` instances. + """ + return self._builtin_tools @property @abstractmethod @@ -122,6 +137,9 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]): def from_standard_tools(self, tools: Any) -> List[Any] | NotGiven: """Convert tools from standard format to provider format. + Built-in tools are automatically merged into the schema before conversion so that every + inference request receives them without the user having to declare them explicitly. + Args: tools: Tools in standard format or provider-specific format. @@ -129,8 +147,26 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]): List of tools converted to provider format, or original tools if not in standard format. """ + if self._builtin_tools: + if isinstance(tools, ToolsSchema): + tools = ToolsSchema( + standard_tools=tools.standard_tools + self._builtin_tools, + custom_tools=tools.custom_tools, + ) + else: + # User supplied tools in a legacy/provider-specific format; + # we cannot safely merge — build a schema from builtins only. + if tools is not None: + logger.warning( + "Built-in tools could not be merged because the supplied tools are not" + " a ToolsSchema instance. Only built-in tools will be sent." + ) + tools = ToolsSchema(standard_tools=self._builtin_tools) + if isinstance(tools, ToolsSchema): logger.debug(f"Retrieving the tools using the adapter: {type(self)}") + tool_names = [tool.name for tool in tools.standard_tools] + logger.debug(f"Tool names: {tool_names}") return self.to_provider_tools_format(tools) # Fallback to return the same tools in case they are not in a standard format return tools diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 03fcb115f..127b9d60e 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -16,6 +16,7 @@ from typing import ( Awaitable, Callable, Dict, + List, Mapping, Optional, Protocol, @@ -60,6 +61,11 @@ from pipecat.services.ai_service import AIService from pipecat.services.settings import LLMSettings from pipecat.services.websocket_service import WebsocketService from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionLLMServiceMixin +from pipecat.utils.async_tool_cancellation import ( + ASYNC_TOOL_CANCELLATION_INSTRUCTIONS, + CANCEL_ASYNC_TOOL_NAME, + CANCEL_ASYNC_TOOL_SCHEMA, +) from pipecat.utils.context.llm_context_summarization import ( DEFAULT_SUMMARIZATION_TIMEOUT, LLMContextSummarizationUtil, @@ -230,6 +236,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): self._group_parallel_tools = group_parallel_tools self._function_call_timeout_secs = function_call_timeout_secs self._filter_incomplete_user_turns: bool = False + self._async_cancellation_enabled: bool = False self._base_system_instruction: Optional[str] = None self._adapter = self.adapter_class() self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} @@ -291,6 +298,8 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): await super().start(frame) if not self._run_in_parallel: await self._create_sequential_runner_task() + if self._has_async_functions(): + self._setup_async_tool_cancellation() async def stop(self, frame: EndFrame): """Stop the LLM service. @@ -315,17 +324,20 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): await self._cancel_summary_task() def _compose_system_instruction(self): - """Compose system_instruction by appending turn completion instructions. + """Compose system_instruction from the base and all active addon instructions. Combines the base system instruction with turn completion instructions - and writes the result to ``self._settings.system_instruction``. + (when enabled) and async tool cancellation instructions (when enabled), + writing the result to ``self._settings.system_instruction``. """ base = self._base_system_instruction - completion_instructions = self._user_turn_completion_config.completion_instructions - if base: - self._settings.system_instruction = f"{base}\n\n{completion_instructions}" - else: - self._settings.system_instruction = completion_instructions + parts = [base] if base else [] + if self._filter_incomplete_user_turns: + parts.append(self._user_turn_completion_config.completion_instructions) + if self._async_cancellation_enabled: + parts.append(ASYNC_TOOL_CANCELLATION_INSTRUCTIONS) + composed = "\n\n".join(p for p in parts if p) + self._settings.system_instruction = composed or None async def _update_settings(self, delta: LLMSettings) -> dict[str, Any]: """Apply a settings delta, handling turn-completion fields. @@ -361,10 +373,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): if ( "system_instruction" in changed - and self._filter_incomplete_user_turns + and (self._filter_incomplete_user_turns or self._async_cancellation_enabled) and "filter_incomplete_user_turns" not in changed ): - # system_instruction changed while turn completion is active. + # system_instruction changed while composition is active. # Treat the new value as the new base and recompose. self._base_system_instruction = self._settings.system_instruction self._compose_system_instruction() @@ -849,6 +861,91 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): if timeout_task and not timeout_task.done(): await self.cancel_task(timeout_task) + def _has_async_functions(self) -> bool: + """Return True if at least one non-builtin async function is registered.""" + return any( + not item.cancel_on_interruption + for name, item in self._functions.items() + if name != CANCEL_ASYNC_TOOL_NAME + ) + + def _setup_async_tool_cancellation(self): + """Enable async tool cancellation. + + Saves the base system instruction, recomposes to include cancellation + instructions, registers the built-in ``cancel_async_tool_call`` handler, + and injects its schema into the adapter's built-in tool list. + """ + logger.debug(f"{self}: Enabling async tool cancellation") + + self._async_cancellation_enabled = True + + if self._base_system_instruction is None: + self._base_system_instruction = self._settings.system_instruction + + self._compose_system_instruction() + + if not any(t.name == CANCEL_ASYNC_TOOL_NAME for t in self._adapter.builtin_tools): + self._adapter.builtin_tools.append(CANCEL_ASYNC_TOOL_SCHEMA) + + if CANCEL_ASYNC_TOOL_NAME not in self._functions: + self._functions[CANCEL_ASYNC_TOOL_NAME] = FunctionCallRegistryItem( + function_name=CANCEL_ASYNC_TOOL_NAME, + handler=self._cancel_async_tool_call_handler, + cancel_on_interruption=True, + ) + + async def _cancel_async_tool_call_handler(self, params: "FunctionCallParams"): + """Handle a ``cancel_async_tool_call`` invocation from the LLM. + + Args: + params: Function call parameters containing ``tool_call_id`` to cancel. + """ + logger.info("_cancel_async_tool_call_handler invoked!") + + tool_call_id: Optional[str] = params.arguments.get("tool_call_id") + if not tool_call_id: + logger.warning(f"{self} cancel_async_tool_call called with no tool_call_id") + await params.result_callback({"cancelled": None}) + return + + await self._cancel_function_calls_by_tool_call_id(tool_call_id) + await params.result_callback( + {"cancelled": tool_call_id}, + properties=FunctionCallResultProperties(run_llm=True), + ) + + async def _cancel_function_calls_by_tool_call_id(self, tool_call_id: str): + """Cancel in-progress function call tasks by their tool_call_id. + + Args: + tool_call_id: tool_call_id to cancel. + """ + cancelled_tasks = set() + for task, runner_item in self._function_call_tasks.items(): + if runner_item.tool_call_id == tool_call_id: + name = runner_item.function_name + tool_call_id = runner_item.tool_call_id + + logger.debug( + f"{self} Cancelling async function call [{name}:{tool_call_id}] " + "by LLM request..." + ) + + if task: + task.remove_done_callback(self._function_call_task_finished) + await self.cancel_task(task) + cancelled_tasks.add(task) + + await self.broadcast_frame( + FunctionCallCancelFrame, function_name=name, tool_call_id=tool_call_id + ) + + logger.debug(f"{self} Async function call [{name}:{tool_call_id}] cancelled") + + for task in cancelled_tasks: + self._function_call_task_finished(task) + async def _cancel_function_call(self, function_name: Optional[str]): cancelled_tasks = set() for task, runner_item in self._function_call_tasks.items(): diff --git a/src/pipecat/utils/async_tool_cancellation.py b/src/pipecat/utils/async_tool_cancellation.py new file mode 100644 index 000000000..e00741508 --- /dev/null +++ b/src/pipecat/utils/async_tool_cancellation.py @@ -0,0 +1,49 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Constants for the built-in async tool cancellation feature. + +When an ``LLMService`` has functions registered with +``cancel_on_interruption=False`` (async tools), it automatically injects the +``cancel_async_tool_call`` tool and the instructions below into every inference +request so the LLM can cancel stale in-progress calls. +""" + +from pipecat.adapters.schemas.function_schema import FunctionSchema + +CANCEL_ASYNC_TOOL_NAME = "cancel_async_tool_call" + +ASYNC_TOOL_CANCELLATION_INSTRUCTIONS = """ASYNC TOOL CANCELLATION: +Some tool calls run asynchronously in the background. When one starts, a tool response \ +is added to the conversation whose content is a JSON object with \ +"type": "tool", "status": "started", and a "tool_call_id" field containing the \ +exact ID of that call (e.g. {"type": "tool", "status": "started", "tool_call_id": "..."}). + +If the user changes topic, explicitly says they no longer need the result, or the pending \ +result would clearly be stale, call cancel_async_tool_call. \ +To find the correct tool_call_id: locate the most recent tool response in the conversation \ +whose content has "status": "started" and whose call has NOT already been cancelled, \ +then copy the "tool_call_id" value from that content exactly as-is. \ +Never invent or guess a tool_call_id.""" + +CANCEL_ASYNC_TOOL_SCHEMA = FunctionSchema( + name=CANCEL_ASYNC_TOOL_NAME, + description=( + "Cancel a single async tool call that is no longer needed. " + "Use this when the user changes topic, indicates a pending result is " + "no longer relevant, or when processing the result would produce a " + "stale or confusing response. " + "The tool_call_id must be the exact 'id' value from the assistant's " + "tool call which we wish to cancel, visible in the conversation history." + ), + properties={ + "tool_call_id": { + "type": "string", + "description": ("The exact id of the async call to cancel."), + } + }, + required=["tool_call_id"], +)