wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enable async tool cancellation feature.

Browse Source
This commit is contained in:
filipi87
2026-04-09 10:29:23 -03:00
parent 76601944c6
commit 772fb57090
3 changed files with 191 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/pipecat/adapters/base_llm_adapter.py
View File

@@ -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

115
src/pipecat/services/llm_service.py
View File

@@ -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():

49
src/pipecat/utils/async_tool_cancellation.py Normal file
View File

@@ -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"],
)