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,