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: