From a50a407415dd789b3fe730b2182b4ace012d02fa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 25 Apr 2025 20:04:35 -0700 Subject: [PATCH 01/12] LLMService: run function calls sequentially --- CHANGELOG.md | 7 + src/pipecat/frames/frames.py | 2 + src/pipecat/pipeline/runner.py | 4 +- .../processors/aggregators/llm_response.py | 10 +- src/pipecat/services/anthropic/llm.py | 9 + .../services/gemini_multimodal_live/gemini.py | 5 +- src/pipecat/services/google/llm_openai.py | 6 +- src/pipecat/services/llm_service.py | 190 ++++++++++++------ src/pipecat/services/openai/base_llm.py | 5 +- 9 files changed, 166 insertions(+), 72 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c4940edf..dc8a43bdd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Function calls can now be executed sequentially (in the order received in the + completion) by passing `run_in_parallel=False` when creating your LLM + service. By default, function calls run in parallel, so if the LLM completion + returns 2 or more function calls they run concurrently. In both cases, + concurrently and sequentailly, a new LLM completion will run when the last + function call finishes. + - Added OpenTelemetry tracing for `GeminiMultimodalLiveLLMService` and `OpenAIRealtimeBetaLLMService`. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 45c1fc6c2..0c2cec5bd 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -675,6 +675,7 @@ class FunctionCallInProgressFrame(SystemFrame): tool_call_id: str arguments: Any cancel_on_interruption: bool = False + run_concurrently: bool = False @dataclass @@ -701,6 +702,7 @@ class FunctionCallResultFrame(SystemFrame): tool_call_id: str arguments: Any result: Any + run_llm: Optional[bool] = None properties: Optional[FunctionCallResultProperties] = None diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index 7ac07064f..23c43c06e 100644 --- a/src/pipecat/pipeline/runner.py +++ b/src/pipecat/pipeline/runner.py @@ -59,7 +59,7 @@ class PipelineRunner(BaseObject): await asyncio.gather(*[t.stop_when_done() for t in self._tasks.values()]) async def cancel(self): - logger.debug(f"Canceling runner {self}") + logger.debug(f"Cancelling runner {self}") await asyncio.gather(*[t.cancel() for t in self._tasks.values()]) def _setup_sigint(self): @@ -72,7 +72,7 @@ class PipelineRunner(BaseObject): self._sig_task = asyncio.create_task(self._sig_cancel()) async def _sig_cancel(self): - logger.warning(f"Interruption detected. Canceling runner {self}") + logger.warning(f"Interruption detected. Cancelling runner {self}") await self.cancel() def _gc_collect(self): diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index be9c6f77f..f4a7dcb9c 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -591,6 +591,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): ) return + in_progress = self._function_calls_in_progress[frame.tool_call_id] + del self._function_calls_in_progress[frame.tool_call_id] properties = frame.properties @@ -600,12 +602,14 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): # Run inference if the function call result requires it. if frame.result: run_llm = False - if properties and properties.run_llm is not None: # If the tool call result has a run_llm property, use it run_llm = properties.run_llm - else: - # Default behavior is to run the LLM if there are no function calls in progress + elif frame.run_llm is not None: + # If the frame is indicating we should run the LLM, do it. + run_llm = frame.run_llm + elif in_progress.run_concurrently: + # If this was a parallel function call and there are no pending function call, run the LLM. run_llm = not bool(self._function_calls_in_progress) if run_llm: diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 2c4e9078d..5e0df1dd7 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -202,6 +202,12 @@ class AnthropicLLMService(LLMService): tool_use_block = None json_accumulator = "" + total_func_calls = 0 + async for event in response: + if event.type == "content_block_start" and event.content_block.type == "tool_use": + total_func_calls += 1 + + current_func_call = 0 async for event in response: # logger.debug(f"Anthropic LLM event: {event}") @@ -226,12 +232,15 @@ class AnthropicLLMService(LLMService): and event.delta.stop_reason == "tool_use" ): if tool_use_block: + run_llm = current_func_call == total_func_calls - 1 await self.call_function( context=context, tool_call_id=tool_use_block.id, function_name=tool_use_block.name, arguments=json.loads(json_accumulator) if json_accumulator else dict(), + run_llm=run_llm, ) + current_func_call += 1 # Calculate usage. Do this here in its own if statement, because there may be usage # data embedded in messages that we do other processing for, above. diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 6dd93720b..ee3dc6245 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -891,12 +891,15 @@ class GeminiMultimodalLiveLLMService(LLMService): return if not self._context: logger.error("Function calls are not supported without a context object.") - for call in function_calls: + total_items = len(function_calls) + for index, call in enumerate(function_calls): + run_llm = index == total_items - 1 await self.call_function( context=self._context, tool_call_id=call.id, function_name=call.name, arguments=call.args, + run_llm=run_llm, ) @traced_gemini_live(operation="llm_response") diff --git a/src/pipecat/services/google/llm_openai.py b/src/pipecat/services/google/llm_openai.py index 94b99072a..764572d68 100644 --- a/src/pipecat/services/google/llm_openai.py +++ b/src/pipecat/services/google/llm_openai.py @@ -112,8 +112,10 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): logger.debug( f"Function list: {functions_list}, Arguments list: {arguments_list}, Tool ID list: {tool_id_list}" ) + + total_func_calls = len(functions_list) for index, (function_name, arguments, tool_id) in enumerate( - zip(functions_list, arguments_list, tool_id_list), start=1 + zip(functions_list, arguments_list, tool_id_list) ): if function_name == "": # TODO: Remove the _process_context method once Google resolves the bug @@ -121,8 +123,8 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): # which currently results in an empty function name(''). continue if self.has_function(function_name): - run_llm = False arguments = json.loads(arguments) + run_llm = index == total_func_calls - 1 await self.call_function( context=context, function_name=function_name, diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 21b62325d..767208026 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -7,18 +7,21 @@ import asyncio import inspect from dataclasses import dataclass -from typing import Any, Awaitable, Callable, Mapping, Optional, Protocol, Set, Tuple, Type +from typing import Any, Awaitable, Callable, Dict, Mapping, Optional, Protocol, Type from loguru import logger from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter from pipecat.frames.frames import ( + CancelFrame, + EndFrame, Frame, FunctionCallCancelFrame, FunctionCallInProgressFrame, FunctionCallResultFrame, FunctionCallResultProperties, + StartFrame, StartInterruptionFrame, UserImageRequestFrame, ) @@ -42,19 +45,43 @@ class FunctionCallResultCallback(Protocol): @dataclass -class FunctionCallEntry: - """Represents an internal entry for a function call. +class FunctionCallItem: + """Represents an entry of our function call registry. Attributes: function_name (Optional[str]): The name of the function. handler (FunctionCallHandler): The handler for processing function call parameters. cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption. + run_concurrently (bool): Flag to indicate if this function call should run concurrently or not. """ function_name: Optional[str] handler: FunctionCallHandler cancel_on_interruption: bool + run_concurrently: bool + + +@dataclass +class FunctionCallRunnerItem: + """Represents a function call entry for our function call runner. The runner + executes function calls in order. + + Attributes: + registry_name (Optional[str]): The function call name registration (could be None). + function_name (str): The name of the function. + tool_call_id (str): A unique identifier for the function call. + arguments (Mapping[str, Any]): The arguments for the function. + context (OpenAILLMContext): The LLM context. + + """ + + registry_name: Optional[str] + function_name: str + tool_call_id: str + arguments: Mapping[str, Any] + context: OpenAILLMContext + run_llm: bool @dataclass @@ -88,10 +115,10 @@ class LLMService(AIService): def __init__(self, **kwargs): super().__init__(**kwargs) - self._functions = {} self._start_callbacks = {} self._adapter = self.adapter_class() - self._function_call_tasks: Set[Tuple[asyncio.Task, str, str]] = set() + self._functions: Dict[Optional[str], FunctionCallItem] = {} + self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunnerItem] = {} self._register_event_handler("on_completion_timeout") @@ -107,13 +134,28 @@ class LLMService(AIService): ) -> Any: pass + async def start(self, frame: StartFrame): + await super().start(frame) + self._function_call_runner_queue = asyncio.Queue() + self._function_call_runner_task = self.create_task(self._function_call_runner_handler()) + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._cancel_function_call(None) + await self.cancel_task(self._function_call_runner_task) + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._cancel_function_call(None) + await self.cancel_task(self._function_call_runner_task) + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, StartInterruptionFrame): await self._handle_interruptions(frame) - async def _handle_interruptions(self, frame: StartInterruptionFrame): + async def _handle_interruptions(self, _: StartInterruptionFrame): for function_name, entry in self._functions.items(): if entry.cancel_on_interruption: await self._cancel_function_call(function_name) @@ -125,13 +167,15 @@ class LLMService(AIService): start_callback=None, *, cancel_on_interruption: bool = False, + run_concurrently: bool = False, ): # Registering a function with the function_name set to None will run # that handler for all functions - self._functions[function_name] = FunctionCallEntry( + self._functions[function_name] = FunctionCallItem( function_name=function_name, handler=handler, cancel_on_interruption=cancel_on_interruption, + run_concurrently=run_concurrently, ) # Start callbacks are now deprecated. @@ -166,16 +210,29 @@ class LLMService(AIService): arguments: Mapping[str, Any], run_llm: bool = True, ): - if not function_name in self._functions.keys() and not None in self._functions.keys(): + if function_name in self._functions.keys(): + item = self._functions[function_name] + registry_name = function_name + elif None in self._functions.keys(): + item = self._functions[None] + registry_name = None + else: return - task = self.create_task( - self._run_function_call(context, tool_call_id, function_name, arguments, run_llm) + runner_item = FunctionCallRunnerItem( + registry_name=registry_name, + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + context=context, + run_llm=run_llm, ) - - self._function_call_tasks.add((task, tool_call_id, function_name)) - - task.add_done_callback(self._function_call_task_finished) + if item.run_concurrently: + task = self.create_task(self._run_function_call(runner_item)) + self._function_call_tasks[task] = runner_item + task.add_done_callback(self._function_call_task_finished) + else: + await self._function_call_runner_queue.put(runner_item) async def call_start_function(self, context: OpenAILLMContext, function_name: str): if function_name in self._start_callbacks.keys(): @@ -203,43 +260,45 @@ class LLMService(AIService): FrameDirection.UPSTREAM, ) - async def _run_function_call( - self, - context: OpenAILLMContext, - tool_call_id: str, - function_name: str, - arguments: Mapping[str, Any], - run_llm: bool = True, - ): - if function_name in self._functions.keys(): - entry = self._functions[function_name] + async def _function_call_runner_handler(self): + while True: + runner_item = await self._function_call_runner_queue.get() + task = self.create_task(self._run_function_call(runner_item)) + await self.wait_for_task(task) + + async def _run_function_call(self, runner_item: FunctionCallRunnerItem): + if runner_item.function_name in self._functions.keys(): + item = self._functions[runner_item.function_name] elif None in self._functions.keys(): - entry = self._functions[None] + item = self._functions[None] else: return logger.debug( - f"{self} Calling function [{function_name}:{tool_call_id}] with arguments {arguments}" + f"{self} Calling function [{runner_item.function_name}:{runner_item.tool_call_id}] with arguments {runner_item.arguments}" ) # NOTE(aleix): This needs to be removed after we remove the deprecation. - await self.call_start_function(context, function_name) + await self.call_start_function(runner_item.context, runner_item.function_name) - # Push a SystemFrame downstream. This frame will let our assistant context aggregator - # know that we are in the middle of a function call. Some contexts/aggregators may - # not need this. But some definitely do (Anthropic, for example). - # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter. + # Push a function call in-progress downstream. This frame will let our + # assistant context aggregator know that we are in the middle of a + # function call. Some contexts/aggregators may not need this. But some + # definitely do (Anthropic, for example). Also push it upstream for use + # by other processors, like STTMuteFilter. progress_frame_downstream = FunctionCallInProgressFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - cancel_on_interruption=entry.cancel_on_interruption, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, + cancel_on_interruption=item.cancel_on_interruption, + run_concurrently=item.run_concurrently, ) progress_frame_upstream = FunctionCallInProgressFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - cancel_on_interruption=entry.cancel_on_interruption, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, + cancel_on_interruption=item.cancel_on_interruption, + run_concurrently=item.run_concurrently, ) # Push frame both downstream and upstream @@ -251,24 +310,26 @@ class LLMService(AIService): result: Any, *, properties: Optional[FunctionCallResultProperties] = None ): result_frame_downstream = FunctionCallResultFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, result=result, + run_llm=runner_item.run_llm, properties=properties, ) result_frame_upstream = FunctionCallResultFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, result=result, + run_llm=runner_item.run_llm, properties=properties, ) await self.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM) await self.push_frame(result_frame_upstream, FrameDirection.UPSTREAM) - signature = inspect.signature(entry.handler) + signature = inspect.signature(item.handler) if len(signature.parameters) > 1: import warnings @@ -279,24 +340,32 @@ class LLMService(AIService): DeprecationWarning, ) - await entry.handler( - function_name, tool_call_id, arguments, self, context, function_call_result_callback + await item.handler( + runner_item.function_name, + runner_item.tool_call_id, + runner_item.arguments, + self, + runner_item.context, + function_call_result_callback, ) else: params = FunctionCallParams( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, + function_name=runner_item.function_name, + tool_call_id=runner_item.tool_call_id, + arguments=runner_item.arguments, llm=self, - context=context, + context=runner_item.context, result_callback=function_call_result_callback, ) - await entry.handler(params) + await item.handler(params) - async def _cancel_function_call(self, function_name: str): + async def _cancel_function_call(self, function_name: Optional[str]): cancelled_tasks = set() - for task, tool_call_id, name in self._function_call_tasks: - if name == function_name: + for task, runner_item in self._function_call_tasks.items(): + if runner_item.registry_name == function_name: + name = runner_item.function_name + tool_call_id = runner_item.tool_call_id + # We remove the callback because we are going to cancel the task # now, otherwise we will be removing it from the set while we # are iterating. @@ -306,9 +375,7 @@ class LLMService(AIService): await self.cancel_task(task) - frame = FunctionCallCancelFrame( - function_name=function_name, tool_call_id=tool_call_id - ) + frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id) await self.push_frame(frame) logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled") @@ -320,9 +387,8 @@ class LLMService(AIService): self._function_call_task_finished(task) def _function_call_task_finished(self, task: asyncio.Task): - tuple_to_remove = next((t for t in self._function_call_tasks if t[0] == task), None) - if tuple_to_remove: - self._function_call_tasks.discard(tuple_to_remove) + if task in self._function_call_tasks: + del self._function_call_tasks[task] # The task is finished so this should exit immediately. We need to # do this because otherwise the task manager would report a dangling # task if we don't remove it. diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index e90f87f3c..bb6f2ce8c 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -260,11 +260,12 @@ class BaseOpenAILLMService(LLMService): arguments_list.append(arguments) tool_id_list.append(tool_call_id) + total_func_calls = len(functions_list) for index, (function_name, arguments, tool_id) in enumerate( - zip(functions_list, arguments_list, tool_id_list), start=1 + zip(functions_list, arguments_list, tool_id_list) ): if self.has_function(function_name): - run_llm = False + run_llm = index == total_func_calls - 1 arguments = json.loads(arguments) await self.call_function( context=context, From 52569bcdb2acc19f393945b34b0fa61b10e050b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 28 Apr 2025 11:49:29 -0700 Subject: [PATCH 02/12] LLMService: don't allow running functions concurrently for now --- src/pipecat/frames/frames.py | 1 - .../processors/aggregators/llm_response.py | 6 +- src/pipecat/services/llm_service.py | 102 ++++++++---------- 3 files changed, 46 insertions(+), 63 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 0c2cec5bd..bb6143c61 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -675,7 +675,6 @@ class FunctionCallInProgressFrame(SystemFrame): tool_call_id: str arguments: Any cancel_on_interruption: bool = False - run_concurrently: bool = False @dataclass diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index f4a7dcb9c..4574d624d 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -603,13 +603,13 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): if frame.result: run_llm = False if properties and properties.run_llm is not None: - # If the tool call result has a run_llm property, use it + # If the tool call result has a run_llm property, use it. run_llm = properties.run_llm elif frame.run_llm is not None: # If the frame is indicating we should run the LLM, do it. run_llm = frame.run_llm - elif in_progress.run_concurrently: - # If this was a parallel function call and there are no pending function call, run the LLM. + else: + # If this is the last function call in progress, run the LLM. run_llm = not bool(self._function_calls_in_progress) if run_llm: diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 767208026..915f69a47 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -52,14 +52,12 @@ class FunctionCallItem: function_name (Optional[str]): The name of the function. handler (FunctionCallHandler): The handler for processing function call parameters. cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption. - run_concurrently (bool): Flag to indicate if this function call should run concurrently or not. """ function_name: Optional[str] handler: FunctionCallHandler cancel_on_interruption: bool - run_concurrently: bool @dataclass @@ -76,7 +74,7 @@ class FunctionCallRunnerItem: """ - registry_name: Optional[str] + registry_item: FunctionCallItem function_name: str tool_call_id: str arguments: Mapping[str, Any] @@ -118,7 +116,7 @@ class LLMService(AIService): self._start_callbacks = {} self._adapter = self.adapter_class() self._functions: Dict[Optional[str], FunctionCallItem] = {} - self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunnerItem] = {} + self._function_call_runner_task: Optional[asyncio.Task] = None self._register_event_handler("on_completion_timeout") @@ -136,18 +134,17 @@ class LLMService(AIService): async def start(self, frame: StartFrame): await super().start(frame) - self._function_call_runner_queue = asyncio.Queue() - self._function_call_runner_task = self.create_task(self._function_call_runner_handler()) + await self._create_runner_task() async def stop(self, frame: EndFrame): await super().stop(frame) - await self._cancel_function_call(None) - await self.cancel_task(self._function_call_runner_task) + await self._cancel_function_call() + await self._cancel_runner_task() async def cancel(self, frame: CancelFrame): await super().cancel(frame) - await self._cancel_function_call(None) - await self.cancel_task(self._function_call_runner_task) + await self._cancel_function_call() + await self._cancel_runner_task() async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -156,9 +153,9 @@ class LLMService(AIService): await self._handle_interruptions(frame) async def _handle_interruptions(self, _: StartInterruptionFrame): - for function_name, entry in self._functions.items(): - if entry.cancel_on_interruption: - await self._cancel_function_call(function_name) + await self._cancel_function_call() + await self._cancel_runner_task() + await self._create_runner_task() def register_function( self, @@ -167,7 +164,6 @@ class LLMService(AIService): start_callback=None, *, cancel_on_interruption: bool = False, - run_concurrently: bool = False, ): # Registering a function with the function_name set to None will run # that handler for all functions @@ -175,7 +171,6 @@ class LLMService(AIService): function_name=function_name, handler=handler, cancel_on_interruption=cancel_on_interruption, - run_concurrently=run_concurrently, ) # Start callbacks are now deprecated. @@ -212,27 +207,21 @@ class LLMService(AIService): ): if function_name in self._functions.keys(): item = self._functions[function_name] - registry_name = function_name elif None in self._functions.keys(): item = self._functions[None] - registry_name = None else: return runner_item = FunctionCallRunnerItem( - registry_name=registry_name, + registry_item=item, function_name=function_name, tool_call_id=tool_call_id, arguments=arguments, context=context, run_llm=run_llm, ) - if item.run_concurrently: - task = self.create_task(self._run_function_call(runner_item)) - self._function_call_tasks[task] = runner_item - task.add_done_callback(self._function_call_task_finished) - else: - await self._function_call_runner_queue.put(runner_item) + + await self._function_call_runner_queue.put(runner_item) async def call_start_function(self, context: OpenAILLMContext, function_name: str): if function_name in self._start_callbacks.keys(): @@ -260,11 +249,25 @@ class LLMService(AIService): FrameDirection.UPSTREAM, ) + async def _create_runner_task(self): + if not self._function_call_runner_task: + self._current_runner: Optional[FunctionCallRunnerItem] = None + self._current_task: Optional[asyncio.Task] = None + self._function_call_runner_queue = asyncio.Queue() + self._function_call_runner_task = self.create_task(self._function_call_runner_handler()) + + async def _cancel_runner_task(self): + if self._function_call_runner_task: + await self.cancel_task(self._function_call_runner_task) + self._function_call_runner_task = None + async def _function_call_runner_handler(self): while True: - runner_item = await self._function_call_runner_queue.get() - task = self.create_task(self._run_function_call(runner_item)) - await self.wait_for_task(task) + self._current_runner = await self._function_call_runner_queue.get() + self._current_task = self.create_task(self._run_function_call(self._current_runner)) + await self.wait_for_task(self._current_task) + self._current_runner = None + self._current_task = None async def _run_function_call(self, runner_item: FunctionCallRunnerItem): if runner_item.function_name in self._functions.keys(): @@ -291,14 +294,12 @@ class LLMService(AIService): tool_call_id=runner_item.tool_call_id, arguments=runner_item.arguments, cancel_on_interruption=item.cancel_on_interruption, - run_concurrently=item.run_concurrently, ) progress_frame_upstream = FunctionCallInProgressFrame( function_name=runner_item.function_name, tool_call_id=runner_item.tool_call_id, arguments=runner_item.arguments, cancel_on_interruption=item.cancel_on_interruption, - run_concurrently=item.run_concurrently, ) # Push frame both downstream and upstream @@ -359,37 +360,20 @@ class LLMService(AIService): ) await item.handler(params) - async def _cancel_function_call(self, function_name: Optional[str]): - cancelled_tasks = set() - for task, runner_item in self._function_call_tasks.items(): - if runner_item.registry_name == function_name: - name = runner_item.function_name - tool_call_id = runner_item.tool_call_id + async def _cancel_function_call(self): + if ( + self._current_runner + and self._current_task + and self._current_runner.registry_item.cancel_on_interruption + ): + name = self._current_runner.function_name + tool_call_id = self._current_runner.tool_call_id - # We remove the callback because we are going to cancel the task - # now, otherwise we will be removing it from the set while we - # are iterating. - task.remove_done_callback(self._function_call_task_finished) + logger.debug(f"{self} Cancelling function call [{name}:{tool_call_id}]...") - logger.debug(f"{self} Cancelling function call [{name}:{tool_call_id}]...") + await self.cancel_task(self._current_task) - await self.cancel_task(task) + frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id) + await self.push_frame(frame) - frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id) - await self.push_frame(frame) - - logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled") - - cancelled_tasks.add(task) - - # Remove all cancelled tasks from our set. - for task in cancelled_tasks: - self._function_call_task_finished(task) - - def _function_call_task_finished(self, task: asyncio.Task): - if task in self._function_call_tasks: - del self._function_call_tasks[task] - # The task is finished so this should exit immediately. We need to - # do this because otherwise the task manager would report a dangling - # task if we don't remove it. - asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop()) + logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled") From 1eb50ad88fe63868ab73a16166295099df39f772 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 28 Apr 2025 13:43:28 -0700 Subject: [PATCH 03/12] LLMService: pass LLM function calls all at once --- src/pipecat/services/anthropic/llm.py | 29 +++---- src/pipecat/services/aws/llm.py | 16 ++-- .../services/gemini_multimodal_live/gemini.py | 20 +++-- src/pipecat/services/google/llm.py | 17 ++-- src/pipecat/services/google/llm_openai.py | 28 +++--- src/pipecat/services/llm_service.py | 86 +++++++++++-------- src/pipecat/services/openai/base_llm.py | 30 +++---- .../services/openai_realtime_beta/openai.py | 35 +++----- 8 files changed, 134 insertions(+), 127 deletions(-) diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 5e0df1dd7..49b4f2f6a 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -45,7 +45,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallLLM, LLMService from pipecat.utils.tracing.service_decorators import traced_llm try: @@ -202,15 +202,8 @@ class AnthropicLLMService(LLMService): tool_use_block = None json_accumulator = "" - total_func_calls = 0 + function_calls = [] async for event in response: - if event.type == "content_block_start" and event.content_block.type == "tool_use": - total_func_calls += 1 - - current_func_call = 0 - async for event in response: - # logger.debug(f"Anthropic LLM event: {event}") - # Aggregate streaming content, create frames, trigger events if event.type == "content_block_delta": @@ -232,15 +225,15 @@ class AnthropicLLMService(LLMService): and event.delta.stop_reason == "tool_use" ): if tool_use_block: - run_llm = current_func_call == total_func_calls - 1 - await self.call_function( - context=context, - tool_call_id=tool_use_block.id, - function_name=tool_use_block.name, - arguments=json.loads(json_accumulator) if json_accumulator else dict(), - run_llm=run_llm, + args = json.loads(json_accumulator) if json_accumulator else {} + function_calls.append( + FunctionCallLLM( + context=context, + tool_call_id=tool_use_block.id, + function_name=tool_use_block.name, + arguments=args, + ) ) - current_func_call += 1 # Calculate usage. Do this here in its own if statement, because there may be usage # data embedded in messages that we do other processing for, above. @@ -286,6 +279,8 @@ class AnthropicLLMService(LLMService): if total_input_tokens >= 1024: context.turns_above_cache_threshold += 1 + await self.run_function_calls(function_calls) + except asyncio.CancelledError: # If we're interrupted, we won't get a complete usage report. So set our flag to use the # token estimate. The reraise the exception so all the processors running in this task diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index a90620b20..dcec91463 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -21,6 +21,7 @@ from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter from pipecat.frames.frames import ( Frame, FunctionCallCancelFrame, + FunctionCallFromLLM, FunctionCallInProgressFrame, FunctionCallResultFrame, LLMFullResponseEndFrame, @@ -708,6 +709,7 @@ class AWSBedrockLLMService(LLMService): tool_use_block = None json_accumulator = "" + function_calls = [] for event in response["stream"]: # Handle text content if "contentBlockDelta" in event: @@ -740,11 +742,13 @@ class AWSBedrockLLMService(LLMService): # Only call function if it's not the no_operation tool if not using_noop_tool: - await self.call_function( - context=context, - tool_call_id=tool_use_block["id"], - function_name=tool_use_block["name"], - arguments=arguments, + function_calls.append( + FunctionCallFromLLM( + context=context, + tool_call_id=tool_use_block["id"], + function_name=tool_use_block["name"], + arguments=arguments, + ) ) else: logger.debug("Ignoring no_operation tool call") @@ -758,7 +762,7 @@ class AWSBedrockLLMService(LLMService): completion_tokens += usage.get("outputTokens", 0) cache_read_input_tokens += usage.get("cacheReadInputTokens", 0) cache_creation_input_tokens += usage.get("cacheWriteInputTokens", 0) - + await self.run_function_calls(function_calls) except asyncio.CancelledError: # If we're interrupted, we won't get a complete usage report. So set our flag to use the # token estimate. The reraise the exception so all the processors running in this task diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index ee3dc6245..3096184d2 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -52,7 +52,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallLLM, LLMService from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, OpenAIUserContextAggregator, @@ -891,16 +891,18 @@ class GeminiMultimodalLiveLLMService(LLMService): return if not self._context: logger.error("Function calls are not supported without a context object.") - total_items = len(function_calls) - for index, call in enumerate(function_calls): - run_llm = index == total_items - 1 - await self.call_function( + + function_calls_llm = [ + FunctionCallLLM( context=self._context, - tool_call_id=call.id, - function_name=call.name, - arguments=call.args, - run_llm=run_llm, + tool_call_id=f.id, + function_name=f.name, + arguments=f.args, ) + for f in function_calls + ] + + await self.run_function_calls(function_calls_llm) @traced_gemini_live(operation="llm_response") async def _handle_evt_turn_complete(self, evt): diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 5efbbe8c0..68e6f2406 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -42,7 +42,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.google.frames import LLMSearchResponseFrame -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallLLM, LLMService from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, OpenAIUserContextAggregator, @@ -557,6 +557,7 @@ class GoogleLLMService(LLMService): ) await self.stop_ttfb_metrics() + function_calls = [] async for chunk in response: if chunk.usage_metadata: prompt_tokens += chunk.usage_metadata.prompt_token_count or 0 @@ -576,11 +577,13 @@ class GoogleLLMService(LLMService): function_call = part.function_call id = function_call.id or str(uuid.uuid4()) logger.debug(f"Function call: {function_call.name}:{id}") - await self.call_function( - context=context, - tool_call_id=id, - function_name=function_call.name, - arguments=function_call.args or {}, + function_calls.append( + FunctionCallLLM( + context=context, + tool_call_id=id, + function_name=function_call.name, + arguments=function_call.args or {}, + ) ) if ( @@ -621,6 +624,8 @@ class GoogleLLMService(LLMService): "rendered_content": rendered_content, "origins": origins, } + + await self.run_function_calls(function_calls) except DeadlineExceeded: await self._call_event_handler("on_completion_timeout") except Exception as e: diff --git a/src/pipecat/services/google/llm_openai.py b/src/pipecat/services/google/llm_openai.py index 764572d68..db395705f 100644 --- a/src/pipecat/services/google/llm_openai.py +++ b/src/pipecat/services/google/llm_openai.py @@ -10,6 +10,8 @@ import os from openai import AsyncStream from openai.types.chat import ChatCompletionChunk +from pipecat.services.llm_service import FunctionCallLLM + # Suppress gRPC fork warnings os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" @@ -18,7 +20,6 @@ from loguru import logger from pipecat.frames.frames import LLMTextFrame from pipecat.metrics.metrics import LLMTokenUsage from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.services.openai.base_llm import OpenAIUnhandledFunctionException from pipecat.services.openai.llm import OpenAILLMService @@ -113,26 +114,25 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): f"Function list: {functions_list}, Arguments list: {arguments_list}, Tool ID list: {tool_id_list}" ) - total_func_calls = len(functions_list) - for index, (function_name, arguments, tool_id) in enumerate( - zip(functions_list, arguments_list, tool_id_list) + function_calls = [] + for function_name, arguments, tool_id in zip( + functions_list, arguments_list, tool_id_list ): if function_name == "": # TODO: Remove the _process_context method once Google resolves the bug # where the index is incorrectly set to None instead of returning the actual index, # which currently results in an empty function name(''). continue - if self.has_function(function_name): - arguments = json.loads(arguments) - run_llm = index == total_func_calls - 1 - await self.call_function( + + arguments = json.loads(arguments) + + function_calls.append( + FunctionCallLLM( context=context, + tool_call_id=tool_id, function_name=function_name, arguments=arguments, - tool_call_id=tool_id, - run_llm=run_llm, - ) - else: - raise OpenAIUnhandledFunctionException( - f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function." ) + ) + + await self.run_function_calls(function_calls) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 915f69a47..69eabc686 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -7,7 +7,7 @@ import asyncio import inspect from dataclasses import dataclass -from typing import Any, Awaitable, Callable, Dict, Mapping, Optional, Protocol, Type +from typing import Any, Awaitable, Callable, Dict, Mapping, Optional, Protocol, Sequence, Type from loguru import logger @@ -45,7 +45,7 @@ class FunctionCallResultCallback(Protocol): @dataclass -class FunctionCallItem: +class FunctionCallRegistryItem: """Represents an entry of our function call registry. Attributes: @@ -61,9 +61,27 @@ class FunctionCallItem: @dataclass -class FunctionCallRunnerItem: - """Represents a function call entry for our function call runner. The runner - executes function calls in order. +class FunctionCallLLM: + """Represents a function call returned by the LLM to be registered for execution. + + Attributes: + function_name (str): The name of the function. + tool_call_id (str): A unique identifier for the function call. + arguments (Mapping[str, Any]): The arguments for the function. + context (OpenAILLMContext): The LLM context. + + """ + + function_name: str + tool_call_id: str + arguments: Mapping[str, Any] + context: OpenAILLMContext + + +@dataclass +class FunctionCallRunner: + """Represents an internal function call entry to our function call + runner. The runner executes function calls in order. Attributes: registry_name (Optional[str]): The function call name registration (could be None). @@ -74,7 +92,7 @@ class FunctionCallRunnerItem: """ - registry_item: FunctionCallItem + registry_item: FunctionCallRegistryItem function_name: str tool_call_id: str arguments: Mapping[str, Any] @@ -115,7 +133,7 @@ class LLMService(AIService): super().__init__(**kwargs) self._start_callbacks = {} self._adapter = self.adapter_class() - self._functions: Dict[Optional[str], FunctionCallItem] = {} + self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} self._function_call_runner_task: Optional[asyncio.Task] = None self._register_event_handler("on_completion_timeout") @@ -167,7 +185,7 @@ class LLMService(AIService): ): # Registering a function with the function_name set to None will run # that handler for all functions - self._functions[function_name] = FunctionCallItem( + self._functions[function_name] = FunctionCallRegistryItem( function_name=function_name, handler=handler, cancel_on_interruption=cancel_on_interruption, @@ -196,32 +214,32 @@ class LLMService(AIService): return True return function_name in self._functions.keys() - async def call_function( - self, - *, - context: OpenAILLMContext, - tool_call_id: str, - function_name: str, - arguments: Mapping[str, Any], - run_llm: bool = True, - ): - if function_name in self._functions.keys(): - item = self._functions[function_name] - elif None in self._functions.keys(): - item = self._functions[None] - else: - return + async def run_function_calls(self, function_calls: Sequence[FunctionCallLLM]): + total_function_calls = len(function_calls) + for index, function_call in enumerate(function_calls): + if function_call.function_name in self._functions.keys(): + item = self._functions[function_call.function_name] + elif None in self._functions.keys(): + item = self._functions[None] + else: + logger.warning( + f"{self} is calling '{function_call.function_name}', but it's not registered." + ) + continue - runner_item = FunctionCallRunnerItem( - registry_item=item, - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - context=context, - run_llm=run_llm, - ) + # Run inference on the last function call. + run_llm = index == total_function_calls - 1 - await self._function_call_runner_queue.put(runner_item) + runner_item = FunctionCallRunner( + registry_item=item, + function_name=function_call.function_name, + tool_call_id=function_call.tool_call_id, + arguments=function_call.arguments, + context=function_call.context, + run_llm=run_llm, + ) + + await self._function_call_runner_queue.put(runner_item) async def call_start_function(self, context: OpenAILLMContext, function_name: str): if function_name in self._start_callbacks.keys(): @@ -251,7 +269,7 @@ class LLMService(AIService): async def _create_runner_task(self): if not self._function_call_runner_task: - self._current_runner: Optional[FunctionCallRunnerItem] = None + self._current_runner: Optional[FunctionCallRunner] = None self._current_task: Optional[asyncio.Task] = None self._function_call_runner_queue = asyncio.Queue() self._function_call_runner_task = self.create_task(self._function_call_runner_handler()) @@ -269,7 +287,7 @@ class LLMService(AIService): self._current_runner = None self._current_task = None - async def _run_function_call(self, runner_item: FunctionCallRunnerItem): + async def _run_function_call(self, runner_item: FunctionCallRunner): if runner_item.function_name in self._functions.keys(): item = self._functions[runner_item.function_name] elif None in self._functions.keys(): diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index bb6f2ce8c..7f3303a2e 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -34,14 +34,10 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallLLM, LLMService from pipecat.utils.tracing.service_decorators import traced_llm -class OpenAIUnhandledFunctionException(Exception): - pass - - class BaseOpenAILLMService(LLMService): """This is the base for all services that use the AsyncOpenAI client. @@ -260,24 +256,22 @@ class BaseOpenAILLMService(LLMService): arguments_list.append(arguments) tool_id_list.append(tool_call_id) - total_func_calls = len(functions_list) - for index, (function_name, arguments, tool_id) in enumerate( - zip(functions_list, arguments_list, tool_id_list) + function_calls = [] + + for function_name, arguments, tool_id in zip( + functions_list, arguments_list, tool_id_list ): - if self.has_function(function_name): - run_llm = index == total_func_calls - 1 - arguments = json.loads(arguments) - await self.call_function( + arguments = json.loads(arguments) + function_calls.append( + FunctionCallLLM( context=context, + tool_call_id=tool_id, function_name=function_name, arguments=arguments, - tool_call_id=tool_id, - run_llm=run_llm, - ) - else: - raise OpenAIUnhandledFunctionException( - f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function." ) + ) + + await self.run_function_calls(function_calls) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 9957cb134..343c61415 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -48,7 +48,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallLLM, LLMService from pipecat.services.openai.llm import OpenAIContextAggregatorPair from pipecat.transcriptions.language import Language from pipecat.utils.time import time_now_iso8601 @@ -78,10 +78,6 @@ class CurrentAudioResponse: total_size: int = 0 -class OpenAIUnhandledFunctionException(Exception): - pass - - class OpenAIRealtimeBetaLLMService(LLMService): # Overriding the default adapter to use the OpenAIRealtimeLLMAdapter one. adapter_class = OpenAIRealtimeLLMAdapter @@ -587,25 +583,18 @@ class OpenAIRealtimeBetaLLMService(LLMService): await self._handle_function_call_items(function_calls) async def _handle_function_call_items(self, items): - total_items = len(items) - for index, item in enumerate(items): - function_name = item.name - tool_id = item.call_id - arguments = json.loads(item.arguments) - if self.has_function(function_name): - run_llm = index == total_items - 1 - if function_name in self._functions.keys() or None in self._functions.keys(): - await self.call_function( - context=self._context, - tool_call_id=tool_id, - function_name=function_name, - arguments=arguments, - run_llm=run_llm, - ) - else: - raise OpenAIUnhandledFunctionException( - f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function." + function_calls = [] + for item in items: + args = json.loads(item.arguments) + function_calls.append( + FunctionCallLLM( + context=self._context, + tool_call_id=item.call_id, + function_name=item.name, + arguments=args, ) + ) + await self.run_function_calls(function_calls) # # state and client events for the current conversation From 4809684a136afbd745501890a8cc3bf160033754 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 28 Apr 2025 14:12:26 -0700 Subject: [PATCH 04/12] LLMService: cancel function calls on interruptions by default --- CHANGELOG.md | 6 ++++++ src/pipecat/services/llm_service.py | 2 +- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dc8a43bdd..8d5bb1bb1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -60,6 +60,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Function calls are now cancelled by default if there's an interruption. To + disable this behavior you can set `cancel_on_interruption=False` when + registering the function call. Since function calls are executed as tasks you + can tell if a function call has been cancelled by catching the + `asyncio.CancelledError` exception (and don't forget to raise it again!). + - Updated OpenTelemetry tracing attribute `metrics.ttfb_ms` to `metrics.ttfb`. The attribute reports TTFB in seconds. diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 69eabc686..2a94598fa 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -181,7 +181,7 @@ class LLMService(AIService): handler: Any, start_callback=None, *, - cancel_on_interruption: bool = False, + cancel_on_interruption: bool = True, ): # Registering a function with the function_name set to None will run # that handler for all functions From 04bf85ddfef7e7c840155987dc7cef855c9fb6e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 20 May 2025 22:23:26 -0700 Subject: [PATCH 05/12] LLMService: allow executing tasks sequentially and in parallel --- CHANGELOG.md | 7 +- .../processors/aggregators/llm_response.py | 2 - src/pipecat/services/llm_service.py | 115 +++++++++++------- 3 files changed, 75 insertions(+), 49 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d5bb1bb1..715f8c7c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,10 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Function calls can now be executed sequentially (in the order received in the completion) by passing `run_in_parallel=False` when creating your LLM - service. By default, function calls run in parallel, so if the LLM completion - returns 2 or more function calls they run concurrently. In both cases, - concurrently and sequentailly, a new LLM completion will run when the last - function call finishes. + service. By default, if the LLM completion returns 2 or more function calls + they run concurrently. In both cases, concurrently and sequentially, a new LLM + completion will run when the last function call finishes. - Added OpenTelemetry tracing for `GeminiMultimodalLiveLLMService` and `OpenAIRealtimeBetaLLMService`. diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 4574d624d..94dad4d1a 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -591,8 +591,6 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): ) return - in_progress = self._function_calls_in_progress[frame.tool_call_id] - del self._function_calls_in_progress[frame.tool_call_id] properties = frame.properties diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 2a94598fa..3f2eefc30 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -97,7 +97,7 @@ class FunctionCallRunner: tool_call_id: str arguments: Mapping[str, Any] context: OpenAILLMContext - run_llm: bool + run_llm: Optional[bool] = None @dataclass @@ -129,12 +129,14 @@ class LLMService(AIService): # However, subclasses should override this with a more specific adapter when necessary. adapter_class: Type[BaseLLMAdapter] = OpenAILLMAdapter - def __init__(self, **kwargs): + def __init__(self, run_in_parallel: bool = True, **kwargs): super().__init__(**kwargs) + self._run_in_parallel = run_in_parallel self._start_callbacks = {} self._adapter = self.adapter_class() self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} - self._function_call_runner_task: Optional[asyncio.Task] = None + self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunner] = {} + self._sequential_runner_task: Optional[asyncio.Task] = None self._register_event_handler("on_completion_timeout") @@ -152,17 +154,18 @@ class LLMService(AIService): async def start(self, frame: StartFrame): await super().start(frame) - await self._create_runner_task() + if not self._run_in_parallel: + await self._create_sequential_runner_task() async def stop(self, frame: EndFrame): await super().stop(frame) - await self._cancel_function_call() - await self._cancel_runner_task() + if not self._run_in_parallel: + await self._cancel_sequential_runner_task() async def cancel(self, frame: CancelFrame): await super().cancel(frame) - await self._cancel_function_call() - await self._cancel_runner_task() + if not self._run_in_parallel: + await self._cancel_sequential_runner_task() async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -171,9 +174,9 @@ class LLMService(AIService): await self._handle_interruptions(frame) async def _handle_interruptions(self, _: StartInterruptionFrame): - await self._cancel_function_call() - await self._cancel_runner_task() - await self._create_runner_task() + for function_name, entry in self._functions.items(): + if entry.cancel_on_interruption: + await self._cancel_function_call(function_name) def register_function( self, @@ -227,8 +230,12 @@ class LLMService(AIService): ) continue - # Run inference on the last function call. - run_llm = index == total_function_calls - 1 + # If we are not running in parallel, run inference on the last + # function call. Otherwise, the last function call to finish is the + # one that will run the inference. + run_llm = None + if not self._run_in_parallel: + run_llm = index == total_function_calls - 1 runner_item = FunctionCallRunner( registry_item=item, @@ -239,7 +246,12 @@ class LLMService(AIService): run_llm=run_llm, ) - await self._function_call_runner_queue.put(runner_item) + if self._run_in_parallel: + task = self.create_task(self._run_function_call(runner_item)) + self._function_call_tasks[task] = runner_item + task.add_done_callback(self._function_call_task_finished) + else: + await self._sequential_runner_queue.put(runner_item) async def call_start_function(self, context: OpenAILLMContext, function_name: str): if function_name in self._start_callbacks.keys(): @@ -267,25 +279,25 @@ class LLMService(AIService): FrameDirection.UPSTREAM, ) - async def _create_runner_task(self): - if not self._function_call_runner_task: - self._current_runner: Optional[FunctionCallRunner] = None - self._current_task: Optional[asyncio.Task] = None - self._function_call_runner_queue = asyncio.Queue() - self._function_call_runner_task = self.create_task(self._function_call_runner_handler()) + async def _create_sequential_runner_task(self): + if not self._sequential_runner_task: + self._sequential_runner_queue = asyncio.Queue() + self._sequential_runner_task = self.create_task(self._sequential_runner_handler()) - async def _cancel_runner_task(self): - if self._function_call_runner_task: - await self.cancel_task(self._function_call_runner_task) - self._function_call_runner_task = None + async def _cancel_sequential_runner_task(self): + if self._sequential_runner_task: + await self.cancel_task(self._sequential_runner_task) + self._sequential_runner_task = None - async def _function_call_runner_handler(self): + async def _sequential_runner_handler(self): while True: - self._current_runner = await self._function_call_runner_queue.get() - self._current_task = self.create_task(self._run_function_call(self._current_runner)) - await self.wait_for_task(self._current_task) - self._current_runner = None - self._current_task = None + runner_item = await self._sequential_runner_queue.get() + task = self.create_task(self._run_function_call(runner_item)) + self._function_call_tasks[task] = runner_item + # Since we run tasks sequentially we don't need to call + # task.add_done_callback(self._function_call_task_finished). + await self.wait_for_task(task) + del self._function_call_tasks[task] async def _run_function_call(self, runner_item: FunctionCallRunner): if runner_item.function_name in self._functions.keys(): @@ -378,20 +390,37 @@ class LLMService(AIService): ) await item.handler(params) - async def _cancel_function_call(self): - if ( - self._current_runner - and self._current_task - and self._current_runner.registry_item.cancel_on_interruption - ): - name = self._current_runner.function_name - tool_call_id = self._current_runner.tool_call_id + async def _cancel_function_call(self, function_name: Optional[str]): + cancelled_tasks = set() + for task, runner_item in self._function_call_tasks.items(): + if runner_item.registry_item.function_name == function_name: + name = runner_item.function_name + tool_call_id = runner_item.tool_call_id - logger.debug(f"{self} Cancelling function call [{name}:{tool_call_id}]...") + # We remove the callback because we are going to cancel the task + # now, otherwise we will be removing it from the set while we + # are iterating. + task.remove_done_callback(self._function_call_task_finished) - await self.cancel_task(self._current_task) + logger.debug(f"{self} Cancelling function call [{name}:{tool_call_id}]...") - frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id) - await self.push_frame(frame) + await self.cancel_task(task) - logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled") + frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id) + await self.push_frame(frame) + + cancelled_tasks.add(task) + + logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled") + + # Remove all cancelled tasks from our set. + for task in cancelled_tasks: + self._function_call_task_finished(task) + + def _function_call_task_finished(self, task: asyncio.Task): + if task in self._function_call_tasks: + del self._function_call_tasks[task] + # The task is finished so this should exit immediately. We need to + # do this because otherwise the task manager would report a dangling + # task if we don't remove it. + asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop()) From 40b52caddec9640a7967f2f744907a8e6a05ac2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 21 May 2025 11:38:37 -0700 Subject: [PATCH 06/12] LLMService: s/FunctionCallLLM/FunctionCallFromLLM/ s/FunctionCallRunner/FunctionCallRunnerItem/ --- src/pipecat/services/anthropic/llm.py | 4 +- .../services/gemini_multimodal_live/gemini.py | 4 +- src/pipecat/services/google/llm.py | 4 +- src/pipecat/services/google/llm_openai.py | 4 +- src/pipecat/services/llm_service.py | 121 +++++++++--------- src/pipecat/services/openai/base_llm.py | 4 +- .../services/openai_realtime_beta/openai.py | 4 +- 7 files changed, 73 insertions(+), 72 deletions(-) diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 49b4f2f6a..236f269fa 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -45,7 +45,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import FunctionCallLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.utils.tracing.service_decorators import traced_llm try: @@ -227,7 +227,7 @@ class AnthropicLLMService(LLMService): if tool_use_block: args = json.loads(json_accumulator) if json_accumulator else {} function_calls.append( - FunctionCallLLM( + FunctionCallFromLLM( context=context, tool_call_id=tool_use_block.id, function_name=tool_use_block.name, diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 3096184d2..894e53285 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -52,7 +52,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import FunctionCallLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, OpenAIUserContextAggregator, @@ -893,7 +893,7 @@ class GeminiMultimodalLiveLLMService(LLMService): logger.error("Function calls are not supported without a context object.") function_calls_llm = [ - FunctionCallLLM( + FunctionCallFromLLM( context=self._context, tool_call_id=f.id, function_name=f.name, diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 68e6f2406..b63b3786c 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -42,7 +42,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.google.frames import LLMSearchResponseFrame -from pipecat.services.llm_service import FunctionCallLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, OpenAIUserContextAggregator, @@ -578,7 +578,7 @@ class GoogleLLMService(LLMService): id = function_call.id or str(uuid.uuid4()) logger.debug(f"Function call: {function_call.name}:{id}") function_calls.append( - FunctionCallLLM( + FunctionCallFromLLM( context=context, tool_call_id=id, function_name=function_call.name, diff --git a/src/pipecat/services/google/llm_openai.py b/src/pipecat/services/google/llm_openai.py index db395705f..a497cb229 100644 --- a/src/pipecat/services/google/llm_openai.py +++ b/src/pipecat/services/google/llm_openai.py @@ -10,7 +10,7 @@ import os from openai import AsyncStream from openai.types.chat import ChatCompletionChunk -from pipecat.services.llm_service import FunctionCallLLM +from pipecat.services.llm_service import FunctionCallFromLLM # Suppress gRPC fork warnings os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" @@ -127,7 +127,7 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): arguments = json.loads(arguments) function_calls.append( - FunctionCallLLM( + FunctionCallFromLLM( context=context, tool_call_id=tool_id, function_name=function_name, diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 3f2eefc30..1ea55fb13 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -44,62 +44,6 @@ class FunctionCallResultCallback(Protocol): ) -> None: ... -@dataclass -class FunctionCallRegistryItem: - """Represents an entry of our function call registry. - - Attributes: - function_name (Optional[str]): The name of the function. - handler (FunctionCallHandler): The handler for processing function call parameters. - cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption. - - """ - - function_name: Optional[str] - handler: FunctionCallHandler - cancel_on_interruption: bool - - -@dataclass -class FunctionCallLLM: - """Represents a function call returned by the LLM to be registered for execution. - - Attributes: - function_name (str): The name of the function. - tool_call_id (str): A unique identifier for the function call. - arguments (Mapping[str, Any]): The arguments for the function. - context (OpenAILLMContext): The LLM context. - - """ - - function_name: str - tool_call_id: str - arguments: Mapping[str, Any] - context: OpenAILLMContext - - -@dataclass -class FunctionCallRunner: - """Represents an internal function call entry to our function call - runner. The runner executes function calls in order. - - Attributes: - registry_name (Optional[str]): The function call name registration (could be None). - function_name (str): The name of the function. - tool_call_id (str): A unique identifier for the function call. - arguments (Mapping[str, Any]): The arguments for the function. - context (OpenAILLMContext): The LLM context. - - """ - - registry_item: FunctionCallRegistryItem - function_name: str - tool_call_id: str - arguments: Mapping[str, Any] - context: OpenAILLMContext - run_llm: Optional[bool] = None - - @dataclass class FunctionCallParams: """Parameters for a function call. @@ -122,6 +66,63 @@ class FunctionCallParams: result_callback: FunctionCallResultCallback +@dataclass +class FunctionCallFromLLM: + """Represents a function call returned by the LLM to be registered for execution. + + Attributes: + function_name (str): The name of the function. + tool_call_id (str): A unique identifier for the function call. + arguments (Mapping[str, Any]): The arguments for the function. + context (OpenAILLMContext): The LLM context. + + """ + + function_name: str + tool_call_id: str + arguments: Mapping[str, Any] + context: OpenAILLMContext + + +@dataclass +class FunctionCallRegistryItem: + """Represents an entry in our function call registry. This is what the user + registers. + + Attributes: + function_name (Optional[str]): The name of the function. + handler (FunctionCallHandler): The handler for processing function call parameters. + cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption. + + """ + + function_name: Optional[str] + handler: FunctionCallHandler + cancel_on_interruption: bool + + +@dataclass +class FunctionCallRunnerItem: + """Represents an internal function call entry to our function call + runner. The runner executes function calls in order. + + Attributes: + registry_name (Optional[str]): The function call name registration (could be None). + function_name (str): The name of the function. + tool_call_id (str): A unique identifier for the function call. + arguments (Mapping[str, Any]): The arguments for the function. + context (OpenAILLMContext): The LLM context. + + """ + + registry_item: FunctionCallRegistryItem + function_name: str + tool_call_id: str + arguments: Mapping[str, Any] + context: OpenAILLMContext + run_llm: Optional[bool] = None + + class LLMService(AIService): """This class is a no-op but serves as a base class for LLM services.""" @@ -135,7 +136,7 @@ class LLMService(AIService): self._start_callbacks = {} self._adapter = self.adapter_class() self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} - self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunner] = {} + self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunnerItem] = {} self._sequential_runner_task: Optional[asyncio.Task] = None self._register_event_handler("on_completion_timeout") @@ -217,7 +218,7 @@ class LLMService(AIService): return True return function_name in self._functions.keys() - async def run_function_calls(self, function_calls: Sequence[FunctionCallLLM]): + async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]): total_function_calls = len(function_calls) for index, function_call in enumerate(function_calls): if function_call.function_name in self._functions.keys(): @@ -237,7 +238,7 @@ class LLMService(AIService): if not self._run_in_parallel: run_llm = index == total_function_calls - 1 - runner_item = FunctionCallRunner( + runner_item = FunctionCallRunnerItem( registry_item=item, function_name=function_call.function_name, tool_call_id=function_call.tool_call_id, @@ -299,7 +300,7 @@ class LLMService(AIService): await self.wait_for_task(task) del self._function_call_tasks[task] - async def _run_function_call(self, runner_item: FunctionCallRunner): + async def _run_function_call(self, runner_item: FunctionCallRunnerItem): if runner_item.function_name in self._functions.keys(): item = self._functions[runner_item.function_name] elif None in self._functions.keys(): diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 7f3303a2e..2badfed96 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -34,7 +34,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import FunctionCallLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.utils.tracing.service_decorators import traced_llm @@ -263,7 +263,7 @@ class BaseOpenAILLMService(LLMService): ): arguments = json.loads(arguments) function_calls.append( - FunctionCallLLM( + FunctionCallFromLLM( context=context, tool_call_id=tool_id, function_name=function_name, diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 343c61415..4ea5843bf 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -48,7 +48,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import FunctionCallLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.openai.llm import OpenAIContextAggregatorPair from pipecat.transcriptions.language import Language from pipecat.utils.time import time_now_iso8601 @@ -587,7 +587,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): for item in items: args = json.loads(item.arguments) function_calls.append( - FunctionCallLLM( + FunctionCallFromLLM( context=self._context, tool_call_id=item.call_id, function_name=item.name, From f0cbdc4e6831a890914a4d0f3951e524582ae730 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 09:18:01 -0700 Subject: [PATCH 07/12] LLMService: add `on_function_calls_started` event --- CHANGELOG.md | 4 ++++ src/pipecat/services/llm_service.py | 21 ++++++++++++++++++++- 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 715f8c7c7..3304454da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added LLM services `on_function_calls_started` event. This event will be + triggered when the LLM service receives function calls from the model and is + going to start executing them. + - Function calls can now be executed sequentially (in the order received in the completion) by passing `run_in_parallel=False` when creating your LLM service. By default, if the LLM completion returns 2 or more function calls diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 1ea55fb13..6b419ad70 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -124,7 +124,23 @@ class FunctionCallRunnerItem: class LLMService(AIService): - """This class is a no-op but serves as a base class for LLM services.""" + """This is the base class for all LLM services. It handles function calling + registration and execution. The class also provides event handlers. + + An event to know when an LLM service completion timeout occurs: + + @task.event_handler("on_completion_timeout") + async def on_completion_timeout(service): + ... + + And an event to know that function calls have been received from the LLM + service and that we are going to start executing them: + + @task.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls: Sequence[FunctionCallFromLLM]): + ... + + """ # OpenAILLMAdapter is used as the default adapter since it aligns with most LLM implementations. # However, subclasses should override this with a more specific adapter when necessary. @@ -139,6 +155,7 @@ class LLMService(AIService): self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunnerItem] = {} self._sequential_runner_task: Optional[asyncio.Task] = None + self._register_event_handler("on_function_calls_started") self._register_event_handler("on_completion_timeout") def get_llm_adapter(self) -> BaseLLMAdapter: @@ -219,6 +236,8 @@ class LLMService(AIService): return function_name in self._functions.keys() async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]): + await self._call_event_handler("on_function_calls_started", function_calls) + total_function_calls = len(function_calls) for index, function_call in enumerate(function_calls): if function_call.function_name in self._functions.keys(): From 297afdd1266185af1a5cf69b6cf11153cd74d090 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 09:55:40 -0700 Subject: [PATCH 08/12] LLMService: add new FunctionCallsStartedFrame --- CHANGELOG.md | 4 ++ src/pipecat/frames/frames.py | 26 +++++++++++++ .../processors/aggregators/llm_response.py | 18 +++++++-- src/pipecat/services/llm_service.py | 37 +++++-------------- 4 files changed, 53 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3304454da..61f3b6e76 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added a new frame `FunctionCallsStartedFrame`. This frame is pushed both + upstream and downstream from the LLM service to indicate that one or more + function calls are going to be executed. + - Added LLM services `on_function_calls_started` event. This event will be triggered when the LLM service receives function calls from the model and is going to start executing them. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index bb6143c61..6ad0f089f 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -667,6 +667,32 @@ class MetricsFrame(SystemFrame): data: List[MetricsData] +@dataclass +class FunctionCallFromLLM: + """Represents a function call returned by the LLM to be registered for execution. + + Attributes: + function_name (str): The name of the function. + tool_call_id (str): A unique identifier for the function call. + arguments (Mapping[str, Any]): The arguments for the function. + context (OpenAILLMContext): The LLM context. + + """ + + function_name: str + tool_call_id: str + arguments: Mapping[str, Any] + context: Any + + +@dataclass +class FunctionCallsStartedFrame(SystemFrame): + """A frame signaling that one or more function call execution is going to + start.""" + + function_calls: Sequence[FunctionCallFromLLM] + + @dataclass class FunctionCallInProgressFrame(SystemFrame): """A frame signaling that a function call is in progress.""" diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 94dad4d1a..ae2382cd6 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -23,6 +23,7 @@ from pipecat.frames.frames import ( FunctionCallCancelFrame, FunctionCallInProgressFrame, FunctionCallResultFrame, + FunctionCallsStartedFrame, InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -500,7 +501,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): self._params.expect_stripped_words = kwargs["expect_stripped_words"] self._started = 0 - self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {} + self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {} self._context_updated_tasks: Set[asyncio.Task] = set() async def handle_aggregation(self, aggregation: str): @@ -538,6 +539,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): self.set_tools(frame.tools) elif isinstance(frame, LLMSetToolChoiceFrame): self.set_tool_choice(frame.tool_choice) + elif isinstance(frame, FunctionCallsStartedFrame): + await self._handle_function_calls_started(frame) elif isinstance(frame, FunctionCallInProgressFrame): await self._handle_function_call_in_progress(frame) elif isinstance(frame, FunctionCallResultFrame): @@ -574,6 +577,12 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): self._started = 0 self.reset() + async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame): + function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls] + logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}") + for function_call in frame.function_calls: + self._function_calls_in_progress[function_call.tool_call_id] = None + async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): logger.debug( f"{self} FunctionCallInProgressFrame: [{frame.function_name}:{frame.tool_call_id}]" @@ -597,9 +606,10 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): await self.handle_function_call_result(frame) + run_llm = False + # Run inference if the function call result requires it. if frame.result: - run_llm = False if properties and properties.run_llm is not None: # If the tool call result has a run_llm property, use it. run_llm = properties.run_llm @@ -610,8 +620,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): # If this is the last function call in progress, run the LLM. run_llm = not bool(self._function_calls_in_progress) - if run_llm: - await self.push_context_frame(FrameDirection.UPSTREAM) + if run_llm: + await self.push_context_frame(FrameDirection.UPSTREAM) # Call the `on_context_updated` callback once the function call result # is added to the context. Also, run this in a separate task to make diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 6b419ad70..7c95e4e57 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -18,9 +18,11 @@ from pipecat.frames.frames import ( EndFrame, Frame, FunctionCallCancelFrame, + FunctionCallFromLLM, FunctionCallInProgressFrame, FunctionCallResultFrame, FunctionCallResultProperties, + FunctionCallsStartedFrame, StartFrame, StartInterruptionFrame, UserImageRequestFrame, @@ -66,24 +68,6 @@ class FunctionCallParams: result_callback: FunctionCallResultCallback -@dataclass -class FunctionCallFromLLM: - """Represents a function call returned by the LLM to be registered for execution. - - Attributes: - function_name (str): The name of the function. - tool_call_id (str): A unique identifier for the function call. - arguments (Mapping[str, Any]): The arguments for the function. - context (OpenAILLMContext): The LLM context. - - """ - - function_name: str - tool_call_id: str - arguments: Mapping[str, Any] - context: OpenAILLMContext - - @dataclass class FunctionCallRegistryItem: """Represents an entry in our function call registry. This is what the user @@ -238,8 +222,13 @@ class LLMService(AIService): async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]): await self._call_event_handler("on_function_calls_started", function_calls) - total_function_calls = len(function_calls) - for index, function_call in enumerate(function_calls): + # Push frame both downstream and upstream + started_frame_downstream = FunctionCallsStartedFrame(function_calls=function_calls) + started_frame_upstream = FunctionCallsStartedFrame(function_calls=function_calls) + await self.push_frame(started_frame_downstream, FrameDirection.DOWNSTREAM) + await self.push_frame(started_frame_upstream, FrameDirection.UPSTREAM) + + for function_call in function_calls: if function_call.function_name in self._functions.keys(): item = self._functions[function_call.function_name] elif None in self._functions.keys(): @@ -250,20 +239,12 @@ class LLMService(AIService): ) continue - # If we are not running in parallel, run inference on the last - # function call. Otherwise, the last function call to finish is the - # one that will run the inference. - run_llm = None - if not self._run_in_parallel: - run_llm = index == total_function_calls - 1 - runner_item = FunctionCallRunnerItem( registry_item=item, function_name=function_call.function_name, tool_call_id=function_call.tool_call_id, arguments=function_call.arguments, context=function_call.context, - run_llm=run_llm, ) if self._run_in_parallel: From d86343c38d2f018221f860599ba3b9ce45452628 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 10:10:57 -0700 Subject: [PATCH 09/12] examples: update to use on_function_calls_started event --- examples/foundational/14-function-calling.py | 5 ++++- examples/foundational/14c-function-calling-together.py | 5 ++++- examples/foundational/14e-function-calling-gemini.py | 5 ++++- examples/foundational/14f-function-calling-groq.py | 5 ++++- examples/foundational/14h-function-calling-azure.py | 5 ++++- examples/foundational/14i-function-calling-fireworks.py | 5 ++++- examples/foundational/14j-function-calling-nim.py | 5 ++++- examples/foundational/14k-function-calling-cerebras.py | 5 ++++- examples/foundational/14l-function-calling-deepseek.py | 5 ++++- examples/foundational/14m-function-calling-openrouter.py | 5 ++++- .../14o-function-calling-gemini-openai-format.py | 5 ++++- .../foundational/14p-function-calling-gemini-vertex-ai.py | 5 ++++- examples/foundational/14q-function-calling-qwen.py | 5 ++++- examples/foundational/22b-natural-conversation-proposal.py | 5 ++++- examples/foundational/22c-natural-conversation-mixed-llms.py | 5 ++++- examples/open-telemetry/jaeger/bot.py | 5 ++++- examples/open-telemetry/langfuse/bot.py | 5 ++++- 17 files changed, 68 insertions(+), 17 deletions(-) diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index 7c80416b1..85f836228 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -72,6 +71,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14c-function-calling-together.py b/examples/foundational/14c-function-calling-together.py index 54545b0b2..1188f32e2 100644 --- a/examples/foundational/14c-function-calling-together.py +++ b/examples/foundational/14c-function-calling-together.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -74,6 +73,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14e-function-calling-gemini.py b/examples/foundational/14e-function-calling-gemini.py index 4e6a582f3..181cd576b 100644 --- a/examples/foundational/14e-function-calling-gemini.py +++ b/examples/foundational/14e-function-calling-gemini.py @@ -35,7 +35,6 @@ client_id = "" async def get_weather(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) location = params.arguments["location"] await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") @@ -94,6 +93,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si llm.register_function("get_weather", get_weather) llm.register_function("get_image", get_image) + @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", diff --git a/examples/foundational/14f-function-calling-groq.py b/examples/foundational/14f-function-calling-groq.py index 698845028..3ff56a915 100644 --- a/examples/foundational/14f-function-calling-groq.py +++ b/examples/foundational/14f-function-calling-groq.py @@ -31,7 +31,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -74,6 +73,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14h-function-calling-azure.py b/examples/foundational/14h-function-calling-azure.py index a54fb1429..25cf7defa 100644 --- a/examples/foundational/14h-function-calling-azure.py +++ b/examples/foundational/14h-function-calling-azure.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -75,6 +74,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14i-function-calling-fireworks.py b/examples/foundational/14i-function-calling-fireworks.py index 5467e685a..028d9fa64 100644 --- a/examples/foundational/14i-function-calling-fireworks.py +++ b/examples/foundational/14i-function-calling-fireworks.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -74,6 +73,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14j-function-calling-nim.py b/examples/foundational/14j-function-calling-nim.py index f9b437f83..d254e0d4f 100644 --- a/examples/foundational/14j-function-calling-nim.py +++ b/examples/foundational/14j-function-calling-nim.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -72,6 +71,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14k-function-calling-cerebras.py b/examples/foundational/14k-function-calling-cerebras.py index 7aa5a2af4..70ffceffd 100644 --- a/examples/foundational/14k-function-calling-cerebras.py +++ b/examples/foundational/14k-function-calling-cerebras.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -71,6 +70,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14l-function-calling-deepseek.py b/examples/foundational/14l-function-calling-deepseek.py index 0daa814e6..7e992942d 100644 --- a/examples/foundational/14l-function-calling-deepseek.py +++ b/examples/foundational/14l-function-calling-deepseek.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -71,6 +70,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14m-function-calling-openrouter.py b/examples/foundational/14m-function-calling-openrouter.py index 13487b3b6..023f725f6 100644 --- a/examples/foundational/14m-function-calling-openrouter.py +++ b/examples/foundational/14m-function-calling-openrouter.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -75,6 +74,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14o-function-calling-gemini-openai-format.py b/examples/foundational/14o-function-calling-gemini-openai-format.py index 8e1e0fdf1..8f0036e07 100644 --- a/examples/foundational/14o-function-calling-gemini-openai-format.py +++ b/examples/foundational/14o-function-calling-gemini-openai-format.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -71,6 +70,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14p-function-calling-gemini-vertex-ai.py b/examples/foundational/14p-function-calling-gemini-vertex-ai.py index 42567787a..4348b663a 100644 --- a/examples/foundational/14p-function-calling-gemini-vertex-ai.py +++ b/examples/foundational/14p-function-calling-gemini-vertex-ai.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -76,6 +75,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/14q-function-calling-qwen.py b/examples/foundational/14q-function-calling-qwen.py index d3740d25f..6e07a97e1 100644 --- a/examples/foundational/14q-function-calling-qwen.py +++ b/examples/foundational/14q-function-calling-qwen.py @@ -30,7 +30,6 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -72,6 +71,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 58259b266..1b03f341e 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -198,7 +198,6 @@ class OutputGate(FrameProcessor): async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -245,6 +244,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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.")) + tools = [ ChatCompletionToolParam( type="function", diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index 7d6a1ce18..0bcbf75bd 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -402,7 +402,6 @@ class OutputGate(FrameProcessor): async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -449,6 +448,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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.")) + tools = [ ChatCompletionToolParam( type="function", diff --git a/examples/open-telemetry/jaeger/bot.py b/examples/open-telemetry/jaeger/bot.py index 74e587410..0f3084fd3 100644 --- a/examples/open-telemetry/jaeger/bot.py +++ b/examples/open-telemetry/jaeger/bot.py @@ -49,7 +49,6 @@ if IS_TRACING_ENABLED: async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -93,6 +92,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", diff --git a/examples/open-telemetry/langfuse/bot.py b/examples/open-telemetry/langfuse/bot.py index 22ff399a0..9c4f34695 100644 --- a/examples/open-telemetry/langfuse/bot.py +++ b/examples/open-telemetry/langfuse/bot.py @@ -46,7 +46,6 @@ if IS_TRACING_ENABLED: async def fetch_weather_from_api(params: FunctionCallParams): - await params.llm.push_frame(TTSSpeakFrame("Let me check on that.")) await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -90,6 +89,10 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) + @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", From 897a944478f880c9d4ff7fb38f92d43fc3c58224 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 10:15:21 -0700 Subject: [PATCH 10/12] examples(14,14a): add restaurant recommendation function call --- examples/foundational/14-function-calling.py | 18 +++++++++++++++- .../14a-function-calling-anthropic.py | 21 +++++++++++++++++-- 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index 85f836228..b4e09c99a 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -33,6 +33,10 @@ async def fetch_weather_from_api(params: FunctionCallParams): await params.result_callback({"conditions": "nice", "temperature": "75"}) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -70,6 +74,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # 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_restaurant_recommendation", fetch_restaurant_recommendation) @llm.event_handler("on_function_calls_started") async def on_function_calls_started(service, function_calls): @@ -91,7 +96,18 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["location", "format"], ) - tools = ToolsSchema(standard_tools=[weather_function]) + 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]) messages = [ { diff --git a/examples/foundational/14a-function-calling-anthropic.py b/examples/foundational/14a-function-calling-anthropic.py index f46a42db1..bf34e211c 100644 --- a/examples/foundational/14a-function-calling-anthropic.py +++ b/examples/foundational/14a-function-calling-anthropic.py @@ -33,6 +33,10 @@ async def get_weather(params: FunctionCallParams): 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"}) + + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -66,9 +70,11 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si ) llm = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-7-sonnet-latest" + api_key=os.getenv("ANTHROPIC_API_KEY"), + model="claude-3-7-sonnet-latest", ) llm.register_function("get_weather", get_weather) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) weather_function = FunctionSchema( name="get_weather", @@ -81,7 +87,18 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["location"], ) - tools = ToolsSchema(standard_tools=[weather_function]) + 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]) # todo: test with very short initial user message From ed84637b5588371f097bcfa89e2da48787dc00ae Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 31 May 2025 09:34:43 -0400 Subject: [PATCH 11/12] Add additional function call for testing to 14e, 14r, 19, 19a, 26b --- .../14e-function-calling-gemini.py | 20 +++++++++++++-- .../foundational/14r-function-calling-aws.py | 18 ++++++++++++- .../foundational/19-openai-realtime-beta.py | 25 +++++++++++++++++-- .../foundational/19a-azure-realtime-beta.py | 25 +++++++++++++++++-- ...gemini-multimodal-live-function-calling.py | 25 ++++++++++++++++--- 5 files changed, 103 insertions(+), 10 deletions(-) diff --git a/examples/foundational/14e-function-calling-gemini.py b/examples/foundational/14e-function-calling-gemini.py index 181cd576b..ccbe952fb 100644 --- a/examples/foundational/14e-function-calling-gemini.py +++ b/examples/foundational/14e-function-calling-gemini.py @@ -39,6 +39,10 @@ async def get_weather(params: FunctionCallParams): 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): question = params.arguments["question"] logger.debug(f"Requesting image with user_id={client_id}, question={question}") @@ -92,6 +96,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001") llm.register_function("get_weather", get_weather) 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): @@ -113,6 +118,17 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, 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="Get an image from the video stream.", @@ -124,14 +140,14 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["question"], ) - tools = ToolsSchema(standard_tools=[weather_function, get_image_function]) + tools = ToolsSchema(standard_tools=[weather_function, get_image_function, restaurant_function]) 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 two tools: get_weather and get_image. +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. diff --git a/examples/foundational/14r-function-calling-aws.py b/examples/foundational/14r-function-calling-aws.py index f7796157a..4443bec27 100644 --- a/examples/foundational/14r-function-calling-aws.py +++ b/examples/foundational/14r-function-calling-aws.py @@ -32,6 +32,10 @@ async def fetch_weather_from_api(params: FunctionCallParams): await params.result_callback({"conditions": "nice", "temperature": "75"}) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -74,6 +78,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # 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_restaurant_recommendation", fetch_restaurant_recommendation) weather_function = FunctionSchema( name="get_current_weather", @@ -91,7 +96,18 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["location", "format"], ) - tools = ToolsSchema(standard_tools=[weather_function]) + 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]) messages = [ { diff --git a/examples/foundational/19-openai-realtime-beta.py b/examples/foundational/19-openai-realtime-beta.py index f1dd7269d..81e8dc6c3 100644 --- a/examples/foundational/19-openai-realtime-beta.py +++ b/examples/foundational/19-openai-realtime-beta.py @@ -45,6 +45,10 @@ async def fetch_weather_from_api(params: FunctionCallParams): ) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + weather_function = FunctionSchema( name="get_current_weather", description="Get the current weather", @@ -62,8 +66,20 @@ weather_function = FunctionSchema( 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"], +) + # Create tools schema -tools = ToolsSchema(standard_tools=[weather_function]) +tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -100,7 +116,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # turn_detection=False, input_audio_noise_reduction=InputAudioNoiseReduction(type="near_field"), # tools=tools, - instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI. + instructions="""You are a helpful and friendly AI. Act like a human, but remember that you aren't a human and that you can't do human things in the real world. Your voice and personality should be warm and engaging, with a lively and @@ -113,6 +129,10 @@ even if you're asked about them. You are participating in a voice conversation. Keep your responses concise, short, and to the point unless specifically asked to elaborate on a topic. +You have access to the following tools: +- get_current_weather: Get the current weather for a given location. +- get_restaurant_recommendation: Get a restaurant recommendation for a given location. + Remember, your responses should be short. Just one or two sentences, usually.""", ) @@ -125,6 +145,7 @@ Remember, your responses should be short. Just one or two sentences, usually.""" # you can either register a single function for all function calls, or specific functions # llm.register_function(None, fetch_weather_from_api) llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) # Create a standard OpenAI LLM context object using the normal messages format. The # OpenAIRealtimeBetaLLMService will convert this internally to messages that the diff --git a/examples/foundational/19a-azure-realtime-beta.py b/examples/foundational/19a-azure-realtime-beta.py index 76180e397..54f0302e5 100644 --- a/examples/foundational/19a-azure-realtime-beta.py +++ b/examples/foundational/19a-azure-realtime-beta.py @@ -43,6 +43,10 @@ async def fetch_weather_from_api(params: FunctionCallParams): ) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + # Define weather function using standardized schema weather_function = FunctionSchema( name="get_current_weather", @@ -61,8 +65,20 @@ weather_function = FunctionSchema( 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"], +) + # Create tools schema -tools = ToolsSchema(standard_tools=[weather_function]) +tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -98,7 +114,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # Or set to False to disable openai turn detection and use transport VAD # turn_detection=False, # tools=tools, - instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI. + instructions="""You are a helpful and friendly AI. Act like a human, but remember that you aren't a human and that you can't do human things in the real world. Your voice and personality should be warm and engaging, with a lively and @@ -111,6 +127,10 @@ even if you're asked about them. You are participating in a voice conversation. Keep your responses concise, short, and to the point unless specifically asked to elaborate on a topic. +You have access to the following tools: +- get_current_weather: Get the current weather for a given location. +- get_restaurant_recommendation: Get a restaurant recommendation for a given location. + Remember, your responses should be short. Just one or two sentences, usually.""", ) @@ -124,6 +144,7 @@ Remember, your responses should be short. Just one or two sentences, usually.""" # you can either register a single function for all function calls, or specific functions # llm.register_function(None, fetch_weather_from_api) llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) # Create a standard OpenAI LLM context object using the normal messages format. The # OpenAIRealtimeBetaLLMService will convert this internally to messages that the diff --git a/examples/foundational/26b-gemini-multimodal-live-function-calling.py b/examples/foundational/26b-gemini-multimodal-live-function-calling.py index 84c7d5045..15db2faa0 100644 --- a/examples/foundational/26b-gemini-multimodal-live-function-calling.py +++ b/examples/foundational/26b-gemini-multimodal-live-function-calling.py @@ -40,11 +40,17 @@ async def fetch_weather_from_api(params: FunctionCallParams): ) +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + system_instruction = """ You are a helpful assistant who can answer questions and use tools. -You have a tool called "get_current_weather" that can be used to get the current weather. If the user asks -for the weather, call this function. +You have three tools available to you: +1. get_current_weather: Use this tool to get the current weather in a specific location. +2. get_restaurant_recommendation: Use this tool to get a restaurant recommendation in a specific location. +3. google_search: Use this tool to search the web for information. """ @@ -101,9 +107,21 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, 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"], + ) search_tool = {"google_search": {}} tools = ToolsSchema( - standard_tools=[weather_function], custom_tools={AdapterType.GEMINI: [search_tool]} + standard_tools=[weather_function, restaurant_function], + custom_tools={AdapterType.GEMINI: [search_tool]}, ) llm = GeminiMultimodalLiveLLMService( @@ -113,6 +131,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si ) llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) context = OpenAILLMContext( [{"role": "user", "content": "Say hello."}], From ef3143d5581bf000ef2cd0f1622c7b5f182cdd59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sat, 31 May 2025 14:01:56 -0700 Subject: [PATCH 12/12] LLMService: don't run function calls if none are given --- src/pipecat/services/llm_service.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 7c95e4e57..5619cd35e 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -220,6 +220,9 @@ class LLMService(AIService): return function_name in self._functions.keys() async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]): + if len(function_calls) == 0: + return + await self._call_event_handler("on_function_calls_started", function_calls) # Push frame both downstream and upstream