diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index b5aba449c..9141029ca 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -34,7 +34,12 @@ logger.add(sys.stderr, level="DEBUG") async def start_fetch_weather(function_name, llm, context): - await llm.push_frame(TextFrame("Let me check on that.")) + # note: we can't push a frame to the LLM here. the bot + # can interrupt itself and/or cause audio overlapping glitches. + # possible question for Aleix and Chad about what the right way + # to trigger speech is, now, with the new queues/async/sync refactors. + # await llm.push_frame(TextFrame("Let me check on that.")) + logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): @@ -106,11 +111,11 @@ async def main(): pipeline = Pipeline( [ - fl_in, + # fl_in, transport.input(), context_aggregator.user(), llm, - fl_out, + # fl_out, tts, transport.output(), context_aggregator.assistant(), diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 8059b904b..f7faa8ef0 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -585,6 +585,7 @@ class FunctionCallResultFrame(DataFrame): tool_call_id: str arguments: str result: Any + run_llm: bool = True @dataclass diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 83ec3e57f..4bf3f042c 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -133,6 +133,7 @@ class OpenAILLMContext: tool_call_id: str, arguments: str, llm: FrameProcessor, + run_llm: bool = True, ) -> None: # 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 @@ -153,6 +154,7 @@ class OpenAILLMContext: tool_call_id=tool_call_id, arguments=arguments, result=result, + run_llm=run_llm, ) ) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 5eadb475b..a46ad3fab 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -110,7 +110,13 @@ class LLMService(AIService): return function_name in self._callbacks.keys() async def call_function( - self, *, context: OpenAILLMContext, tool_call_id: str, function_name: str, arguments: str + self, + *, + context: OpenAILLMContext, + tool_call_id: str, + function_name: str, + arguments: str, + run_llm: bool, ) -> None: f = None if function_name in self._callbacks.keys(): @@ -120,7 +126,12 @@ class LLMService(AIService): else: return None await context.call_function( - f, function_name=function_name, tool_call_id=tool_call_id, arguments=arguments, llm=self + f, + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + llm=self, + run_llm=run_llm, ) # QUESTION FOR CB: maybe this isn't needed anymore? diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index c17916f2d..49fd04371 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -205,6 +205,10 @@ class BaseOpenAILLMService(LLMService): return chunks async def _process_context(self, context: OpenAILLMContext): + functions_list = [] + arguments_list = [] + tool_id_list = [] + func_idx = 0 function_name = "" arguments = "" tool_call_id = "" @@ -242,6 +246,14 @@ class BaseOpenAILLMService(LLMService): # yield a frame containing the function name and the arguments. tool_call = chunk.choices[0].delta.tool_calls[0] + if tool_call.index != func_idx: + functions_list.append(function_name) + arguments_list.append(arguments) + tool_id_list.append(tool_call_id) + function_name = "" + arguments = "" + tool_call_id = "" + func_idx += 1 if tool_call.function and tool_call.function.name: function_name += tool_call.function.name tool_call_id = tool_call.id @@ -257,21 +269,29 @@ class BaseOpenAILLMService(LLMService): # the context, and re-prompt to get a chat answer. If we don't have a registered # handler, raise an exception. if function_name and arguments: - if self.has_function(function_name): - await self._handle_function_call(context, tool_call_id, function_name, arguments) - else: - raise OpenAIUnhandledFunctionException( - f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function." - ) + # added to the list as last function name and arguments not added to the list + functions_list.append(function_name) + arguments_list.append(arguments) + tool_id_list.append(tool_call_id) - async def _handle_function_call(self, context, tool_call_id, function_name, arguments): - arguments = json.loads(arguments) - await self.call_function( - context=context, - tool_call_id=tool_call_id, - function_name=function_name, - arguments=arguments, - ) + total_items = len(functions_list) + for index, (function_name, arguments, tool_id) in enumerate( + zip(functions_list, arguments_list, tool_id_list), start=1 + ): + if self.has_function(function_name): + run_llm = index == total_items + arguments = json.loads(arguments) + await self.call_function( + context=context, + 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." + ) async def _update_settings(self, frame: LLMUpdateSettingsFrame): if frame.model is not None: @@ -465,31 +485,27 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): def __init__(self, user_context_aggregator: OpenAIUserContextAggregator, **kwargs): super().__init__(context=user_context_aggregator._context, **kwargs) self._user_context_aggregator = user_context_aggregator - self._function_call_in_progress = None + self._function_calls_in_progress = {} self._function_call_result = None async def process_frame(self, frame, direction): await super().process_frame(frame, direction) # See note above about not calling push_frame() here. if isinstance(frame, StartInterruptionFrame): - self._function_call_in_progress = None + self._function_calls_in_progress.clear() self._function_call_finished = None elif isinstance(frame, FunctionCallInProgressFrame): - self._function_call_in_progress = frame + self._function_calls_in_progress[frame.tool_call_id] = frame elif isinstance(frame, FunctionCallResultFrame): - if ( - self._function_call_in_progress - and self._function_call_in_progress.tool_call_id == frame.tool_call_id - ): - self._function_call_in_progress = None + if frame.tool_call_id in self._function_calls_in_progress: + del self._function_calls_in_progress[frame.tool_call_id] self._function_call_result = frame # TODO-CB: Kwin wants us to refactor this out of here but I REFUSE await self._push_aggregation() else: logger.warning( - "FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id" + "FunctionCallResultFrame tool_call_id does not match any function call in progress" ) - self._function_call_in_progress = None self._function_call_result = None async def _push_aggregation(self): @@ -528,7 +544,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): "tool_call_id": frame.tool_call_id, } ) - run_llm = True + run_llm = frame.run_llm else: self._context.add_message({"role": "assistant", "content": aggregation})