Merge pull request #4241 from pipecat-ai/filipi/async_tools_cancellable

Enable async tool cancellation feature.
This commit is contained in:
Filipi da Silva Fuchter
2026-04-10 15:28:01 -03:00
committed by GitHub
11 changed files with 328 additions and 18 deletions

View File

@@ -118,6 +118,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = AnthropicLLMService( llm = AnthropicLLMService(
api_key=os.getenv("ANTHROPIC_API_KEY"), api_key=os.getenv("ANTHROPIC_API_KEY"),
enable_async_tool_cancellation=True,
settings=AnthropicLLMService.Settings( settings=AnthropicLLMService.Settings(
system_instruction=( system_instruction=(
"You are a helpful assistant in a voice conversation. " "You are a helpful assistant in a voice conversation. "
@@ -139,9 +140,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
timeout_secs=30, timeout_secs=30,
) )
@llm.event_handler("on_function_calls_started") @llm.event_handler("on_function_calls_cancelled")
async def on_function_calls_started(service, function_calls): async def on_function_calls_cancelled(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now.")) for item in function_calls:
logger.info(f"Function call cancelled: {item.function_name} [{item.tool_call_id}]")
location_function = FunctionSchema( location_function = FunctionSchema(
name="track_current_location", name="track_current_location",

View File

@@ -77,6 +77,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = AnthropicLLMService( llm = AnthropicLLMService(
api_key=os.getenv("ANTHROPIC_API_KEY"), api_key=os.getenv("ANTHROPIC_API_KEY"),
enable_async_tool_cancellation=True,
settings=AnthropicLLMService.Settings( settings=AnthropicLLMService.Settings(
system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.", system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
), ),
@@ -92,6 +93,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
) )
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
@llm.event_handler("on_function_calls_cancelled")
async def on_function_calls_cancelled(service, function_calls):
for item in function_calls:
logger.info(f"Function call cancelled: {item.function_name} [{item.tool_call_id}]")
weather_function = FunctionSchema( weather_function = FunctionSchema(
name="get_current_weather", name="get_current_weather",
description="Get the current weather", description="Get the current weather",

View File

@@ -118,6 +118,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = GoogleLLMService( llm = GoogleLLMService(
api_key=os.getenv("GOOGLE_API_KEY"), api_key=os.getenv("GOOGLE_API_KEY"),
enable_async_tool_cancellation=True,
settings=GoogleLLMService.Settings( settings=GoogleLLMService.Settings(
system_instruction=( system_instruction=(
"You are a helpful assistant in a voice conversation. " "You are a helpful assistant in a voice conversation. "
@@ -143,6 +144,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
async def on_function_calls_started(service, function_calls): async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now.")) await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now."))
@llm.event_handler("on_function_calls_cancelled")
async def on_function_calls_cancelled(service, function_calls):
for item in function_calls:
logger.info(f"Function call cancelled: {item.function_name} [{item.tool_call_id}]")
location_function = FunctionSchema( location_function = FunctionSchema(
name="track_current_location", name="track_current_location",
description="Start tracking the user's current GPS location, reporting position updates until the user reaches their destination.", description="Start tracking the user's current GPS location, reporting position updates until the user reaches their destination.",

View File

@@ -128,6 +128,7 @@ indicate you should use the get_image tool are:
llm = GoogleLLMService( llm = GoogleLLMService(
api_key=os.getenv("GOOGLE_API_KEY"), api_key=os.getenv("GOOGLE_API_KEY"),
enable_async_tool_cancellation=True,
settings=GoogleLLMService.Settings( settings=GoogleLLMService.Settings(
system_instruction=system_prompt, system_instruction=system_prompt,
), ),
@@ -140,6 +141,11 @@ indicate you should use the get_image tool are:
async def on_function_calls_started(service, function_calls): async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
@llm.event_handler("on_function_calls_cancelled")
async def on_function_calls_cancelled(service, function_calls):
for item in function_calls:
logger.info(f"Function call cancelled: {item.function_name} [{item.tool_call_id}]")
weather_function = FunctionSchema( weather_function = FunctionSchema(
name="get_weather", name="get_weather",
description="Get the current weather", description="Get the current weather",

View File

@@ -118,6 +118,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = OpenAILLMService( llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"), api_key=os.getenv("OPENAI_API_KEY"),
enable_async_tool_cancellation=True,
settings=OpenAILLMService.Settings( settings=OpenAILLMService.Settings(
system_instruction=( system_instruction=(
"You are a helpful assistant in a voice conversation. " "You are a helpful assistant in a voice conversation. "
@@ -143,6 +144,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
async def on_function_calls_started(service, function_calls): async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now.")) await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now."))
@llm.event_handler("on_function_calls_cancelled")
async def on_function_calls_cancelled(service, function_calls):
for item in function_calls:
logger.info(f"Function call cancelled: {item.function_name} [{item.tool_call_id}]")
location_function = FunctionSchema( location_function = FunctionSchema(
name="track_current_location", name="track_current_location",
description="Start tracking the user's current GPS location, reporting position updates until the user reaches their destination.", description="Start tracking the user's current GPS location, reporting position updates until the user reaches their destination.",
@@ -181,6 +187,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
async def on_client_connected(transport, client): async def on_client_connected(transport, client):
logger.info(f"Client connected") logger.info(f"Client connected")
# Kick off the conversation. # Kick off the conversation.
context.add_message(
{"role": "developer", "content": "Please introduce yourself to the user."}
)
await task.queue_frames([LLMRunFrame()]) await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected") @transport.event_handler("on_client_disconnected")

View File

@@ -87,6 +87,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = OpenAILLMService( llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"), api_key=os.getenv("OPENAI_API_KEY"),
enable_async_tool_cancellation=True,
settings=OpenAILLMService.Settings( settings=OpenAILLMService.Settings(
system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.", system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
), ),
@@ -106,6 +107,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
async def on_function_calls_started(service, function_calls): async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
@llm.event_handler("on_function_calls_cancelled")
async def on_function_calls_cancelled(service, function_calls):
for item in function_calls:
logger.info(f"Function call cancelled: {item.function_name} [{item.tool_call_id}]")
weather_function = FunctionSchema( weather_function = FunctionSchema(
name="get_current_weather", name="get_current_weather",
description="Get the current weather", description="Get the current weather",
@@ -165,6 +171,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
async def on_client_connected(transport, client): async def on_client_connected(transport, client):
logger.info(f"Client connected") logger.info(f"Client connected")
# Kick off the conversation. # Kick off the conversation.
context.add_message(
{"role": "developer", "content": "Please introduce yourself to the user."}
)
await task.queue_frames([LLMRunFrame()]) await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected") @transport.event_handler("on_client_disconnected")

View File

@@ -118,6 +118,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = OpenAIResponsesLLMService( llm = OpenAIResponsesLLMService(
api_key=os.getenv("OPENAI_API_KEY"), api_key=os.getenv("OPENAI_API_KEY"),
enable_async_tool_cancellation=True,
settings=OpenAIResponsesLLMService.Settings( settings=OpenAIResponsesLLMService.Settings(
system_instruction=( system_instruction=(
"You are a helpful assistant in a voice conversation. " "You are a helpful assistant in a voice conversation. "
@@ -143,6 +144,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
async def on_function_calls_started(service, function_calls): async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now.")) await tts.queue_frame(TTSSpeakFrame("Sure, tracking your location now."))
@llm.event_handler("on_function_calls_cancelled")
async def on_function_calls_cancelled(service, function_calls):
for item in function_calls:
logger.info(f"Function call cancelled: {item.function_name} [{item.tool_call_id}]")
location_function = FunctionSchema( location_function = FunctionSchema(
name="track_current_location", name="track_current_location",
description="Track the device's current GPS location during a road trip, reporting position updates as the vehicle moves through cities until it reaches the final destination.", description="Track the device's current GPS location during a road trip, reporting position updates as the vehicle moves through cities until it reaches the final destination.",

View File

@@ -77,6 +77,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = OpenAIResponsesLLMService( llm = OpenAIResponsesLLMService(
api_key=os.getenv("OPENAI_API_KEY"), api_key=os.getenv("OPENAI_API_KEY"),
enable_async_tool_cancellation=True,
settings=OpenAIResponsesLLMService.Settings( settings=OpenAIResponsesLLMService.Settings(
system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.", system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
), ),
@@ -104,6 +105,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
# matching, forcing a full context resend. # matching, forcing a full context resend.
await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False)) await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False))
@llm.event_handler("on_function_calls_cancelled")
async def on_function_calls_cancelled(service, function_calls):
for item in function_calls:
logger.info(f"Function call cancelled: {item.function_name} [{item.tool_call_id}]")
weather_function = FunctionSchema( weather_function = FunctionSchema(
name="get_current_weather", name="get_current_weather",
description="Get the current weather", description="Get the current weather",

View File

@@ -10,11 +10,13 @@ This module provides the abstract base class for implementing LLM provider-speci
adapters that handle tool format conversion and standardization. adapters that handle tool format conversion and standardization.
""" """
import warnings
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import Any, Dict, Generic, List, Optional, TypeVar from typing import Any, Dict, Generic, List, Optional, TypeVar
from loguru import logger from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import ( from pipecat.processors.aggregators.llm_context import (
LLMContext, LLMContext,
@@ -48,6 +50,21 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
def __init__(self): def __init__(self):
"""Initialize the adapter.""" """Initialize the adapter."""
self._warned_system_instruction = False self._warned_system_instruction = False
self._builtin_tools: Dict[str, FunctionSchema] = {}
@property
def builtin_tools(self) -> Dict[str, FunctionSchema]:
"""Built-in tools automatically merged into every inference request.
Keyed by tool name for O(1) lookup, insertion, and removal. The
service injects tools here so they are sent transparently on every
inference request without the user having to add them to their
``ToolsSchema``.
Returns:
Mutable dict mapping tool name to ``FunctionSchema``.
"""
return self._builtin_tools
@property @property
@abstractmethod @abstractmethod
@@ -122,6 +139,9 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
def from_standard_tools(self, tools: Any) -> List[Any] | NotGiven: def from_standard_tools(self, tools: Any) -> List[Any] | NotGiven:
"""Convert tools from standard format to provider format. """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: Args:
tools: Tools in standard format or provider-specific format. tools: Tools in standard format or provider-specific format.
@@ -129,8 +149,31 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
List of tools converted to provider format, or original tools List of tools converted to provider format, or original tools
if not in standard format. if not in standard format.
""" """
if self._builtin_tools:
if isinstance(tools, ToolsSchema):
tools = ToolsSchema(
standard_tools=tools.standard_tools + list(self._builtin_tools.values()),
custom_tools=tools.custom_tools,
)
else:
# User supplied tools in a legacy/provider-specific format.
# Built-in tools cannot be safely merged, so they will not be injected.
# Migrate to ToolsSchema to enable built-in tool support; use custom_tools
# as an escape hatch for any provider-specific tools that don't fit the
# standard schema.
if tools is not None:
warnings.warn(
"Built-in tools (e.g. async tool cancellation) could not be injected "
"because the supplied tools are not a ToolsSchema instance. "
"Migrate to ToolsSchema to enable built-in tool support. "
"Use ToolsSchema(custom_tools=...) as an escape hatch for any "
"provider-specific tools that don't fit the standard schema.",
DeprecationWarning,
stacklevel=2,
)
# Fall through and return the original tools unchanged.
if isinstance(tools, ToolsSchema): if isinstance(tools, ToolsSchema):
logger.debug(f"Retrieving the tools using the adapter: {type(self)}")
return self.to_provider_tools_format(tools) return self.to_provider_tools_format(tools)
# Fallback to return the same tools in case they are not in a standard format # Fallback to return the same tools in case they are not in a standard format
return tools return tools

View File

@@ -16,6 +16,7 @@ from typing import (
Awaitable, Awaitable,
Callable, Callable,
Dict, Dict,
List,
Mapping, Mapping,
Optional, Optional,
Protocol, Protocol,
@@ -60,6 +61,11 @@ from pipecat.services.ai_service import AIService
from pipecat.services.settings import LLMSettings from pipecat.services.settings import LLMSettings
from pipecat.services.websocket_service import WebsocketService from pipecat.services.websocket_service import WebsocketService
from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionLLMServiceMixin 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 ( from pipecat.utils.context.llm_context_summarization import (
DEFAULT_SUMMARIZATION_TIMEOUT, DEFAULT_SUMMARIZATION_TIMEOUT,
LLMContextSummarizationUtil, LLMContextSummarizationUtil,
@@ -177,7 +183,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
- on_completion_timeout: Called when an LLM completion timeout occurs - on_completion_timeout: Called when an LLM completion timeout occurs
- on_function_calls_started: Called when function calls are received and - on_function_calls_started: Called when function calls are received and
execution is about to start execution is about to start. Built-in tools (e.g. ``cancel_async_tool_call``)
are excluded from this event.
- on_function_calls_cancelled: Called after one or more async tool calls are
cancelled.
Example:: Example::
@@ -186,8 +195,12 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
logger.warning("LLM completion timed out") logger.warning("LLM completion timed out")
@task.event_handler("on_function_calls_started") @task.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls): async def on_function_calls_started(service, function_calls: List[FunctionCallFromLLM]):
logger.info(f"Starting {len(function_calls)} function calls") logger.info(f"Starting {len(function_calls)} function calls")
@task.event_handler("on_function_calls_cancelled")
async def on_function_calls_cancelled(service, function_calls: List[FunctionCallFromLLM]):
logger.info(f"Cancelled {len(function_calls)} function calls")
""" """
_settings: LLMSettings _settings: LLMSettings
@@ -201,6 +214,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
run_in_parallel: bool = True, run_in_parallel: bool = True,
group_parallel_tools: bool = True, group_parallel_tools: bool = True,
function_call_timeout_secs: Optional[float] = None, function_call_timeout_secs: Optional[float] = None,
enable_async_tool_cancellation: bool = False,
settings: Optional[LLMSettings] = None, settings: Optional[LLMSettings] = None,
**kwargs, **kwargs,
): ):
@@ -215,6 +229,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
arrives. Defaults to True. arrives. Defaults to True.
function_call_timeout_secs: Optional timeout in seconds for deferred function function_call_timeout_secs: Optional timeout in seconds for deferred function
calls. calls.
enable_async_tool_cancellation: When True and at least one async function
(``cancel_on_interruption=False``) is registered, automatically injects
the ``cancel_async_tool_call`` built-in tool and its system instructions
so the LLM can cancel stale in-progress calls. Defaults to False.
settings: The runtime-updatable settings for the LLM service. settings: The runtime-updatable settings for the LLM service.
**kwargs: Additional arguments passed to the parent AIService. **kwargs: Additional arguments passed to the parent AIService.
@@ -229,7 +247,9 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
self._run_in_parallel = run_in_parallel self._run_in_parallel = run_in_parallel
self._group_parallel_tools = group_parallel_tools self._group_parallel_tools = group_parallel_tools
self._function_call_timeout_secs = function_call_timeout_secs self._function_call_timeout_secs = function_call_timeout_secs
self._enable_async_tool_cancellation: bool = enable_async_tool_cancellation
self._filter_incomplete_user_turns: bool = False self._filter_incomplete_user_turns: bool = False
self._async_tool_cancellation_enabled: bool = False
self._base_system_instruction: Optional[str] = None self._base_system_instruction: Optional[str] = None
self._adapter = self.adapter_class() self._adapter = self.adapter_class()
self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {}
@@ -239,6 +259,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
self._summary_task: Optional[asyncio.Task] = None self._summary_task: Optional[asyncio.Task] = None
self._register_event_handler("on_function_calls_started") self._register_event_handler("on_function_calls_started")
self._register_event_handler("on_function_calls_cancelled")
self._register_event_handler("on_completion_timeout") self._register_event_handler("on_completion_timeout")
def get_llm_adapter(self) -> BaseLLMAdapter: def get_llm_adapter(self) -> BaseLLMAdapter:
@@ -291,6 +312,8 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
await super().start(frame) await super().start(frame)
if not self._run_in_parallel: if not self._run_in_parallel:
await self._create_sequential_runner_task() await self._create_sequential_runner_task()
if self._enable_async_tool_cancellation and self._has_async_tools():
self._setup_async_tool_cancellation()
async def stop(self, frame: EndFrame): async def stop(self, frame: EndFrame):
"""Stop the LLM service. """Stop the LLM service.
@@ -315,17 +338,20 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
await self._cancel_summary_task() await self._cancel_summary_task()
def _compose_system_instruction(self): 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 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 base = self._base_system_instruction
completion_instructions = self._user_turn_completion_config.completion_instructions parts = [base] if base else []
if base: if self._filter_incomplete_user_turns:
self._settings.system_instruction = f"{base}\n\n{completion_instructions}" parts.append(self._user_turn_completion_config.completion_instructions)
else: if self._async_tool_cancellation_enabled:
self._settings.system_instruction = completion_instructions 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]: async def _update_settings(self, delta: LLMSettings) -> dict[str, Any]:
"""Apply a settings delta, handling turn-completion fields. """Apply a settings delta, handling turn-completion fields.
@@ -361,10 +387,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
if ( if (
"system_instruction" in changed "system_instruction" in changed
and self._filter_incomplete_user_turns and (self._filter_incomplete_user_turns or self._async_tool_cancellation_enabled)
and "filter_incomplete_user_turns" not in changed 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. # Treat the new value as the new base and recompose.
self._base_system_instruction = self._settings.system_instruction self._base_system_instruction = self._settings.system_instruction
self._compose_system_instruction() self._compose_system_instruction()
@@ -576,6 +602,11 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
``function_call_timeout_secs`` for this specific function. Defaults to ``function_call_timeout_secs`` for this specific function. Defaults to
None, which uses the global timeout. None, which uses the global timeout.
""" """
if function_name == CANCEL_ASYNC_TOOL_NAME:
raise ValueError(
f"'{CANCEL_ASYNC_TOOL_NAME}' is a reserved built-in tool name and cannot be "
"registered by user code."
)
# Registering a function with the function_name set to None will run # Registering a function with the function_name set to None will run
# that handler for all functions # that handler for all functions
self._functions[function_name] = FunctionCallRegistryItem( self._functions[function_name] = FunctionCallRegistryItem(
@@ -610,6 +641,11 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
None, which uses the global timeout. None, which uses the global timeout.
""" """
wrapper = DirectFunctionWrapper(handler) wrapper = DirectFunctionWrapper(handler)
if wrapper.name == CANCEL_ASYNC_TOOL_NAME:
raise ValueError(
f"'{CANCEL_ASYNC_TOOL_NAME}' is a reserved built-in tool name and cannot be "
"registered by user code."
)
self._functions[wrapper.name] = FunctionCallRegistryItem( self._functions[wrapper.name] = FunctionCallRegistryItem(
function_name=wrapper.name, function_name=wrapper.name,
handler=wrapper, handler=wrapper,
@@ -624,6 +660,8 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
function_name: The name of the function handler to remove. function_name: The name of the function handler to remove.
""" """
del self._functions[function_name] del self._functions[function_name]
if self._async_tool_cancellation_enabled and not self._has_async_tools():
self._teardown_async_tool_cancellation()
def unregister_direct_function(self, handler: Any): def unregister_direct_function(self, handler: Any):
"""Remove a registered direct function handler. """Remove a registered direct function handler.
@@ -634,6 +672,8 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
wrapper = DirectFunctionWrapper(handler) wrapper = DirectFunctionWrapper(handler)
del self._functions[wrapper.name] del self._functions[wrapper.name]
# Note: no need to remove start callback here, as direct functions don't support start callbacks. # Note: no need to remove start callback here, as direct functions don't support start callbacks.
if self._async_tool_cancellation_enabled and not self._has_async_tools():
self._teardown_async_tool_cancellation()
def has_function(self, function_name: str): def has_function(self, function_name: str):
"""Check if a function handler is registered. """Check if a function handler is registered.
@@ -661,9 +701,14 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
if len(function_calls) == 0: if len(function_calls) == 0:
return return
await self._call_event_handler("on_function_calls_started", function_calls) # Exclude the built-in cancel tool — it's an internal mechanism and
# should not be surfaced to user-facing event handlers or frames.
await self.broadcast_frame(FunctionCallsStartedFrame, function_calls=function_calls) user_visible_calls = [
fc for fc in function_calls if fc.function_name != CANCEL_ASYNC_TOOL_NAME
]
if user_visible_calls:
await self._call_event_handler("on_function_calls_started", user_visible_calls)
await self.broadcast_frame(FunctionCallsStartedFrame, function_calls=user_visible_calls)
# When group_parallel_tools is True all calls share a group_id so the # When group_parallel_tools is True all calls share a group_id so the
# aggregator triggers the LLM exactly once after the last one completes. # aggregator triggers the LLM exactly once after the last one completes.
@@ -849,8 +894,118 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
if timeout_task and not timeout_task.done(): if timeout_task and not timeout_task.done():
await self.cancel_task(timeout_task) await self.cancel_task(timeout_task)
def _has_async_tools(self) -> bool:
"""Return True if at least one non-builtin async tool 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 dict.
"""
logger.debug(f"{self}: Enabling async tool cancellation")
self._async_tool_cancellation_enabled = True
if self._base_system_instruction is None:
self._base_system_instruction = self._settings.system_instruction
self._compose_system_instruction()
self._adapter.builtin_tools[CANCEL_ASYNC_TOOL_NAME] = 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,
)
def _teardown_async_tool_cancellation(self):
"""Disable async tool cancellation.
Removes the built-in ``cancel_async_tool_call`` handler and its schema,
recomposes the system instruction without cancellation instructions.
"""
logger.debug(f"{self}: Disabling async tool cancellation")
self._async_tool_cancellation_enabled = False
self._adapter.builtin_tools.pop(CANCEL_ASYNC_TOOL_NAME, None)
self._functions.pop(CANCEL_ASYNC_TOOL_NAME, None)
self._compose_system_instruction()
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.debug(f"{self}: cancel_async_tool_call 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()
cancelled_items = []
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
)
cancelled_items.append(
FunctionCallFromLLM(
function_name=runner_item.function_name,
tool_call_id=runner_item.tool_call_id,
arguments=runner_item.arguments,
context=runner_item.context,
)
)
logger.debug(f"{self} Async function call [{name}:{tool_call_id}] cancelled")
for task in cancelled_tasks:
self._function_call_task_finished(task)
if cancelled_items:
await self._call_event_handler("on_function_calls_cancelled", cancelled_items)
async def _cancel_function_call(self, function_name: Optional[str]): async def _cancel_function_call(self, function_name: Optional[str]):
cancelled_tasks = set() cancelled_tasks = set()
cancelled_items = []
for task, runner_item in self._function_call_tasks.items(): for task, runner_item in self._function_call_tasks.items():
if runner_item.registry_item.function_name == function_name: if runner_item.registry_item.function_name == function_name:
name = runner_item.function_name name = runner_item.function_name
@@ -870,12 +1025,23 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
FunctionCallCancelFrame, function_name=name, tool_call_id=tool_call_id FunctionCallCancelFrame, function_name=name, tool_call_id=tool_call_id
) )
cancelled_items.append(
FunctionCallFromLLM(
function_name=runner_item.function_name,
tool_call_id=runner_item.tool_call_id,
arguments=runner_item.arguments,
context=runner_item.context,
)
)
logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled") logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled")
# Remove all cancelled tasks from our set. # Remove all cancelled tasks from our set.
for task in cancelled_tasks: for task in cancelled_tasks:
self._function_call_task_finished(task) self._function_call_task_finished(task)
if cancelled_items:
await self._call_event_handler("on_function_calls_cancelled", cancelled_items)
def _function_call_task_finished(self, task: asyncio.Task): def _function_call_task_finished(self, task: asyncio.Task):
if task in self._function_call_tasks: if task in self._function_call_tasks:
del self._function_call_tasks[task] del self._function_call_tasks[task]

View File

@@ -0,0 +1,51 @@
#
# 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": "async_tool", "status": "running", and a "tool_call_id" field containing the \
exact ID of that call (e.g. {"type": "async_tool", "status": "running", "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": "running" 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 whose results are 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 copied exactly from the 'tool_call_id' field "
"in the async tool's 'running' response visible in the conversation history."
),
properties={
"tool_call_id": {
"type": "string",
"description": (
"The exact tool_call_id from the async tool's 'running' response to cancel."
),
}
},
required=["tool_call_id"],
)