From 283f6df20579ad10bf57a6dd2d4035e946d76fa9 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 2 Apr 2026 16:57:22 -0300 Subject: [PATCH 01/13] Creating a FrameQueue so we can properly reset without discarding uninterruptible frames. --- src/pipecat/utils/frame_queue.py | 71 ++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 src/pipecat/utils/frame_queue.py diff --git a/src/pipecat/utils/frame_queue.py b/src/pipecat/utils/frame_queue.py new file mode 100644 index 000000000..888bacb8e --- /dev/null +++ b/src/pipecat/utils/frame_queue.py @@ -0,0 +1,71 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Frame queue utilities for Pipecat pipeline processors.""" + +import asyncio +from typing import Any, Callable + +from pipecat.frames.frames import Frame, UninterruptibleFrame + + +class FrameQueue(asyncio.Queue): + """An asyncio.Queue that tracks whether any UninterruptibleFrame is enqueued. + + Extends ``asyncio.Queue`` and maintains an O(1) ``has_uninterruptible`` + flag so interrupt-handling code can decide whether to cancel a task or + merely drain non-uninterruptible items without scanning the queue. + + Items may be raw ``Frame`` objects or tuples whose first element is a + ``Frame`` (e.g. ``(frame, direction, callback)``). Pass a ``frame_getter`` + callable to extract the frame from each item; the default treats the item + itself as the frame. + + Also exposes a ``reset()`` helper that drains all non-``UninterruptibleFrame`` + items while keeping uninterruptible ones in place. + """ + + def __init__(self, frame_getter: Callable[[Any], Frame] = lambda item: item): + """Initialize the FrameQueue. + + Args: + frame_getter: Callable that extracts a ``Frame`` from a queue item. + Defaults to the identity function (item is a raw ``Frame``). + Pass ``lambda item: item[0]`` when items are + ``(frame, direction, callback)`` tuples. + """ + super().__init__() + self._frame_getter = frame_getter + self._uninterruptible_count: int = 0 + + @property + def has_uninterruptible(self) -> bool: + """Return True if any UninterruptibleFrame is currently in the queue.""" + return self._uninterruptible_count > 0 + + def _put(self, item: Any) -> None: + if isinstance(self._frame_getter(item), UninterruptibleFrame): + self._uninterruptible_count += 1 + super()._put(item) + + def _get(self) -> Any: + item = super()._get() + if isinstance(self._frame_getter(item), UninterruptibleFrame): + self._uninterruptible_count -= 1 + return item + + def reset(self) -> None: + """Remove all non-UninterruptibleFrame items, keeping uninterruptible ones.""" + kept: asyncio.Queue = asyncio.Queue() + while not self.empty(): + item = self.get_nowait() + if isinstance(self._frame_getter(item), UninterruptibleFrame): + kept.put_nowait(item) + self.task_done() + while not kept.empty(): + item = kept.get_nowait() + self.put_nowait(item) + kept.task_done() From 4c8734c5e1ba5651f2a1489139aca057c37fb240 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 2 Apr 2026 16:57:46 -0300 Subject: [PATCH 02/13] Fixing an issue where the BotOutputTransport was discarding the UninterruptibleFrames. --- src/pipecat/processors/frame_processor.py | 31 ++++++++--------------- src/pipecat/transports/base_output.py | 13 +++++++--- 2 files changed, 21 insertions(+), 23 deletions(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 02cf6ce7b..fe9e5b90f 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -48,6 +48,7 @@ from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FrameP from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics from pipecat.utils.asyncio.task_manager import BaseTaskManager from pipecat.utils.base_object import BaseObject +from pipecat.utils.frame_queue import FrameQueue class FrameDirection(Enum): @@ -228,7 +229,7 @@ class FrameProcessor(BaseObject): # called. To resume processing frames we need to call # `resume_processing_frames()` which will wake up the event. self.__should_block_frames = False - self.__process_queue = asyncio.Queue() + self.__process_queue = FrameQueue(frame_getter=lambda item: item[0]) self.__process_event: Optional[asyncio.Event] = None self.__process_frame_task: Optional[asyncio.Task] = None self.__process_current_frame: Optional[Frame] = None @@ -818,9 +819,14 @@ class FrameProcessor(BaseObject): async def _start_interruption(self): """Start handling an interruption by cancelling current tasks.""" try: - if isinstance(self.__process_current_frame, UninterruptibleFrame): - # We don't want to cancel UninterruptibleFrame, so we simply - # cleanup the queue. + current_is_uninterruptible = isinstance( + self.__process_current_frame, UninterruptibleFrame + ) + if current_is_uninterruptible or self.__process_queue.has_uninterruptible: + # We don't want to cancel an UninterruptibleFrame (either the + # one currently being processed or one waiting in the queue), + # so we simply cleanup the queue keeping only + # UninterruptibleFrames. self.__reset_process_queue() else: # Cancel and re-create the process task. @@ -920,22 +926,7 @@ class FrameProcessor(BaseObject): def __reset_process_queue(self): """Reset non-system frame processing queue.""" - # Create a new queue to insert UninterruptibleFrame frames. - new_queue = asyncio.Queue() - - # Process current queue and keep UninterruptibleFrame frames. - while not self.__process_queue.empty(): - item = self.__process_queue.get_nowait() - frame = item[0] - if isinstance(frame, UninterruptibleFrame): - new_queue.put_nowait(item) - self.__process_queue.task_done() - - # Put back UninterruptibleFrame frames into our process queue. - while not new_queue.empty(): - item = new_queue.get_nowait() - self.__process_queue.put_nowait(item) - new_queue.task_done() + self.__process_queue.reset() async def __cancel_process_task(self): """Cancel the non-system frame processing task.""" diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 1d26d5e53..ee4571425 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -48,6 +48,7 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.base_transport import TransportParams +from pipecat.utils.frame_queue import FrameQueue from pipecat.utils.time import nanoseconds_to_seconds BOT_VAD_STOP_SECS = 0.35 @@ -518,14 +519,20 @@ class BaseOutputTransport(FrameProcessor): _: The start interruption frame (unused). """ # Cancel tasks. - await self._cancel_audio_task() await self._cancel_clock_task() await self._cancel_video_task() + if self._audio_queue.has_uninterruptible: + # Keep the audio task running but drain all interruptible frames + # so the pending UninterruptibleFrames are still delivered. + self._audio_queue.reset() + else: + await self._cancel_audio_task() + self._create_audio_task() + # Create tasks. self._create_video_task() self._create_clock_task() - self._create_audio_task() # Let's send a bot stopped speaking if we have to. await self._bot_stopped_speaking() @@ -609,7 +616,7 @@ class BaseOutputTransport(FrameProcessor): def _create_audio_task(self): """Create the audio processing task.""" if not self._audio_task: - self._audio_queue = asyncio.Queue() + self._audio_queue = FrameQueue() self._audio_task = self._transport.create_task(self._audio_task_handler()) async def _cancel_audio_task(self): From 3724ecd378eaf91550251f040cc00c1fbc820b28 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 2 Apr 2026 16:58:19 -0300 Subject: [PATCH 03/13] Supporting async function calls. --- .../function-calling-anthropic.py | 21 ++++-- .../function-calling-openai.py | 15 +++- src/pipecat/frames/frames.py | 7 ++ .../aggregators/llm_response_universal.py | 75 +++++++++++++++---- src/pipecat/services/llm_service.py | 32 +++++++- .../context/llm_context_summarization.py | 64 +++++++++++++++- 6 files changed, 184 insertions(+), 30 deletions(-) diff --git a/examples/function-calling/function-calling-anthropic.py b/examples/function-calling/function-calling-anthropic.py index b8bb4eac6..09cafab8e 100644 --- a/examples/function-calling/function-calling-anthropic.py +++ b/examples/function-calling/function-calling-anthropic.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # - +import asyncio import os from dotenv import load_dotenv @@ -35,9 +35,10 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams load_dotenv(override=True) -async def get_weather(params: FunctionCallParams): - location = params.arguments["location"] - await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") +async def fetch_weather_from_api(params: FunctionCallParams): + # Simulate a long-running API call, so we can test async function calls. + await asyncio.sleep(20) + await params.result_callback({"conditions": "nice", "temperature": "75"}) async def fetch_restaurant_recommendation(params: FunctionCallParams): @@ -80,11 +81,19 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): 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.", ), ) - llm.register_function("get_weather", get_weather) + + # You can also register a function_name of None to get all functions + # sent to the same callback with an additional function_name parameter. + llm.register_function( + "get_current_weather", + fetch_weather_from_api, + cancel_on_interruption=False, + timeout_secs=30, + ) llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) weather_function = FunctionSchema( - name="get_weather", + name="get_current_weather", description="Get the current weather", properties={ "location": { diff --git a/examples/function-calling/function-calling-openai.py b/examples/function-calling/function-calling-openai.py index 2b59d7072..b5c5b83ac 100644 --- a/examples/function-calling/function-calling-openai.py +++ b/examples/function-calling/function-calling-openai.py @@ -4,6 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import asyncio import os from dotenv import load_dotenv @@ -12,7 +13,10 @@ from loguru import logger from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame +from pipecat.frames.frames import ( + LLMRunFrame, + TTSSpeakFrame, +) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -35,6 +39,8 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): + # Simulate a long-running API call, so we can test async function calls. + await asyncio.sleep(20) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -88,7 +94,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function( + "get_current_weather", + fetch_weather_from_api, + cancel_on_interruption=False, + timeout_secs=30, + ) llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) @llm.event_handler("on_function_calls_started") diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 86a93825b..27796ff62 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1642,12 +1642,19 @@ class FunctionCallInProgressFrame(ControlFrame, UninterruptibleFrame): tool_call_id: Unique identifier for this function call. arguments: Arguments passed to the function. cancel_on_interruption: Whether to cancel this call if interrupted. + When ``False`` the call is treated as asynchronous: the LLM + continues the conversation immediately without waiting for the + result, and the result is injected later via a developer message. + group_id: Identifier shared by all function calls originating from the + same LLM response batch. Used to determine when the last call in a + group completes so the LLM can be triggered exactly once. """ function_name: str tool_call_id: str arguments: Any cancel_on_interruption: bool = False + group_id: Optional[str] = None @dataclass diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index f8c2ffd43..b99b251d4 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -866,6 +866,8 @@ class LLMAssistantAggregator(LLMContextAggregator): self._function_calls_image_results: Dict[str, UserImageRawFrame] = {} self._context_updated_tasks: Set[asyncio.Task] = set() + self._user_speaking: bool = False + self._assistant_turn_start_timestamp = "" self._thought_append_to_context = False @@ -968,6 +970,12 @@ class LLMAssistantAggregator(LLMContextAggregator): await self._handle_user_image_frame(frame) elif isinstance(frame, AssistantImageRawFrame): await self._handle_assistant_image_frame(frame) + elif isinstance(frame, UserStartedSpeakingFrame): + self._user_speaking = True + await self.push_frame(frame, direction) + elif isinstance(frame, UserStoppedSpeakingFrame): + self._user_speaking = False + await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) @@ -1047,13 +1055,24 @@ class LLMAssistantAggregator(LLMContextAggregator): ], } ) - self._context.add_message( - { - "role": "tool", - "content": "IN_PROGRESS", - "tool_call_id": frame.tool_call_id, - } - ) + + is_async = not frame.cancel_on_interruption + if is_async: + self._context.add_message( + { + "role": "tool", + "content": json.dumps({"type": "async_tool", "status": "started"}), + "tool_call_id": frame.tool_call_id, + } + ) + else: + self._context.add_message( + { + "role": "tool", + "content": "IN_PROGRESS", + "tool_call_id": frame.tool_call_id, + } + ) self._function_calls_in_progress[frame.tool_call_id] = frame @@ -1067,16 +1086,34 @@ class LLMAssistantAggregator(LLMContextAggregator): ) return + in_progress_frame = self._function_calls_in_progress[frame.tool_call_id] + is_async = not in_progress_frame.cancel_on_interruption if in_progress_frame else False + group_id = in_progress_frame.group_id if in_progress_frame else None + del self._function_calls_in_progress[frame.tool_call_id] properties = frame.properties - # Update context with the function call result - if frame.result: - result = json.dumps(frame.result, ensure_ascii=False) - self._update_function_call_result(frame.function_name, frame.tool_call_id, result) + result = json.dumps(frame.result, ensure_ascii=False) if frame.result else "COMPLETED" + + if is_async: + # For async function calls instead of updating the existing IN_PROGRESS tool message we inject + # a new developer message so the LLM is notified of the completed result. + self._context.add_message( + { + "role": "developer", + "content": json.dumps( + { + "type": "async_tool", + "tool_call_id": frame.tool_call_id, + "status": "finished", + "result": result, + } + ), + } + ) else: - self._update_function_call_result(frame.function_name, frame.tool_call_id, "COMPLETED") + self._update_function_call_result(frame.function_name, frame.tool_call_id, result) run_llm = False @@ -1098,10 +1135,18 @@ class LLMAssistantAggregator(LLMContextAggregator): # If the frame is indicating we should run the LLM, do it. run_llm = frame.run_llm else: - # If this is the last function call in progress, run the LLM. - run_llm = not bool(self._function_calls_in_progress) + # Run the LLM when this is the last function call in the group + # to complete. If group_id is set, only consider sibling calls; + # otherwise always execute as soon as we receive the result. + if group_id: + run_llm = not any( + f is not None and f.group_id == group_id + for f in self._function_calls_in_progress.values() + ) + else: + run_llm = True - if run_llm: + if run_llm and not self._user_speaking: await self.push_context_frame(FrameDirection.UPSTREAM) # Call the `on_context_updated` callback once the function call result diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 37d7d094a..1430aceac 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -7,8 +7,8 @@ """Base classes for Large Language Model services with function calling support.""" import asyncio -import inspect import json +import uuid import warnings from dataclasses import dataclass from typing import ( @@ -119,6 +119,9 @@ class FunctionCallRegistryItem: function_name: The name of the function (None for catch-all handler). handler: The handler for processing function call parameters. cancel_on_interruption: Whether to cancel the call on interruption. + When ``False`` the call is treated as asynchronous: the LLM + continues the conversation immediately without waiting for the + result, and the result is injected later via a developer message. timeout_secs: Optional per-tool timeout in seconds. Overrides the global ``function_call_timeout_secs`` for this specific function. """ @@ -142,6 +145,9 @@ class FunctionCallRunnerItem: arguments: The arguments for the function. context: The LLM context. run_llm: Optional flag to control LLM execution after function call. + group_id: Shared identifier for all function calls from the same LLM + response batch. Used to trigger the LLM exactly once when the last + call in the group completes. """ registry_item: FunctionCallRegistryItem @@ -150,6 +156,7 @@ class FunctionCallRunnerItem: arguments: Mapping[str, Any] context: LLMContext run_llm: Optional[bool] = None + group_id: Optional[str] = None class LLMService(UserTurnCompletionLLMServiceMixin, AIService): @@ -185,6 +192,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): def __init__( self, run_in_parallel: bool = True, + group_parallel_tools: bool = True, function_call_timeout_secs: Optional[float] = None, settings: Optional[LLMSettings] = None, **kwargs, @@ -194,6 +202,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): Args: run_in_parallel: Whether to run function calls in parallel or sequentially. Defaults to True. + group_parallel_tools: Whether to group parallel function calls so the LLM + is triggered exactly once after all calls in the batch complete. When + False, each function call result triggers the LLM independently as it + arrives. Defaults to True. function_call_timeout_secs: Optional timeout in seconds for deferred function calls. settings: The runtime-updatable settings for the LLM service. @@ -208,6 +220,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): **kwargs, ) self._run_in_parallel = run_in_parallel + self._group_parallel_tools = group_parallel_tools self._function_call_timeout_secs = function_call_timeout_secs self._filter_incomplete_user_turns: bool = False self._base_system_instruction: Optional[str] = None @@ -548,7 +561,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): handler: The function handler. Should accept a single FunctionCallParams parameter. cancel_on_interruption: Whether to cancel this function call when an - interruption occurs. Defaults to True. + interruption occurs. When ``False`` the call is treated as + asynchronous: the LLM continues the conversation immediately + without waiting for the result, and the result is injected later + via a developer message. Defaults to True. timeout_secs: Optional per-tool timeout in seconds. Overrides the global ``function_call_timeout_secs`` for this specific function. Defaults to None, which uses the global timeout. @@ -578,7 +594,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): Args: handler: The direct function to register. Must follow DirectFunction protocol. cancel_on_interruption: Whether to cancel this function call when an - interruption occurs. Defaults to True. + interruption occurs. When ``False`` the call is treated as + asynchronous: the LLM continues the conversation immediately + without waiting for the result, and the result is injected later + via a developer message. Defaults to True. timeout_secs: Optional per-tool timeout in seconds. Overrides the global ``function_call_timeout_secs`` for this specific function. Defaults to None, which uses the global timeout. @@ -639,6 +658,11 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): await self.broadcast_frame(FunctionCallsStartedFrame, function_calls=function_calls) + # 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. + # When False, group_id is None and each result triggers inference independently. + group_id = str(uuid.uuid4()) if self._group_parallel_tools else None + runner_items = [] for function_call in function_calls: if function_call.function_name in self._functions.keys(): @@ -658,6 +682,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): tool_call_id=function_call.tool_call_id, arguments=function_call.arguments, context=function_call.context, + group_id=group_id, ) ) @@ -726,6 +751,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): tool_call_id=runner_item.tool_call_id, arguments=runner_item.arguments, cancel_on_interruption=item.cancel_on_interruption, + group_id=runner_item.group_id, ) timeout_task: Optional[asyncio.Task] = None diff --git a/src/pipecat/utils/context/llm_context_summarization.py b/src/pipecat/utils/context/llm_context_summarization.py index 259d0bae5..afb0ecd1f 100644 --- a/src/pipecat/utils/context/llm_context_summarization.py +++ b/src/pipecat/utils/context/llm_context_summarization.py @@ -10,6 +10,7 @@ This module provides reusable functionality for automatically compressing conver context when token limits are reached, enabling efficient long-running conversations. """ +import json import warnings from dataclasses import dataclass, field from typing import TYPE_CHECKING, List, Optional @@ -381,6 +382,35 @@ class LLMContextSummarizationUtil: return total + @staticmethod + def _is_tool_message_pending(content: str) -> bool: + """Return True if a tool message content represents an unresolved call. + + A tool message is considered pending (unresolved) when its content is + the synchronous ``"IN_PROGRESS"`` sentinel or the async + ``{"type": "async_tool", "status": "started"}`` marker — both indicate + that the actual result has not yet been written back to the context. + + Args: + content: The ``content`` field of a tool-role context message. + + Returns: + True if the tool call should be treated as still in progress. + """ + if content == "IN_PROGRESS": + return True + try: + parsed = json.loads(content) + if ( + isinstance(parsed, dict) + and parsed.get("type") == "async_tool" + and parsed.get("status") == "started" + ): + return True + except (json.JSONDecodeError, ValueError): + pass + return False + @staticmethod def _get_earliest_function_call_not_resolved_in_range( messages: List[dict], start_idx: int, summary_end: int @@ -389,9 +419,13 @@ class LLMContextSummarizationUtil: Scans messages from ``start_idx`` up to (but not including) ``summary_end`` to identify tool calls whose responses either don't - exist yet or fall in the kept portion of the context (>= summary_end). + exist yet, fall in the kept portion of the context (>= summary_end), + or are still marked as ``IN_PROGRESS`` (async calls whose results have + not yet arrived). + This prevents summarizing tool call requests when their responses would - remain in the kept context as orphans, which the OpenAI API rejects. + remain in the kept context as orphans, which the OpenAI API rejects, + and avoids summarizing async function calls before their results arrive. Args: messages: List of messages to check. @@ -428,11 +462,33 @@ class LLMContextSummarizationUtil: if tool_call_id: pending_tool_calls[tool_call_id] = i - # Check for tool results + # Check for tool results — treat IN_PROGRESS and async "started" + # messages as still pending so they are not summarized away before + # their results arrive. if role == "tool": tool_call_id = msg.get("tool_call_id") if tool_call_id and tool_call_id in pending_tool_calls: - pending_tool_calls.pop(tool_call_id) + if not LLMContextSummarizationUtil._is_tool_message_pending( + msg.get("content", "") + ): + pending_tool_calls.pop(tool_call_id) + + # Check for async tool completion — a developer message with + # {"type": "async_tool", "status": "finished"} signals that the + # async result has arrived and the call is now resolved. + if role == "developer": + try: + parsed = json.loads(msg.get("content", "")) + if ( + isinstance(parsed, dict) + and parsed.get("type") == "async_tool" + and parsed.get("status") == "finished" + ): + tool_call_id = parsed.get("tool_call_id") + if tool_call_id and tool_call_id in pending_tool_calls: + pending_tool_calls.pop(tool_call_id) + except (json.JSONDecodeError, ValueError): + pass # If we have pending tool calls, return the earliest index if pending_tool_calls: From 929a0e33f4bb9ae22110d55c0872ddb5eefe6c20 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 2 Apr 2026 16:58:28 -0300 Subject: [PATCH 04/13] Fixing the automated tests. --- tests/test_context_aggregators_universal.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_context_aggregators_universal.py b/tests/test_context_aggregators_universal.py index 59aab4745..e94a68ad1 100644 --- a/tests/test_context_aggregators_universal.py +++ b/tests/test_context_aggregators_universal.py @@ -806,7 +806,7 @@ class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase): function_name="get_weather", tool_call_id="1", arguments={"location": "Los Angeles"}, - cancel_on_interruption=False, + cancel_on_interruption=True, ), SleepFrame(), FunctionCallResultFrame( @@ -838,7 +838,7 @@ class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase): function_name="get_weather", tool_call_id="1", arguments={"location": "Los Angeles"}, - cancel_on_interruption=False, + cancel_on_interruption=True, ), SleepFrame(), FunctionCallResultFrame( From bbb605acccfbb10b749917cbb41c8dc2fcd789c9 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 2 Apr 2026 16:58:42 -0300 Subject: [PATCH 05/13] Changelog entries for the fixes and improvements. --- changelog/4217.added.2.md | 1 + changelog/4217.added.md | 1 + changelog/4217.changed.md | 1 + changelog/4217.fixed.2.md | 1 + changelog/4217.fixed.md | 1 + 5 files changed, 5 insertions(+) create mode 100644 changelog/4217.added.2.md create mode 100644 changelog/4217.added.md create mode 100644 changelog/4217.changed.md create mode 100644 changelog/4217.fixed.2.md create mode 100644 changelog/4217.fixed.md diff --git a/changelog/4217.added.2.md b/changelog/4217.added.2.md new file mode 100644 index 000000000..9621d43c0 --- /dev/null +++ b/changelog/4217.added.2.md @@ -0,0 +1 @@ +- Added `group_parallel_tools` parameter to `LLMService` (default `True`). When `True`, all function calls from the same LLM response batch share a group ID and the LLM is triggered exactly once after the last call completes. Set to `False` to trigger inference independently for each function call result as it arrives. diff --git a/changelog/4217.added.md b/changelog/4217.added.md new file mode 100644 index 000000000..a14d1eb9e --- /dev/null +++ b/changelog/4217.added.md @@ -0,0 +1 @@ +- Added async function call support to `register_function()` and `register_direct_function()` via `cancel_on_interruption=False`. When set to `False`, the LLM continues the conversation immediately without waiting for the function result. The result is injected back into the context as a `developer` message once available, triggering a new LLM inference at that point. diff --git a/changelog/4217.changed.md b/changelog/4217.changed.md new file mode 100644 index 000000000..af8831c3b --- /dev/null +++ b/changelog/4217.changed.md @@ -0,0 +1 @@ +- When multiple function calls are returned in a single LLM response, the LLM is now triggered exactly once after the last call in the batch completes, rather than waiting for all function calls. diff --git a/changelog/4217.fixed.2.md b/changelog/4217.fixed.2.md new file mode 100644 index 000000000..af7bff8fa --- /dev/null +++ b/changelog/4217.fixed.2.md @@ -0,0 +1 @@ +- Fixed `BaseOutputTransport` discarding pending `UninterruptibleFrame` items (e.g. function-call context updates) when an interruption arrived. The audio task is now kept alive and only interruptible frames are drained when uninterruptible frames are present in the queue. diff --git a/changelog/4217.fixed.md b/changelog/4217.fixed.md new file mode 100644 index 000000000..234effdc0 --- /dev/null +++ b/changelog/4217.fixed.md @@ -0,0 +1 @@ +- Fixed spurious LLM inference being triggered when a function call result arrived while the user was actively speaking. The context frame is now suppressed until the user stops speaking. From 7af72eee3e0b9a02fe81619b8a17cdaeb0180a0a Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 2 Apr 2026 18:40:41 -0300 Subject: [PATCH 06/13] Creating new delayed examples for openai and anthropic. --- .../function-calling-anthropic-delayed.py | 174 ++++++++++++++++ .../function-calling-anthropic.py | 21 +- .../function-calling-openai-delayed.py | 189 ++++++++++++++++++ .../function-calling-openai.py | 15 +- 4 files changed, 371 insertions(+), 28 deletions(-) create mode 100644 examples/function-calling/function-calling-anthropic-delayed.py create mode 100644 examples/function-calling/function-calling-openai-delayed.py diff --git a/examples/function-calling/function-calling-anthropic-delayed.py b/examples/function-calling/function-calling-anthropic-delayed.py new file mode 100644 index 000000000..09cafab8e --- /dev/null +++ b/examples/function-calling/function-calling-anthropic-delayed.py @@ -0,0 +1,174 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMRunFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, + LLMUserAggregatorParams, +) +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.anthropic.llm import AnthropicLLMService +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.llm_service import FunctionCallParams +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +load_dotenv(override=True) + + +async def fetch_weather_from_api(params: FunctionCallParams): + # Simulate a long-running API call, so we can test async function calls. + await asyncio.sleep(20) + await params.result_callback({"conditions": "nice", "temperature": "75"}) + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + +# We use lambdas to defer transport parameter creation until the transport +# type is selected at runtime. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + settings=CartesiaTTSService.Settings( + voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ), + ) + + llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), + 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.", + ), + ) + + # You can also register a function_name of None to get all functions + # sent to the same callback with an additional function_name parameter. + llm.register_function( + "get_current_weather", + fetch_weather_from_api, + cancel_on_interruption=False, + timeout_secs=30, + ) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) + + context = LLMContext(tools=tools) + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, + user_aggregator, # User spoken responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + assistant_aggregator, # Assistant spoken responses and tool context + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + context.add_message( + {"role": "developer", "content": "Please introduce yourself to the user."} + ) + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/examples/function-calling/function-calling-anthropic.py b/examples/function-calling/function-calling-anthropic.py index 09cafab8e..b8bb4eac6 100644 --- a/examples/function-calling/function-calling-anthropic.py +++ b/examples/function-calling/function-calling-anthropic.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio + import os from dotenv import load_dotenv @@ -35,10 +35,9 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams load_dotenv(override=True) -async def fetch_weather_from_api(params: FunctionCallParams): - # Simulate a long-running API call, so we can test async function calls. - await asyncio.sleep(20) - await params.result_callback({"conditions": "nice", "temperature": "75"}) +async def get_weather(params: FunctionCallParams): + location = params.arguments["location"] + await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") async def fetch_restaurant_recommendation(params: FunctionCallParams): @@ -81,19 +80,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): 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.", ), ) - - # You can also register a function_name of None to get all functions - # sent to the same callback with an additional function_name parameter. - llm.register_function( - "get_current_weather", - fetch_weather_from_api, - cancel_on_interruption=False, - timeout_secs=30, - ) + llm.register_function("get_weather", get_weather) llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) weather_function = FunctionSchema( - name="get_current_weather", + name="get_weather", description="Get the current weather", properties={ "location": { diff --git a/examples/function-calling/function-calling-openai-delayed.py b/examples/function-calling/function-calling-openai-delayed.py new file mode 100644 index 000000000..b5c5b83ac --- /dev/null +++ b/examples/function-calling/function-calling-openai-delayed.py @@ -0,0 +1,189 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import ( + LLMRunFrame, + TTSSpeakFrame, +) +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, + LLMUserAggregatorParams, +) +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.openai.llm import OpenAILLMService +from pipecat.services.openai.stt import OpenAISTTService +from pipecat.services.openai.tts import OpenAITTSService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +load_dotenv(override=True) + + +async def fetch_weather_from_api(params: FunctionCallParams): + # Simulate a long-running API call, so we can test async function calls. + await asyncio.sleep(20) + await params.result_callback({"conditions": "nice", "temperature": "75"}) + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + +# We use lambdas to defer transport parameter creation until the transport +# type is selected at runtime. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = OpenAISTTService( + api_key=os.getenv("OPENAI_API_KEY"), + settings=OpenAISTTService.Settings( + model="gpt-4o-transcribe", + prompt="Expect words related weather, such as temperature and conditions. And restaurant names.", + ), + ) + + tts = OpenAITTSService( + api_key=os.getenv("OPENAI_API_KEY"), + settings=OpenAITTSService.Settings( + voice="ballad", + ), + instructions="Please speak clearly and at a moderate pace.", + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + 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.", + ), + ) + + # You can also register a function_name of None to get all functions + # sent to the same callback with an additional function_name parameter. + llm.register_function( + "get_current_weather", + fetch_weather_from_api, + cancel_on_interruption=False, + timeout_secs=30, + ) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the user's location.", + }, + }, + required=["location", "format"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) + + context = LLMContext(tools=tools) + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) + + pipeline = Pipeline( + [ + transport.input(), + stt, + user_aggregator, + llm, + tts, + transport.output(), + assistant_aggregator, + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/examples/function-calling/function-calling-openai.py b/examples/function-calling/function-calling-openai.py index b5c5b83ac..2b59d7072 100644 --- a/examples/function-calling/function-calling-openai.py +++ b/examples/function-calling/function-calling-openai.py @@ -4,7 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio import os from dotenv import load_dotenv @@ -13,10 +12,7 @@ from loguru import logger from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import ( - LLMRunFrame, - TTSSpeakFrame, -) +from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -39,8 +35,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - # Simulate a long-running API call, so we can test async function calls. - await asyncio.sleep(20) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -94,12 +88,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function( - "get_current_weather", - fetch_weather_from_api, - cancel_on_interruption=False, - timeout_secs=30, - ) + llm.register_function("get_current_weather", fetch_weather_from_api) llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) @llm.event_handler("on_function_calls_started") From eace78275287b63ab19f17650188d1d904736bec Mon Sep 17 00:00:00 2001 From: filipi87 Date: Fri, 3 Apr 2026 08:20:14 -0300 Subject: [PATCH 07/13] Renaming from async_tool to tool. --- .../processors/aggregators/llm_response_universal.py | 4 ++-- src/pipecat/utils/context/llm_context_summarization.py | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index b99b251d4..51dfcb94d 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -1061,7 +1061,7 @@ class LLMAssistantAggregator(LLMContextAggregator): self._context.add_message( { "role": "tool", - "content": json.dumps({"type": "async_tool", "status": "started"}), + "content": json.dumps({"type": "tool", "status": "started"}), "tool_call_id": frame.tool_call_id, } ) @@ -1104,7 +1104,7 @@ class LLMAssistantAggregator(LLMContextAggregator): "role": "developer", "content": json.dumps( { - "type": "async_tool", + "type": "tool", "tool_call_id": frame.tool_call_id, "status": "finished", "result": result, diff --git a/src/pipecat/utils/context/llm_context_summarization.py b/src/pipecat/utils/context/llm_context_summarization.py index afb0ecd1f..53a353992 100644 --- a/src/pipecat/utils/context/llm_context_summarization.py +++ b/src/pipecat/utils/context/llm_context_summarization.py @@ -388,7 +388,7 @@ class LLMContextSummarizationUtil: A tool message is considered pending (unresolved) when its content is the synchronous ``"IN_PROGRESS"`` sentinel or the async - ``{"type": "async_tool", "status": "started"}`` marker — both indicate + ``{"type": "tool", "status": "started"}`` marker — both indicate that the actual result has not yet been written back to the context. Args: @@ -403,7 +403,7 @@ class LLMContextSummarizationUtil: parsed = json.loads(content) if ( isinstance(parsed, dict) - and parsed.get("type") == "async_tool" + and parsed.get("type") == "tool" and parsed.get("status") == "started" ): return True @@ -474,14 +474,14 @@ class LLMContextSummarizationUtil: pending_tool_calls.pop(tool_call_id) # Check for async tool completion — a developer message with - # {"type": "async_tool", "status": "finished"} signals that the + # {"type": "tool", "status": "finished"} signals that the # async result has arrived and the call is now resolved. if role == "developer": try: parsed = json.loads(msg.get("content", "")) if ( isinstance(parsed, dict) - and parsed.get("type") == "async_tool" + and parsed.get("type") == "tool" and parsed.get("status") == "finished" ): tool_call_id = parsed.get("tool_call_id") From 42335e2ef0c61a0b6a3df9111479a735ec8cda51 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Mon, 6 Apr 2026 09:56:48 -0300 Subject: [PATCH 08/13] Renaming to async_tool and providing description. --- .../processors/aggregators/llm_response_universal.py | 11 +++++++++-- .../utils/context/llm_context_summarization.py | 8 ++++---- 2 files changed, 13 insertions(+), 6 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 51dfcb94d..841ac1727 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -1061,7 +1061,14 @@ class LLMAssistantAggregator(LLMContextAggregator): self._context.add_message( { "role": "tool", - "content": json.dumps({"type": "tool", "status": "started"}), + "content": json.dumps( + { + "type": "async_tool", + "status": "started", + "tool_call_id": frame.tool_call_id, + "description": "The tool associated with this tool_call_id is still in progress, and the result is not yet available. It will be provided in a subsequent message with the same tool_call_id.", + } + ), "tool_call_id": frame.tool_call_id, } ) @@ -1104,7 +1111,7 @@ class LLMAssistantAggregator(LLMContextAggregator): "role": "developer", "content": json.dumps( { - "type": "tool", + "type": "async_tool", "tool_call_id": frame.tool_call_id, "status": "finished", "result": result, diff --git a/src/pipecat/utils/context/llm_context_summarization.py b/src/pipecat/utils/context/llm_context_summarization.py index 53a353992..afb0ecd1f 100644 --- a/src/pipecat/utils/context/llm_context_summarization.py +++ b/src/pipecat/utils/context/llm_context_summarization.py @@ -388,7 +388,7 @@ class LLMContextSummarizationUtil: A tool message is considered pending (unresolved) when its content is the synchronous ``"IN_PROGRESS"`` sentinel or the async - ``{"type": "tool", "status": "started"}`` marker — both indicate + ``{"type": "async_tool", "status": "started"}`` marker — both indicate that the actual result has not yet been written back to the context. Args: @@ -403,7 +403,7 @@ class LLMContextSummarizationUtil: parsed = json.loads(content) if ( isinstance(parsed, dict) - and parsed.get("type") == "tool" + and parsed.get("type") == "async_tool" and parsed.get("status") == "started" ): return True @@ -474,14 +474,14 @@ class LLMContextSummarizationUtil: pending_tool_calls.pop(tool_call_id) # Check for async tool completion — a developer message with - # {"type": "tool", "status": "finished"} signals that the + # {"type": "async_tool", "status": "finished"} signals that the # async result has arrived and the call is now resolved. if role == "developer": try: parsed = json.loads(msg.get("content", "")) if ( isinstance(parsed, dict) - and parsed.get("type") == "tool" + and parsed.get("type") == "async_tool" and parsed.get("status") == "finished" ): tool_call_id = parsed.get("tool_call_id") From 9c7d5a9de2e9dd0fabec4549fb75e04fa955355a Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 7 Apr 2026 09:13:08 -0300 Subject: [PATCH 09/13] Improving changelog description to mention group_parallel_tools. --- changelog/4217.changed.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog/4217.changed.md b/changelog/4217.changed.md index af8831c3b..b53cd8317 100644 --- a/changelog/4217.changed.md +++ b/changelog/4217.changed.md @@ -1 +1 @@ -- When multiple function calls are returned in a single LLM response, the LLM is now triggered exactly once after the last call in the batch completes, rather than waiting for all function calls. +- When multiple function calls are returned in a single LLM response, by default (when `group_parallel_tools=True`) the LLM is now triggered exactly once after the last call in the batch completes, rather than waiting for all function calls. From e863293198ab015e1e6a01d083c1c69c6a158cab Mon Sep 17 00:00:00 2001 From: Filipi da Silva Fuchter Date: Tue, 7 Apr 2026 08:14:39 -0400 Subject: [PATCH 10/13] Improving docstring description. Co-authored-by: kompfner --- examples/function-calling/function-calling-anthropic-delayed.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/function-calling/function-calling-anthropic-delayed.py b/examples/function-calling/function-calling-anthropic-delayed.py index 09cafab8e..979ebcd59 100644 --- a/examples/function-calling/function-calling-anthropic-delayed.py +++ b/examples/function-calling/function-calling-anthropic-delayed.py @@ -36,7 +36,7 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - # Simulate a long-running API call, so we can test async function calls. + # Simulate a long-running API call, so we can test async function calls (cancel_on_interruption=False). await asyncio.sleep(20) await params.result_callback({"conditions": "nice", "temperature": "75"}) From aa061f7e2c920853a96db26fa667b1ea701c2953 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 7 Apr 2026 09:23:45 -0300 Subject: [PATCH 11/13] Renaming the openai and anthropic examples to async instead of delayed. --- ...g-anthropic-delayed.py => function-calling-anthropic-async.py} | 0 ...calling-openai-delayed.py => function-calling-openai-async.py} | 0 2 files changed, 0 insertions(+), 0 deletions(-) rename examples/function-calling/{function-calling-anthropic-delayed.py => function-calling-anthropic-async.py} (100%) rename examples/function-calling/{function-calling-openai-delayed.py => function-calling-openai-async.py} (100%) diff --git a/examples/function-calling/function-calling-anthropic-delayed.py b/examples/function-calling/function-calling-anthropic-async.py similarity index 100% rename from examples/function-calling/function-calling-anthropic-delayed.py rename to examples/function-calling/function-calling-anthropic-async.py diff --git a/examples/function-calling/function-calling-openai-delayed.py b/examples/function-calling/function-calling-openai-async.py similarity index 100% rename from examples/function-calling/function-calling-openai-delayed.py rename to examples/function-calling/function-calling-openai-async.py From d12a8529e2eb5f5f79ddff57bbf0f5ad4d184c9f Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 7 Apr 2026 09:28:01 -0300 Subject: [PATCH 12/13] New example for async function calls using OpenAI responses. --- ...function-calling-openai-responses-async.py | 191 ++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 examples/function-calling/function-calling-openai-responses-async.py diff --git a/examples/function-calling/function-calling-openai-responses-async.py b/examples/function-calling/function-calling-openai-responses-async.py new file mode 100644 index 000000000..ba984bc04 --- /dev/null +++ b/examples/function-calling/function-calling-openai-responses-async.py @@ -0,0 +1,191 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, + LLMUserAggregatorParams, +) +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +load_dotenv(override=True) + + +async def fetch_weather_from_api(params: FunctionCallParams): + # Simulate a long-running API call, so we can test async function calls. + await asyncio.sleep(20) + await params.result_callback({"conditions": "nice", "temperature": "75"}) + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + +# We use lambdas to defer transport parameter creation until the transport +# type is selected at runtime. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + settings=CartesiaTTSService.Settings( + voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ), + ) + + llm = OpenAIResponsesLLMService( + api_key=os.getenv("OPENAI_API_KEY"), + 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.", + ), + ) + + # You can also register a function_name of None to get all functions + # sent to the same callback with an additional function_name parameter. + llm.register_function( + "get_current_weather", + fetch_weather_from_api, + cancel_on_interruption=False, + timeout_secs=30, + ) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + @llm.event_handler("on_connection_error") + async def on_connection_error(service, error): + logger.error(f"LLM connection error: {error}") + + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + # Avoid appending this filler message to the LLM context — it would + # alter the conversation history and prevent + # OpenAIResponsesLLMService's previous_response_id optimization from + # matching, forcing a full context resend. + await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False)) + + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the user's location.", + }, + }, + required=["location", "format"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) + + context = LLMContext(tools=tools) + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) + + pipeline = Pipeline( + [ + transport.input(), + stt, + user_aggregator, + llm, + tts, + transport.output(), + assistant_aggregator, + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + context.add_message( + {"role": "developer", "content": "Please introduce yourself to the user."} + ) + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() From d8dc6bc7d073a14ed5584dd73c85f4138dccb4a4 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 7 Apr 2026 09:31:22 -0300 Subject: [PATCH 13/13] New example for async function calls using Google. --- .../function-calling-google-async.py | 250 ++++++++++++++++++ 1 file changed, 250 insertions(+) create mode 100644 examples/function-calling/function-calling-google-async.py diff --git a/examples/function-calling/function-calling-google-async.py b/examples/function-calling/function-calling-google-async.py new file mode 100644 index 000000000..1b717d4f1 --- /dev/null +++ b/examples/function-calling/function-calling-google-async.py @@ -0,0 +1,250 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame, UserImageRequestFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, + LLMUserAggregatorParams, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import ( + create_transport, + get_transport_client_id, + maybe_capture_participant_camera, +) +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.google.llm import GoogleLLMService +from pipecat.services.llm_service import FunctionCallParams +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams + +load_dotenv(override=True) + + +async def get_weather(params: FunctionCallParams): + # Simulate a long-running API call, so we can test async function calls (cancel_on_interruption=False). + await asyncio.sleep(20) + location = params.arguments["location"] + await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + +async def get_image(params: FunctionCallParams): + """Fetch the user image and push it to the LLM. + + When called, this function pushes a UserImageRequestFrame upstream to the + transport. As a result, the transport will request the user image and push a + UserImageRawFrame downstream which will be added to the context by the LLM + assistant aggregator. The result_callback will be invoked once the image is + retrieved and processed. + """ + user_id = params.arguments["user_id"] + question = params.arguments["question"] + logger.debug(f"Requesting image with user_id={user_id}, question={question}") + + # Request a user image frame and indicate that it should be added to the + # context. Also associate it to the function call. Pass the result_callback + # so it can be invoked when the image is actually retrieved. + await params.llm.push_frame( + UserImageRequestFrame( + user_id=user_id, + text=question, + append_to_context=True, + function_name=params.function_name, + tool_call_id=params.tool_call_id, + result_callback=params.result_callback, + ), + FrameDirection.UPSTREAM, + ) + + +# We use lambdas to defer transport parameter creation until the transport +# type is selected at runtime. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=True, + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + settings=CartesiaTTSService.Settings( + voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ), + ) + + system_prompt = """\ +You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions. + +Your response will be turned into speech so use only simple words and punctuation. + +You have access to three tools: get_weather, get_restaurant_recommendation, and get_image. + +You can respond to questions about the weather using the get_weather tool. + +You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \ +indicate you should use the get_image tool are: +- What do you see? +- What's in the video? +- Can you describe the video? +- Tell me about what you see. +- Tell me something interesting about what you see. +- What's happening in the video? +""" + + llm = GoogleLLMService( + api_key=os.getenv("GOOGLE_API_KEY"), + settings=GoogleLLMService.Settings( + system_instruction=system_prompt, + ), + ) + llm.register_function("get_weather", get_weather, cancel_on_interruption=False, timeout_secs=30) + llm.register_function("get_image", get_image) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + + weather_function = FunctionSchema( + name="get_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the user's location.", + }, + }, + required=["location", "format"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + get_image_function = FunctionSchema( + name="get_image", + description="Called when the user requests a description of their camera feed", + properties={ + "user_id": { + "type": "string", + "description": "The ID of the user to grab the image from", + }, + "question": { + "type": "string", + "description": "The question that the user is asking about the image", + }, + }, + required=["user_id", "question"], + ) + tools = ToolsSchema(standard_tools=[weather_function, get_image_function, restaurant_function]) + + context = LLMContext(tools=tools) + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) + + pipeline = Pipeline( + [ + transport.input(), + stt, + user_aggregator, + llm, + tts, + transport.output(), + assistant_aggregator, + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected: {client}") + + await maybe_capture_participant_camera(transport, client) + + client_id = get_transport_client_id(transport, client) + + # Kick off the conversation. + context.add_message( + { + "role": "developer", + "content": f"Please introduce yourself to the user. Use '{client_id}' as the user ID during function calls.", + } + ) + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main()